iota_node/

lib.rs

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

#[cfg(msim)]
use std::sync::atomic::Ordering;
use std::{
    collections::HashMap,
    fmt,
    future::Future,
    num::NonZeroUsize,
    sync::{Arc, Weak},
    time::Duration,
};

use anemo::Network;
use anemo_tower::{
    callback::CallbackLayer,
    trace::{DefaultMakeSpan, DefaultOnFailure, TraceLayer},
};
use anyhow::{Result, anyhow};
use arc_swap::ArcSwap;
use futures::future::BoxFuture;
pub use handle::IotaNodeHandle;
use iota_common::{debug_fatal, fatal};
use iota_config::{
    ConsensusConfig, NodeConfig, node::RunWithRange, node_config_metrics::NodeConfigMetrics,
};
use iota_core::{
    authority::{
        AuthorityState, AuthorityStore, RandomnessRoundReceiver,
        authority_per_epoch_store::AuthorityPerEpochStore,
        authority_store_pruner::ObjectsCompactionFilter,
        authority_store_tables::{
            AuthorityPerpetualTables, AuthorityPerpetualTablesOptions, AuthorityPrunerTables,
        },
        backpressure::BackpressureManager,
        epoch_start_configuration::{EpochFlag, EpochStartConfigTrait, EpochStartConfiguration},
    },
    authority_aggregator::{
        AggregatorSendCapabilityNotificationError, AuthAggMetrics, AuthorityAggregator,
    },
    authority_client::NetworkAuthorityClient,
    authority_server::{
        ValidatorService, ValidatorServiceMetrics, soft_lock::PreConsensusSoftLocks,
    },
    checkpoint_progress_tracker::CheckpointProgressTracker,
    checkpoints::{
        CheckpointMetrics, CheckpointService, CheckpointStore, FullCheckpointContentsCache,
        FullCheckpointContentsCacheMetrics, SendCheckpointToStateSync, SubmitCheckpointToConsensus,
        checkpoint_executor::{CheckpointExecutor, StopReason, metrics::CheckpointExecutorMetrics},
    },
    connection_monitor::ConnectionMonitor,
    consensus_adapter::{
        CheckConnection, ConnectionMonitorStatus, ConsensusAdapter, ConsensusAdapterMetrics,
        ConsensusClient,
    },
    consensus_handler::ConsensusHandlerInitializer,
    consensus_manager::{ConsensusManager, ConsensusManagerTrait, UpdatableConsensusClient},
    consensus_validator::{IotaTxValidator, IotaTxValidatorMetrics},
    epoch::{
        committee_store::CommitteeStore, consensus_store_pruner::ConsensusStorePruner,
        epoch_metrics::EpochMetrics, randomness::RandomnessManager,
        reconfiguration::ReconfigurationInitiator,
    },
    execution_cache::build_execution_cache,
    global_state_hasher::{GlobalStateHashMetrics, GlobalStateHasher},
    grpc_indexes::{GRPC_INDEXES_DIR, GrpcIndexesStore},
    jsonrpc_index::IndexStore,
    module_cache_metrics::ResolverMetrics,
    overload_monitor::{consensus_queue_overload_monitor, overload_monitor},
    safe_client::SafeClientMetricsBase,
    signature_verifier::SignatureVerifierMetrics,
    storage::{GrpcReadStore, RocksDbStore},
    transaction_orchestrator::TransactionOrchestrator,
    validator_tx_finalizer::ValidatorTxFinalizer,
};
use iota_genesis_common::MigrationTxDataExt;
use iota_grpc_server::{GrpcReader, GrpcServerHandle, start_grpc_server};
use iota_json_rpc::{
    JsonRpcServerBuilder, coin_api::CoinReadApi, governance_api::GovernanceReadApi,
    indexer_api::IndexerApi, move_utils::MoveUtils, read_api::ReadApi,
    transaction_builder_api::TransactionBuilderApi,
    transaction_execution_api::TransactionExecutionApi,
};
use iota_json_rpc_api::JsonRpcMetrics;
use iota_macros::{fail_point, fail_point_async, replay_log};
use iota_metrics::{
    RegistryID, RegistryService,
    metrics_network::{MetricsMakeCallbackHandler, NetworkConnectionMetrics, NetworkMetrics},
    server_timing_middleware, spawn_monitored_task,
};
use iota_names::config::IotaNamesConfig;
use iota_network::{
    api::{ValidatorPeerServer, ValidatorServer, ValidatorV2Server},
    discovery,
    discovery::TrustedPeerChangeEvent,
    randomness, state_sync,
};
use iota_network_stack::server::{IOTA_TLS_SERVER_NAME, ServerBuilder};
use iota_node_transaction_builder::NodeTransactionBuilderLedgerClient;
use iota_protocol_config::{ProtocolConfig, ProtocolVersion};
use iota_sdk_types::{
    RandomnessRound,
    crypto::{Intent, IntentMessage, IntentScope},
};
use iota_snapshot::uploader::StateSnapshotUploader;
use iota_storage::{
    http_key_value_store::HttpKVStore,
    key_value_store::{FallbackTransactionKVStore, TransactionKeyValueStore},
    key_value_store_metrics::KeyValueStoreMetrics,
};
use iota_types::{
    base_types::{AuthorityName, ConciseableName, EpochId},
    committee::Committee,
    crypto::{AuthoritySignature, IotaAuthoritySignature, KeypairTraits},
    digests::ChainIdentifier,
    error::{IotaError, IotaResult},
    executable_transaction::VerifiedExecutableTransaction,
    execution_config_utils::to_binary_config,
    full_checkpoint_content::CheckpointData,
    iota_system_state::{
        IotaSystemState, IotaSystemStateTrait,
        epoch_start_iota_system_state::{EpochStartSystemState, EpochStartSystemStateTrait},
    },
    messages_checkpoint::CheckpointSummaryExt,
    messages_consensus::{
        AuthorityCapabilitiesV1, ConsensusTransaction, ConsensusTransactionKind,
        SignedAuthorityCapabilitiesV1, TransactionDenyRuleProposal,
    },
    messages_grpc::HandleCapabilityNotificationRequestV1,
    quorum_driver_types::QuorumDriverEffectsQueueResult,
    supported_protocol_versions::SupportedProtocolVersions,
    transaction::{SenderSignedTransactionAPI, TransactionEnvelope, VerifiedCertificate},
};
use prometheus_filtered::Registry;
#[cfg(msim)]
use simulator::*;
use tap::tap::TapFallible;
use tokio::{
    sync::{Mutex, broadcast, mpsc, watch},
    task::{JoinHandle, JoinSet},
};
use tokio_util::sync::CancellationToken;
use tower::ServiceBuilder;
use tracing::{Instrument, debug, error, error_span, info, trace_span, warn};
use typed_store::{
    DBMetrics,
    rocks::{check_and_mark_db_corruption, default_db_options, unmark_db_corruption},
};

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

pub mod admin;
mod handle;
pub mod metrics;

pub struct ValidatorComponents {
    validator_server_handle: SpawnOnce,
    validator_overload_monitor_handle: Option<JoinHandle<()>>,
    /// Handle for the consensus queue overload monitor task, present only
    /// when the certificate-less (P-COOL) flow is enabled. The
    /// task self-terminates via `Weak` references; this handle exists purely
    /// for ownership clarity.
    consensus_queue_overload_monitor_handle: Option<JoinHandle<()>>,
    /// Handle for the soft-lock expiry sweep task. The task self-terminates
    /// via a `Weak` reference; this handle exists purely for ownership clarity.
    soft_lock_sweep_handle: JoinHandle<()>,
    overload_notifier_handle: Option<JoinHandle<()>>,
    consensus_manager: Arc<ConsensusManager>,
    consensus_store_pruner: ConsensusStorePruner,
    consensus_adapter: Arc<ConsensusAdapter>,
    soft_locks: Arc<PreConsensusSoftLocks>,
    // Keeping the handle to the checkpoint service tasks to shut them down during reconfiguration.
    checkpoint_service_tasks: JoinSet<()>,
    checkpoint_metrics: Arc<CheckpointMetrics>,
    iota_tx_validator_metrics: Arc<IotaTxValidatorMetrics>,
    validator_registry_id: RegistryID,
}

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

    pub(super) struct SimState {
        pub sim_node: iota_simulator::runtime::NodeHandle,
        pub sim_safe_mode_expected: AtomicBool,
        _leak_detector: iota_simulator::NodeLeakDetector,
    }

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

#[derive(Clone)]
pub struct ServerVersion {
    pub bin: &'static str,
    pub version: &'static str,
}

impl ServerVersion {
    pub fn new(bin: &'static str, version: &'static str) -> Self {
        Self { bin, version }
    }
}

impl std::fmt::Display for ServerVersion {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(self.bin)?;
        f.write_str("/")?;
        f.write_str(self.version)
    }
}

pub struct IotaNode {
    config: NodeConfig,
    validator_components: Mutex<Option<ValidatorComponents>>,
    /// The http server responsible for serving JSON-RPC
    _http_server: Option<iota_http::ServerHandle>,
    state: Arc<AuthorityState>,
    transaction_orchestrator: Option<Arc<TransactionOrchestrator<NetworkAuthorityClient>>>,
    registry_service: RegistryService,
    metrics: Arc<IotaNodeMetrics>,

    _discovery: discovery::Handle,
    state_sync_handle: state_sync::Handle,
    randomness_handle: randomness::Handle,
    checkpoint_store: Arc<CheckpointStore>,
    state_sync_store: RocksDbStore,
    global_state_hasher: Mutex<Option<Arc<GlobalStateHasher>>>,
    connection_monitor_status: Arc<ConnectionMonitorStatus>,

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

    /// Broadcast channel to notify [`DiscoveryEventLoop`] for new validator
    /// peers.
    trusted_peer_change_tx: watch::Sender<TrustedPeerChangeEvent>,

    backpressure_manager: Arc<BackpressureManager>,

    checkpoint_progress_tracker: Arc<CheckpointProgressTracker>,

    #[cfg(msim)]
    sim_state: SimState,

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

    /// Handle to the gRPC server for gRPC streaming and graceful shutdown
    grpc_server_handle: Mutex<Option<GrpcServerHandle>>,

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

    /// Runtime that hosts the client-facing servers and their per-request
    /// handlers, isolating external request load from the node core.
    serving_rt_handle: tokio::runtime::Handle,
}

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

impl IotaNode {
    /// Starts a node that hosts the client-facing servers on the caller's
    /// runtime, alongside everything else.
    ///
    /// This is intentional: this entry point serves the in-process nodes of
    /// iota-swarm, where a separate serving runtime is either impossible
    /// (simtests must keep every task on the simulator's deterministic
    /// scheduler) or not worth the threads (thread-mode swarm runs many nodes
    /// per process). Only the `iota-node` binary isolates client-facing
    /// request handling on the dedicated serving runtime of `IotaRuntimes`,
    /// via [`IotaNode::start_async`].
    pub async fn start(
        config: NodeConfig,
        registry_service: RegistryService,
    ) -> Result<Arc<IotaNode>> {
        Self::start_async(
            config,
            registry_service,
            ServerVersion::new("iota-node", "unknown"),
            tokio::runtime::Handle::current(),
        )
        .await
    }

    /// Starts a background task that polls the authority's load shedding
    /// percentage and broadcasts changes to other validators via consensus.
    /// Returns the task handle if the feature flag is enabled, or `None`
    /// otherwise.
    fn start_overload_notifier(
        config: &NodeConfig,
        state: Arc<AuthorityState>,
        epoch_store: Arc<AuthorityPerEpochStore>,
        consensus_adapter: Arc<ConsensusAdapter>,
    ) -> Option<JoinHandle<()>> {
        if !epoch_store.protocol_config().enable_pcool_flow() {
            return None;
        }

        let poll_interval = config.authority_overload_config.overload_monitor_interval;
        let authority_name = state.name;

        Some(spawn_monitored_task!(async move {
            // Seed from the percentage this authority last broadcasted
            let mut last_notified_percentage: u32 = epoch_store
                .load_overload_notification(&authority_name)
                .unwrap_or(0) as u32;
            loop {
                tokio::time::sleep(poll_interval).await;
                let current = state
                    .overload_info
                    .local_load_shedding_percentage
                    .load(std::sync::atomic::Ordering::Relaxed);
                if current != last_notified_percentage {
                    last_notified_percentage = current;
                    let transaction = ConsensusTransaction::new_overload_notification_v1(
                        authority_name,
                        current as u8,
                    );
                    if let Err(e) = consensus_adapter.submit(transaction, None, &epoch_store) {
                        tracing::warn!(
                            "Failed to submit overload notification to consensus: {:?}",
                            e
                        );
                    }
                }
            }
        }))
    }

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

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

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

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

        let chain_identifier = ChainIdentifier::from(*genesis.checkpoint().digest());
        info!("IOTA chain identifier: {chain_identifier}");

        // Check and set the db_corrupted flag
        let db_corrupted_path = &config.db_path().join("status");
        if let Err(err) = check_and_mark_db_corruption(db_corrupted_path) {
            panic!("Failed to check database corruption: {err}");
        }

        // Initialize metrics to track db usage before creating any stores
        DBMetrics::init(&prometheus_registry);

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

        // Monitor the node-core and serving runtimes so that worker-thread
        // starvation between them is observable. Gated out of simtests, where
        // tokio runs under the deterministic simulator.
        #[cfg(not(msim))]
        {
            let runtime_monitor_metrics =
                iota_metrics::runtime_metrics::RuntimeMonitorMetrics::new(&prometheus_registry);
            iota_metrics::runtime_metrics::start_runtime_monitor(
                "iota_node",
                &tokio::runtime::Handle::current(),
                runtime_monitor_metrics.clone(),
            );
            iota_metrics::runtime_metrics::start_runtime_monitor(
                "serving",
                &serving_rt_handle,
                runtime_monitor_metrics,
            );
        }

        // Register uptime metric
        prometheus_registry
            .register(iota_metrics::uptime_metric(
                if is_validator {
                    "validator"
                } else {
                    "fullnode"
                },
                server_version.version,
                &chain_identifier.to_string(),
            ))
            .expect("Failed registering uptime metric");

        // If genesis come with some migration data then load them into memory from the
        // file path specified in config.
        let migration_tx_data = if genesis.contains_migrations() {
            // Here the load already verifies that the content of the migration blob is
            // valid in respect to the content found in genesis
            Some(config.load_migration_tx_data()?)
        } else {
            None
        };

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

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

        // By default, only enable write stall on validators for perpetual db.
        let enable_write_stall = config.enable_db_write_stall.unwrap_or(is_validator);
        let perpetual_tables_options = AuthorityPerpetualTablesOptions {
            enable_write_stall,
            compaction_filter,
        };
        let perpetual_tables = Arc::new(AuthorityPerpetualTables::open(
            &config.db_path().join("store"),
            Some(perpetual_tables_options),
        ));
        let is_genesis = perpetual_tables
            .database_is_empty()
            .expect("Database read should not fail at init.");
        let checkpoint_store = CheckpointStore::new_with_contents_cache(
            &config.db_path().join("checkpoints"),
            FullCheckpointContentsCache::new(
                config
                    .full_checkpoint_contents_cache_size_mb
                    .saturating_mul(1024 * 1024),
                FullCheckpointContentsCacheMetrics::new(&prometheus_registry),
            ),
        );
        let backpressure_manager =
            BackpressureManager::new_from_checkpoint_store(&checkpoint_store);

        let perpetual_tables_for_progress = perpetual_tables.clone();
        let store = AuthorityStore::open(
            perpetual_tables,
            &genesis,
            &config,
            &prometheus_registry,
            migration_tx_data.as_ref(),
        )
        .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_config,
            &prometheus_registry,
            &store,
            backpressure_manager.clone(),
        );

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

        let chain = match config.chain_override_for_testing {
            Some(chain) => chain,
            None => chain_identifier.chain(),
        };

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

        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 IOTA conservation at genesis");
            // When we are opening the db table, the only time when it's safe to
            // check IOTA conservation is at genesis. Otherwise we may be in the middle of
            // an epoch and the IOTA conservation check will fail. This also initialize
            // the expected_network_iota_amount table.
            cache_traits
                .reconfig_api
                .try_expensive_check_iota_conservation(&epoch_store, None)
                .expect("IOTA 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,
        );

        // Database has everything from genesis, set corrupted key to 0
        unmark_db_corruption(db_corrupted_path)?;

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

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

        let grpc_indexes_store = if is_full_node && config.enable_grpc_api {
            Some(Arc::new(
                GrpcIndexesStore::new(
                    config.db_path().join(GRPC_INDEXES_DIR),
                    Arc::clone(&store),
                    &checkpoint_store,
                )
                .await,
            ))
        } else {
            None
        };

        // Seed the open epoch's `epoch_info` row before services start.
        checkpoint_store
            .seed_epoch_info(&store, chain_identifier)
            .expect("failed to seed the epoch_info chain");

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

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

        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(),
            is_full_node,
        )?;

        let checkpoint_progress_tracker = Arc::new(CheckpointProgressTracker::new());

        let mut genesis_objects = genesis.objects().to_vec();
        if let Some(migration_tx_data) = migration_tx_data.as_ref() {
            genesis_objects.extend(migration_tx_data.get_objects());
        }

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

        info!("create authority state");
        let state = AuthorityState::new(
            authority_name,
            secret,
            config.supported_protocol_versions.unwrap(),
            store.clone(),
            cache_traits.clone(),
            epoch_store.clone(),
            committee_store.clone(),
            index_store.clone(),
            grpc_indexes_store,
            checkpoint_store.clone(),
            &prometheus_registry,
            &genesis_objects,
            config.clone(),
            validator_tx_finalizer,
            chain_identifier,
            pruner_db,
            Some(checkpoint_progress_tracker.clone()),
            config.policy_config.clone(),
            config.firewall_config.clone(),
        )
        .await;

        // ensure genesis and migration txs were executed
        if epoch_store.epoch() == 0 {
            let genesis_tx = &genesis.transaction();
            let span = error_span!("genesis_txn", tx_digest = ?genesis_tx.digest());
            // Execute genesis transaction
            Self::execute_transaction_immediately_at_zero_epoch(
                &state,
                &epoch_store,
                genesis_tx,
                span,
            )
            .await;

            // Execute migration transactions if present
            if let Some(migration_tx_data) = migration_tx_data {
                for (tx_digest, (tx, _, _)) in migration_tx_data.txs_data() {
                    let span = error_span!("migration_txn", tx_digest = ?tx_digest);
                    Self::execute_transaction_immediately_at_zero_epoch(
                        &state,
                        &epoch_store,
                        tx,
                        span,
                    )
                    .await;
                }
            }
        }

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

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

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

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

        // Run the JSON-RPC server (and its per-request handlers) on the serving
        // runtime. `iota_http::Builder::serve` spawns the accept loop via
        // `Handle::current()`, so the builder must execute on the serving runtime.
        let http_server = serving_rt_handle
            .spawn({
                let state = state.clone();
                let transaction_orchestrator = transaction_orchestrator.clone();
                let config = config.clone();
                let prometheus_registry = prometheus_registry.clone();
                async move {
                    build_http_server(
                        state,
                        &transaction_orchestrator,
                        &config,
                        &prometheus_registry,
                    )
                    .await
                }
            })
            .await
            .expect("Failed to join JSON-RPC server startup task")?;

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

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

        let network_connection_metrics =
            NetworkConnectionMetrics::new("iota", &registry_service.default_registry());

        let authority_names_to_peer_ids = ArcSwap::from_pointee(authority_names_to_peer_ids);

        let (_connection_monitor_handle, connection_statuses) = ConnectionMonitor::spawn(
            p2p_network.downgrade(),
            network_connection_metrics,
            HashMap::new(),
            None,
        );

        let connection_monitor_status = ConnectionMonitorStatus {
            connection_statuses,
            authority_names_to_peer_ids,
        };

        let connection_monitor_status = Arc::new(connection_monitor_status);
        let iota_node_metrics =
            Arc::new(IotaNodeMetrics::new(&registry_service.default_registry()));

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

        // Convert transaction orchestrator to executor trait object for gRPC server
        // Note that the transaction_orchestrator (so as executor) will be None if it is
        // a validator node or run_with_range is set
        let executor: Option<Arc<dyn iota_types::transaction_executor::TransactionExecutor>> =
            transaction_orchestrator
                .clone()
                .map(|o| o as Arc<dyn iota_types::transaction_executor::TransactionExecutor>);

        // Run the gRPC read API server (and its per-request handlers) on the
        // serving runtime, for the same reason as the JSON-RPC server above.
        let grpc_server_handle = serving_rt_handle
            .spawn({
                let config = config.clone();
                let state = state.clone();
                let state_sync_store = state_sync_store.clone();
                let prometheus_registry = prometheus_registry.clone();
                async move {
                    build_grpc_server(
                        &config,
                        state,
                        state_sync_store,
                        executor,
                        &prometheus_registry,
                        server_version,
                    )
                    .await
                }
            })
            .await
            .expect("Failed to join gRPC server startup task")?;

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

            components.consensus_adapter.submit_recovered(&epoch_store);

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

            Some(components)
        } else {
            None
        };

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

        let node = Self {
            config,
            validator_components: Mutex::new(validator_components),
            _http_server: http_server,
            state,
            transaction_orchestrator,
            registry_service,
            metrics: iota_node_metrics,

            _discovery: discovery_handle,
            state_sync_handle,
            randomness_handle,
            checkpoint_store,
            state_sync_store,
            global_state_hasher: Mutex::new(Some(global_state_hasher)),
            end_of_epoch_channel,
            connection_monitor_status,
            trusted_peer_change_tx,
            backpressure_manager,
            checkpoint_progress_tracker: checkpoint_progress_tracker.clone(),

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

            _state_snapshot_uploader_handle: state_snapshot_handle,
            shutdown_channel_tx: shutdown_channel,

            grpc_server_handle: Mutex::new(grpc_server_handle),

            auth_agg,

            serving_rt_handle,
        };

        info!("IotaNode 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);
            }
        });

        node.checkpoint_progress_tracker
            .spawn_logging_task(node.checkpoint_store.clone(), perpetual_tables_for_progress);

        Ok(node)
    }

    pub fn subscribe_to_epoch_change(&self) -> broadcast::Receiver<IotaSystemState> {
        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()
    }

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

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

    pub fn set_override_protocol_upgrade_buffer_stake(
        &self,
        epoch: EpochId,
        buffer_stake_bps: u64,
    ) -> IotaResult {
        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) -> IotaResult {
        let epoch_store = self.state.epoch_store_for_testing();
        self.close_epoch(&epoch_store).await
    }

    /// Creates a StateSnapshotUploader and starts it if the StateSnapshotConfig
    /// is set.
    fn start_state_snapshot(
        config: &NodeConfig,
        prometheus_registry: &Registry,
        checkpoint_store: Arc<CheckpointStore>,
        is_full_node: bool,
    ) -> Result<Option<tokio::sync::broadcast::Sender<()>>> {
        if let Some(remote_store_config) = &config.state_snapshot_write_config.object_store_config {
            // Snapshot publication is a fullnode-only role.
            anyhow::ensure!(
                is_full_node,
                "Snapshot upload is configured, but this node is a validator. \
                 Snapshot publication is only supported on fullnodes. Remove the \
                 `state_snapshot_write_config.object_store_config` setting, or move \
                 the upload to a fullnode."
            );
            let snapshot_uploader = StateSnapshotUploader::new(
                &config.db_checkpoint_path(),
                &config.snapshot_path(),
                remote_store_config.clone(),
                config.state_snapshot_write_config.concurrency,
                60,
                prometheus_registry,
                checkpoint_store,
            )?;
            Ok(Some(snapshot_uploader.start()))
        } else {
            Ok(None)
        }
    }

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

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

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

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

            let inbound_network_metrics =
                NetworkMetrics::new("iota", "inbound", prometheus_registry);
            let outbound_network_metrics =
                NetworkMetrics::new("iota", "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(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::DEBUG)),
                )
                .layer(CallbackLayer::new(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!("iota-{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((
            p2p_network,
            discovery_handle,
            state_sync_handle,
            randomness_handle,
        ))
    }

    /// Asynchronously constructs and initializes the components necessary for
    /// the validator node.
    async fn construct_validator_components(
        config: NodeConfig,
        state: Arc<AuthorityState>,
        committee: Arc<Committee>,
        epoch_store: Arc<AuthorityPerEpochStore>,
        checkpoint_store: Arc<CheckpointStore>,
        state_sync_handle: state_sync::Handle,
        randomness_handle: randomness::Handle,
        global_state_hasher: Weak<GlobalStateHasher>,
        backpressure_manager: Arc<BackpressureManager>,
        connection_monitor_status: Arc<ConnectionMonitorStatus>,
        registry_service: &RegistryService,
        serving_rt_handle: tokio::runtime::Handle,
    ) -> Result<ValidatorComponents> {
        let mut config_clone = config.clone();
        let consensus_config = config_clone
            .consensus_config
            .as_mut()
            .ok_or_else(|| anyhow!("Validator is missing consensus config"))?;
        let validator_registry = registry_service.new_registry_custom(None, None)?;
        let validator_registry_id = registry_service.add(validator_registry.clone());

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

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

        let soft_locks = Arc::new(if config.enable_soft_locking {
            PreConsensusSoftLocks::new()
        } else {
            info!("pre-consensus soft-locking disabled via node config");
            PreConsensusSoftLocks::disabled()
        });

        let checkpoint_metrics = CheckpointMetrics::new(&validator_registry);
        let iota_tx_validator_metrics = IotaTxValidatorMetrics::new(&validator_registry);
        let validator_service_metrics = Arc::new(ValidatorServiceMetrics::new(&validator_registry));

        // Spawn the soft-lock sweep once for the lifetime of this validator
        // instance. The task holds only a `Weak<PreConsensusSoftLocks>` so it
        // stops itself automatically: each iteration it tries to upgrade the
        // weak reference, and when all strong `Arc` owners have been dropped
        // (i.e. `ValidatorComponents` is destructured and the old epoch store
        // is released after an epoch transition that removes us from the
        // committee) the upgrade returns `None` and the loop exits. No explicit
        // `abort()` is needed. The same `Arc<PreConsensusSoftLocks>` is reused
        // across epoch transitions (see `start_epoch_specific_validator_components`),
        // so the task keeps running uninterrupted while the node remains a validator.
        let soft_lock_sweep_handle = PreConsensusSoftLocks::spawn_sweep(
            Arc::downgrade(&soft_locks),
            validator_service_metrics.clone(),
        );

        let validator_server_handle = Self::start_grpc_validator_service(
            &config,
            state.clone(),
            consensus_adapter.clone(),
            &validator_registry,
            soft_locks.clone(),
            validator_service_metrics.clone(),
            serving_rt_handle,
        )
        .await?;

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

        // Starts a monitor that periodically refreshes the
        // `consensus_queue_load_shedding_percentage` metric. Without this, the
        // metric goes stale once gRPC traffic stops (the only other update
        // path is `AuthorityState::check_consensus_queue_graduated_limits`, called on
        // each inbound tx). Used in the certificate-less (P-COOL)
        // mode.
        let consensus_queue_overload_monitor_handle =
            if epoch_store.protocol_config().enable_pcool_flow() {
                let consensus_queue_monitor_authority_state = Arc::downgrade(&state);
                let consensus_queue_monitor_consensus_adapter = Arc::downgrade(&consensus_adapter);
                let consensus_queue_monitor_interval =
                    config.authority_overload_config.overload_monitor_interval;
                Some(spawn_monitored_task!(consensus_queue_overload_monitor(
                    consensus_queue_monitor_authority_state,
                    consensus_queue_monitor_consensus_adapter,
                    consensus_queue_monitor_interval,
                )))
            } else {
                None
            };

        Self::start_epoch_specific_validator_components(
            &config,
            state.clone(),
            consensus_adapter,
            checkpoint_store,
            epoch_store,
            state_sync_handle,
            randomness_handle,
            consensus_manager,
            consensus_store_pruner,
            global_state_hasher,
            backpressure_manager,
            soft_locks,
            validator_server_handle,
            validator_overload_monitor_handle,
            consensus_queue_overload_monitor_handle,
            soft_lock_sweep_handle,
            checkpoint_metrics,
            iota_tx_validator_metrics,
            validator_registry_id,
        )
        .await
    }

    /// Initializes and starts components specific to the current
    /// epoch for the validator node.
    async fn start_epoch_specific_validator_components(
        config: &NodeConfig,
        state: Arc<AuthorityState>,
        consensus_adapter: Arc<ConsensusAdapter>,
        checkpoint_store: Arc<CheckpointStore>,
        epoch_store: Arc<AuthorityPerEpochStore>,
        state_sync_handle: state_sync::Handle,
        randomness_handle: randomness::Handle,
        consensus_manager: Arc<ConsensusManager>,
        consensus_store_pruner: ConsensusStorePruner,
        global_state_hasher: Weak<GlobalStateHasher>,
        backpressure_manager: Arc<BackpressureManager>,
        soft_locks: Arc<PreConsensusSoftLocks>,
        validator_server_handle: SpawnOnce,
        validator_overload_monitor_handle: Option<JoinHandle<()>>,
        consensus_queue_overload_monitor_handle: Option<JoinHandle<()>>,
        soft_lock_sweep_handle: JoinHandle<()>,
        checkpoint_metrics: Arc<CheckpointMetrics>,
        iota_tx_validator_metrics: Arc<IotaTxValidatorMetrics>,
        validator_registry_id: RegistryID,
    ) -> Result<ValidatorComponents> {
        let checkpoint_service = Self::build_checkpoint_service(
            config,
            consensus_adapter.clone(),
            checkpoint_store.clone(),
            epoch_store.clone(),
            state.clone(),
            state_sync_handle,
            global_state_hasher,
            checkpoint_metrics.clone(),
        );

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

        consensus_adapter.swap_low_scoring_authorities(low_scoring_authorities.clone());

        // Wire pre-consensus soft locks to the epoch store so that
        // post-consensus processing can release locks once permanent locks are
        // quarantined. Clear stale locks from the previous epoch and spawn a
        // background sweep task.
        soft_locks.clear();
        epoch_store.set_soft_locks(soft_locks.clone());

        // A validator cannot participate without a randomness manager. Aborting here
        // fails loudly at the real cause (startup or reconfiguration) instead
        // of leaving the node to panic later on the first consensus commit that
        // unconditionally expects one.
        let randomness_manager = match RandomnessManager::try_new(
            Arc::downgrade(&epoch_store),
            Box::new(consensus_adapter.clone()),
            randomness_handle,
            config.authority_key_pair(),
        )
        .await
        {
            Ok(randomness_manager) => randomness_manager,
            Err(err) => {
                fatal!(
                    "validator cannot start epoch {} without a randomness manager: {err}",
                    epoch_store.epoch()
                );
            }
        };
        epoch_store
            .set_randomness_manager(randomness_manager)
            .await?;

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

        info!("Starting consensus manager asynchronously");

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

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

        let overload_notifier_handle = Self::start_overload_notifier(
            config,
            state.clone(),
            epoch_store.clone(),
            consensus_adapter.clone(),
        );

        Ok(ValidatorComponents {
            validator_server_handle,
            validator_overload_monitor_handle,
            consensus_queue_overload_monitor_handle,
            soft_lock_sweep_handle,
            overload_notifier_handle,
            consensus_manager,
            consensus_store_pruner,
            consensus_adapter,
            soft_locks,
            checkpoint_service_tasks,
            checkpoint_metrics,
            iota_tx_validator_metrics,
            validator_registry_id,
        })
    }

    /// Starts the checkpoint service for the validator node, initializing
    /// necessary components and settings.
    /// The function ensures proper initialization of the checkpoint service,
    /// preparing it to handle checkpoint creation and submission to consensus,
    /// while also setting up the necessary monitoring and synchronization
    /// mechanisms.
    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,
        global_state_hasher: Weak<GlobalStateHasher>,
        checkpoint_metrics: Arc<CheckpointMetrics>,
    ) -> Arc<CheckpointService> {
        let epoch_start_timestamp_ms = epoch_store.epoch_start_state().epoch_start_timestamp_ms();
        let epoch_duration_ms = epoch_store.epoch_start_state().epoch_duration_ms();

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

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

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

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

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

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

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

        // Each service gets its own concurrency limit so that a flood of client
        // transaction submissions (Validator / ValidatorV2) cannot crowd the
        // validator-peer RPCs sharing this listener out of admission slots.
        // The config value is per core, so the same config scales with the
        // hardware; the effective limit is computed on the machine the server
        // actually runs on.
        let concurrency_limit = config.grpc_concurrency_limit_per_core.saturating_mul(
            NonZeroUsize::new(iota_core::runtime::available_cpu_cores())
                .unwrap_or(NonZeroUsize::MIN),
        );
        let load_shed = config.grpc_load_shed.unwrap_or_default();

        let server_conf = iota_network_stack::config::Config::new();
        let server_builder =
            ServerBuilder::from_config(&server_conf, GrpcMetrics::new(prometheus_registry))
                .add_service_with_concurrency_limit(
                    ValidatorServer::new(validator_service.clone()),
                    concurrency_limit,
                    load_shed,
                )
                .add_service_with_concurrency_limit(
                    ValidatorV2Server::new(validator_service.clone()),
                    concurrency_limit,
                    load_shed,
                )
                .add_service_with_concurrency_limit(
                    ValidatorPeerServer::new(validator_service),
                    concurrency_limit,
                    load_shed,
                );

        let tls_config = iota_tls::create_rustls_server_config(
            config.network_key_pair().copy().private(),
            IOTA_TLS_SERVER_NAME.to_string(),
        );

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

        let bind_future = async move {
            let server = server_builder
                .bind(&network_address, Some(tls_config))
                .await
                .map_err(|err| anyhow!("Failed to bind to {network_address}: {err}"))?;

            let local_addr = server.local_addr();
            info!("Listening to traffic on {local_addr}");

            Ok(server)
        };

        Ok(SpawnOnce::new(bind_future, serving_rt_handle))
    }

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

        for tx in epoch_store.get_all_pending_consensus_transactions() {
            match tx.kind {
                // TODO: what to do with UserTransactionV1 here? It seems like this only applies to
                //  optimistically executed owned-object transactions that possibly didn't go
                //  through  consensus before the node restarted. UserTransactionsV1
                //  always needs to go  through consensus, so it will be replayed
                //  there, just like shared object  transactions.
                //
                // Shared object txns
                // cannot be re-executed at this  point, because we must wait for
                // consensus replay to assign shared  object versions.
                ConsensusTransactionKind::CertifiedTransaction(tx)
                    if !tx.contains_shared_object() =>
                {
                    let tx = *tx;
                    // new_unchecked is safe because we never submit a transaction to consensus
                    // without verifying it
                    let tx = VerifiedExecutableTransaction::new_from_certificate(
                        VerifiedCertificate::new_unchecked(tx),
                    );
                    // we only need to re-execute if we previously signed the effects (which
                    // indicates we returned the effects to a client).
                    if let Some(fx_digest) = epoch_store
                        .get_signed_effects_digest(tx.digest())
                        .expect("db error")
                    {
                        pending_consensus_certificates.push((tx, fx_digest));
                    } else {
                        additional_certs.push(tx);
                    }
                }
                _ => (),
            }
        }

        let digests = pending_consensus_certificates
            .iter()
            .map(|(tx, _)| *tx.digest())
            .collect::<Vec<_>>();

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

        state.enqueue_with_expected_effects_digest(pending_consensus_certificates, epoch_store);
        state.enqueue_transactions_for_execution(additional_certs, epoch_store);

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

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

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

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

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

    /// Clone an AuthorityAggregator from the transaction orchestrator, if
    /// this is a fullnode. The snapshot goes stale after an epoch change;
    /// call again for 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()
    }

    /// Read-only client for the SDK's `TransactionBuilder` backed by this
    /// node's local state instead of a remote endpoint.
    pub fn transaction_builder_ledger_client(&self) -> NodeTransactionBuilderLedgerClient {
        let reader = Arc::new(GrpcReadStore::new(
            self.state.clone(),
            self.state_sync_store.clone(),
        ));
        NodeTransactionBuilderLedgerClient::new(reader)
    }

    /// Subscribe to the quorum driver's effects stream; errors while the
    /// quorum driver is not the currently served flow on this node.
    pub fn subscribe_to_transaction_orchestrator_effects(
        &self,
    ) -> Result<tokio::sync::broadcast::Receiver<QuorumDriverEffectsQueueResult>> {
        self.transaction_orchestrator
            .as_ref()
            .ok_or_else(|| {
                anyhow::anyhow!("Transaction Orchestrator is not enabled in this node.")
            })?
            .subscribe_to_effects_queue()
            .ok_or_else(|| {
                anyhow::anyhow!(
                    "Effects queue is not available: the quorum driver is not the currently \
                     served flow on this node."
                )
            })
    }

    /// This function awaits the completion of checkpoint execution of the
    /// current epoch, after which it initiates reconfiguration of the
    /// entire system. This function also handles role changes for the node when
    /// epoch changes and, if the node is a validator, advertises capabilities
    /// and submits the local deny rule proposal to consensus.
    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()
            );

            // Create closures that handle gRPC type conversion
            let data_sender = if let Ok(guard) = self.grpc_server_handle.try_lock() {
                guard.as_ref().map(|handle| {
                    let tx = handle.checkpoint_data_broadcaster().clone();
                    Box::new(move |data: &CheckpointData| {
                        tx.send_traced(data);
                    }) as Box<dyn Fn(&CheckpointData) + Send + Sync>
                })
            } else {
                None
            };

            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(),
                data_sender,
                Some(self.checkpoint_progress_tracker.clone()),
            );

            let run_with_range = self.config.run_with_range;

            let cur_epoch_store = self.state.load_epoch_store_one_call_per_task();

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

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

                let config = cur_epoch_store.protocol_config();
                let binary_config = to_binary_config(config);
                let transaction = ConsensusTransaction::new_capability_notification_v1(
                    AuthorityCapabilitiesV1::new(
                        self.state.name,
                        cur_epoch_store.get_chain(),
                        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),
                        self.state
                            .get_available_system_packages(&binary_config)
                            .await,
                    ),
                );
                info!(?transaction, "submitting capabilities to consensus");
                components
                    .consensus_adapter
                    .submit(transaction, None, &cur_epoch_store)?;

                // Announce the local deny rules. Recorded proposals are
                // epoch-scoped, so this re-announces on every epoch change.
                // The empty set is announced too: every committee member
                // attests its configuration each epoch, so silence means
                // offline rather than "no rules".
                if config.deny_rule_governance() {
                    let proposed_rules = self.config.transaction_deny_config.to_deny_rule_set();
                    let recorded = cur_epoch_store.recorded_deny_rule_proposal(&self.state.name);
                    let transaction = ConsensusTransaction::new_transaction_deny_rule_proposal(
                        TransactionDenyRuleProposal::new(
                            self.state.name,
                            proposed_rules,
                            recorded.map(|p| p.generation),
                        ),
                    );
                    info!(
                        tracking_id = ?transaction.get_tracking_id(),
                        "submitting deny rule proposal to consensus"
                    );
                    components
                        .consensus_adapter
                        .submit(transaction, None, &cur_epoch_store)?;
                }
            } else if self.state.is_active_validator(&cur_epoch_store)
                && cur_epoch_store
                    .protocol_config()
                    .track_non_committee_eligible_validators()
            {
                // Send signed capabilities to committee validators if we are a non-committee
                // validator in a separate task to not block the caller. Sending is done only if
                // the feature flag supporting it is enabled.
                let epoch_store = cur_epoch_store.clone();
                let node_clone = self.clone();
                spawn_monitored_task!(epoch_store.clone().within_alive_epoch(async move {
                    node_clone
                        .send_signed_capability_notification_to_committee_with_retry(&epoch_store)
                        .instrument(trace_span!(
                            "send_signed_capability_notification_to_committee_with_retry"
                        ))
                        .await;
                }));
            }

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

            if stop_condition == StopReason::RunWithRangeCondition {
                IotaNode::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()
                .try_get_iota_system_state_object_unsafe()
                .expect("Read IOTA 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()) {
                if 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_iota_committee();
            let next_epoch = next_epoch_committee.epoch();
            assert_eq!(cur_epoch_store.epoch() + 1, next_epoch);

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

            fail_point_async!("reconfig_delay");

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

            cur_epoch_store.record_epoch_reconfig_start_time_metric();

            send_trusted_peer_change(
                &self.config,
                &self.trusted_peer_change_tx,
                &new_epoch_start_state,
            );

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

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

            let new_validator_components = if let Some(ValidatorComponents {
                validator_server_handle,
                validator_overload_monitor_handle,
                consensus_queue_overload_monitor_handle,
                soft_lock_sweep_handle,
                overload_notifier_handle,
                consensus_manager,
                consensus_store_pruner,
                consensus_adapter,
                soft_locks,
                mut checkpoint_service_tasks,
                checkpoint_metrics,
                iota_tx_validator_metrics,
                validator_registry_id,
            }) = validator_components_lock_guard.take()
            {
                info!("Reconfiguring the validator.");
                // Cancel the old overload notifier task so a new one can be
                // started for the next epoch.
                if let Some(handle) = overload_notifier_handle {
                    handle.abort();
                }
                // Cancel the old checkpoint service tasks.
                // Waiting for checkpoint builder to finish gracefully is not possible, because
                // it may wait on transactions while consensus on peers have
                // already shut down.
                checkpoint_service_tasks.abort_all();
                while let Some(result) = checkpoint_service_tasks.join_next().await {
                    if let Err(err) = result {
                        if err.is_panic() {
                            std::panic::resume_unwind(err.into_panic());
                        }
                        warn!("Error in checkpoint service task: {:?}", err);
                    }
                }
                info!("Checkpoint service has shut down.");

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

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

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

                consensus_store_pruner.prune(next_epoch).await;

                if self.state.is_committee_validator(&new_epoch_store) {
                    // Only restart consensus if this node is still a validator in the new epoch.
                    Some(
                        Self::start_epoch_specific_validator_components(
                            &self.config,
                            self.state.clone(),
                            consensus_adapter,
                            self.checkpoint_store.clone(),
                            new_epoch_store.clone(),
                            self.state_sync_handle.clone(),
                            self.randomness_handle.clone(),
                            consensus_manager,
                            consensus_store_pruner,
                            weak_hasher,
                            self.backpressure_manager.clone(),
                            soft_locks,
                            validator_server_handle,
                            validator_overload_monitor_handle,
                            consensus_queue_overload_monitor_handle,
                            soft_lock_sweep_handle,
                            checkpoint_metrics,
                            iota_tx_validator_metrics,
                            validator_registry_id,
                        )
                        .await?,
                    )
                } else {
                    info!("This node is no longer a validator after reconfiguration");
                    if self.registry_service.remove(validator_registry_id) {
                        debug!("Removed validator metrics registry");
                    } else {
                        warn!("Failed to remove validator metrics registry");
                    }
                    validator_server_handle.shutdown();
                    debug!("Validator grpc server shutdown triggered");

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

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

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

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

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

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

            // Drop the old epoch store to free its in-memory structures
            // (ConsensusOutputCache, ConsensusQuarantine, DashMaps, etc.).
            // The DB tables were already released above.
            drop(cur_epoch_store);

            // Prune old epoch databases after each epoch transition to prevent
            // accumulation of RocksDB instances during fast catch-up sync
            // (e.g. syncing from genesis).
            self.state.epoch_db_pruner().prune_old_epoch_dbs().await;

            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(),
                        iota_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;
        }

        // Shutdown the gRPC server if it's running
        if let Some(grpc_handle) = self.grpc_server_handle.lock().await.take() {
            info!("Shutting down gRPC server");
            if let Err(e) = grpc_handle.shutdown().await {
                warn!("Failed to gracefully shutdown gRPC server: {e}");
            }
        }
    }

    /// Asynchronously reconfigures the state of the authority node for the next
    /// epoch.
    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>,
    ) -> IotaResult<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 epoch_supply_change = last_checkpoint
            .end_of_epoch_data
            .as_ref()
            .ok_or_else(|| {
                IotaError::from("last checkpoint in epoch should contain end of epoch data")
            })?
            .epoch_supply_change;

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

        Ok(new_epoch_store)
    }

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

    async fn execute_transaction_immediately_at_zero_epoch(
        state: &Arc<AuthorityState>,
        epoch_store: &Arc<AuthorityPerEpochStore>,
        tx: &TransactionEnvelope,
        span: tracing::Span,
    ) {
        let _guard = span.enter();
        let transaction =
            iota_types::executable_transaction::VerifiedExecutableTransaction::new_unchecked(
                iota_types::executable_transaction::ExecutableTransaction::new_from_data_and_sig(
                    tx.data().clone(),
                    iota_types::executable_transaction::CertificateProof::Checkpoint(0, 0),
                ),
            );
        state
            .try_execute_immediately(&transaction, None, epoch_store)
            .unwrap();
    }

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

    /// Returns the registry service holding the node's Prometheus registries
    /// and their shared exposure filter.
    pub(crate) fn registry_service(&self) -> &RegistryService {
        &self.registry_service
    }

    /// Sends signed capability notification to committee validators for
    /// non-committee validators. This method implements retry logic to handle
    /// failed attempts to send the notification. It will retry sending the
    /// notification with an increasing interval until it receives a successful
    /// response from a f+1 committee members or 2f+1 non-retryable errors.
    async fn send_signed_capability_notification_to_committee_with_retry(
        &self,
        epoch_store: &Arc<AuthorityPerEpochStore>,
    ) {
        const INITIAL_RETRY_INTERVAL_SECS: u64 = 5;
        const RETRY_INTERVAL_INCREMENT_SECS: u64 = 5;
        const MAX_RETRY_INTERVAL_SECS: u64 = 300; // 5 minutes

        // Create the capability notification once
        let config = epoch_store.protocol_config();
        let binary_config = to_binary_config(config);

        // Create the capability notification
        let capabilities = AuthorityCapabilitiesV1::new(
            self.state.name,
            epoch_store.get_chain(),
            self.config
                .supported_protocol_versions
                .expect("Supported versions should be populated")
                .truncate_below(config.version),
            self.state
                .get_available_system_packages(&binary_config)
                .await,
        );

        // Sign the capabilities using the authority key pair from config
        let signature = AuthoritySignature::new_secure(
            &IntentMessage::new(
                Intent::iota_app(IntentScope::AuthorityCapabilities),
                &capabilities,
            ),
            &epoch_store.epoch(),
            self.config.authority_key_pair(),
        );

        let request = HandleCapabilityNotificationRequestV1 {
            message: SignedAuthorityCapabilitiesV1::new_from_data_and_sig(capabilities, signature),
        };

        let mut retry_interval = Duration::from_secs(INITIAL_RETRY_INTERVAL_SECS);

        loop {
            let auth_agg = self.auth_agg.load();
            match auth_agg
                .send_capability_notification_to_quorum(request.clone())
                .await
            {
                Ok(_) => {
                    info!("Successfully sent capability notification to committee");
                    break;
                }
                Err(err) => {
                    match &err {
                        AggregatorSendCapabilityNotificationError::RetryableNotification {
                            errors,
                        } => {
                            warn!(
                                "Failed to send capability notification to committee (retryable error), will retry in {:?}: {:?}",
                                retry_interval, errors
                            );
                        }
                        AggregatorSendCapabilityNotificationError::NonRetryableNotification {
                            errors,
                        } => {
                            error!(
                                "Failed to send capability notification to committee (non-retryable error): {:?}",
                                errors
                            );
                            break;
                        }
                    };

                    // Wait before retrying
                    tokio::time::sleep(retry_interval).await;

                    // Increase retry interval for the next attempt, capped at max
                    retry_interval = std::cmp::min(
                        retry_interval + Duration::from_secs(RETRY_INTERVAL_INCREMENT_SECS),
                        Duration::from_secs(MAX_RETRY_INTERVAL_SECS),
                    );
                }
            }
        }
    }
}

#[cfg(msim)]
impl IotaNode {
    pub fn get_sim_node_id(&self) -> iota_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);
    }
}

enum SpawnOnce {
    // Mutex is only needed to make SpawnOnce Sync
    Unstarted(
        Mutex<BoxFuture<'static, Result<iota_network_stack::server::Server>>>,
        tokio::runtime::Handle,
    ),
    #[allow(unused)]
    Started(iota_http::ServerHandle),
}

impl SpawnOnce {
    pub fn new(
        future: impl Future<Output = Result<iota_network_stack::server::Server>> + Send + 'static,
        serving_rt_handle: tokio::runtime::Handle,
    ) -> Self {
        Self::Unstarted(Mutex::new(Box::pin(future)), serving_rt_handle)
    }

    pub async fn start(self) -> Self {
        match self {
            Self::Unstarted(future, serving_rt_handle) => {
                // bind() and serve() must execute on the serving runtime:
                // iota_http::Builder::serve captures Handle::current() there for
                // the accept loop and every request handler.
                let (handle_tx, handle_rx) = tokio::sync::oneshot::channel();
                serving_rt_handle.spawn(async move {
                    let server = future.into_inner().await.unwrap_or_else(|err| {
                        panic!("Failed to start validator gRPC server: {err}")
                    });
                    if handle_tx.send(server.handle().clone()).is_err() {
                        return;
                    }
                    if let Err(err) = server.serve().await {
                        info!("Server stopped: {err}");
                    }
                    info!("Server stopped");
                });
                let handle = handle_rx
                    .await
                    .expect("validator gRPC server exited before returning its handle");
                Self::Started(handle)
            }
            Self::Started(_) => self,
        }
    }

    pub fn shutdown(self) {
        if let SpawnOnce::Started(handle) = self {
            handle.trigger_shutdown();
        }
    }
}

/// Notify [`DiscoveryEventLoop`] that a new list of trusted peers are now
/// available.
fn send_trusted_peer_change(
    config: &NodeConfig,
    sender: &watch::Sender<TrustedPeerChangeEvent>,
    new_epoch_start_state: &EpochStartSystemState,
) {
    let new_committee =
        new_epoch_start_state.get_validator_as_p2p_peers(config.authority_public_key());

    sender.send_modify(|event| {
        core::mem::swap(&mut event.new_committee, &mut event.old_committee);
        event.new_committee = new_committee;
    })
}

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

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

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

/// Builds and starts the gRPC server for the IOTA node based on the node's
/// configuration.
///
/// This function performs the following tasks:
/// 1. Checks if the node is a validator by inspecting the consensus
///    configuration; if so, it returns early as validators do not expose gRPC
///    APIs.
/// 2. Checks if gRPC is enabled in the configuration.
/// 3. Creates broadcast channels for checkpoint streaming.
/// 4. Initializes the gRPC checkpoint service.
/// 5. Spawns the gRPC server to listen for incoming connections.
///
/// Returns a tuple of optional broadcast channels for checkpoint summary and
/// data.
async fn build_grpc_server(
    config: &NodeConfig,
    state: Arc<AuthorityState>,
    state_sync_store: RocksDbStore,
    executor: Option<Arc<dyn iota_types::transaction_executor::TransactionExecutor>>,
    prometheus_registry: &Registry,
    server_version: ServerVersion,
) -> Result<Option<GrpcServerHandle>> {
    // Validators do not expose gRPC APIs
    if config.consensus_config().is_some() || !config.enable_grpc_api {
        return Ok(None);
    }

    let Some(grpc_config) = &config.grpc_api_config else {
        return Err(anyhow!("gRPC API is enabled but no configuration provided"));
    };

    // Get chain identifier from state directly
    let chain_id = state.get_chain_identifier();

    let grpc_read_store = Arc::new(GrpcReadStore::new(state.clone(), state_sync_store));

    // Create cancellation token for proper shutdown hierarchy
    let shutdown_token = CancellationToken::new();

    // Create GrpcReader
    let grpc_reader = Arc::new(GrpcReader::new(
        grpc_read_store,
        Some(server_version.to_string()),
    ));

    // Create gRPC server metrics
    let grpc_server_metrics = iota_grpc_server::GrpcServerMetrics::new(prometheus_registry);
    let client_id_source = config
        .policy_config
        .as_ref()
        .map(|p| p.client_id_source.clone());

    let handle = start_grpc_server(
        grpc_reader,
        executor,
        grpc_config.clone(),
        shutdown_token,
        chain_id,
        Some(grpc_server_metrics),
        state.traffic_controller.clone(),
        client_id_source,
    )
    .await?;

    Ok(Some(handle))
}

/// Builds and starts the HTTP server for the IOTA node, exposing the JSON-RPC
/// API based on the node's configuration.
///
/// This function performs the following tasks:
/// 1. Checks if the node is a validator by inspecting the consensus
///    configuration; if so, it returns early as validators do not expose these
///    APIs.
/// 2. Creates an Axum router to handle HTTP requests.
/// 3. Initializes the JSON-RPC server and registers various RPC modules based
///    on the node's state and configuration, including CoinApi,
///    TransactionBuilderApi, GovernanceApi, TransactionExecutionApi, and
///    IndexerApi.
/// 4. Binds the server to the specified JSON-RPC address and starts listening
///    for incoming connections.
pub async fn build_http_server(
    state: Arc<AuthorityState>,
    transaction_orchestrator: &Option<Arc<TransactionOrchestrator<NetworkAuthorityClient>>>,
    config: &NodeConfig,
    prometheus_registry: &Registry,
) -> Result<Option<iota_http::ServerHandle>> {
    // Validators do not expose these APIs
    if config.consensus_config().is_some() {
        return Ok(None);
    }

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

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

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

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

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

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

        let iota_names_config = config
            .iota_names_config
            .clone()
            .unwrap_or_else(|| IotaNamesConfig::from_chain(&state.get_chain_identifier().chain()));

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

        let server_type = config.jsonrpc_server_type();

        server.to_router(server_type).await?
    };

    router = router.merge(json_rpc_router);

    router = router
        .route("/health", axum::routing::get(health_check_handler))
        .route_layer(axum::Extension(state));

    let layers = ServiceBuilder::new()
        .map_request(|mut request: axum::http::Request<_>| {
            if let Some(connect_info) = request.extensions().get::<iota_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));

    router = router.layer(layers);

    let handle = iota_http::Builder::new()
        .serve(&config.json_rpc_address, router)
        .map_err(|e| anyhow::anyhow!("{e}"))?;
    info!(local_addr =? handle.local_addr(), "IOTA JSON-RPC server listening on {}", handle.local_addr());

    Ok(Some(handle))
}

#[derive(Debug, serde::Serialize, serde::Deserialize)]
pub struct Threshold {
    pub threshold_seconds: Option<u32>,
}

async fn health_check_handler(
    axum::extract::Query(Threshold { threshold_seconds }): axum::extract::Query<Threshold>,
    axum::Extension(state): axum::Extension<Arc<AuthorityState>>,
) -> impl axum::response::IntoResponse {
    if let Some(threshold_seconds) = threshold_seconds {
        // Attempt to get the latest checkpoint
        let summary = match state
            .get_checkpoint_store()
            .get_highest_executed_checkpoint()
        {
            Ok(Some(summary)) => summary,
            Ok(None) => {
                warn!("Highest executed checkpoint not found");
                return (axum::http::StatusCode::SERVICE_UNAVAILABLE, "down");
            }
            Err(err) => {
                warn!("Failed to retrieve highest executed checkpoint: {:?}", err);
                return (axum::http::StatusCode::SERVICE_UNAVAILABLE, "down");
            }
        };

        // Calculate the threshold time based on the provided threshold_seconds
        let latest_chain_time = summary.timestamp();
        let threshold =
            std::time::SystemTime::now() - Duration::from_secs(threshold_seconds as u64);

        // Check if the latest checkpoint is within the threshold
        if latest_chain_time < threshold {
            warn!(
                ?latest_chain_time,
                ?threshold,
                "failing health check due to checkpoint lag"
            );
            return (axum::http::StatusCode::SERVICE_UNAVAILABLE, "down");
        }
    }
    // if health endpoint is responding and no threshold is given, respond success
    (axum::http::StatusCode::OK, "up")
}

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

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

// Not msim: this test asserts routing across real OS worker-thread pools by
// name, which the deterministic simulator collapses onto a single thread.
#[cfg(all(test, not(msim)))]
mod runtime_split_tests {
    use std::{
        sync::mpsc,
        time::{Duration, Instant},
    };

    use anyhow::anyhow;

    use super::SpawnOnce;

    /// A single-worker-thread runtime whose worker thread carries `name`, so a
    /// task can tell which runtime it is running on via `current_pool()`.
    fn runtime(name: &'static str) -> tokio::runtime::Runtime {
        tokio::runtime::Builder::new_multi_thread()
            .worker_threads(1)
            .thread_name(name)
            .enable_all()
            .build()
            .unwrap()
    }

    /// Name of the runtime whose worker thread is executing this code.
    fn current_pool() -> String {
        std::thread::current()
            .name()
            .unwrap_or("<unnamed>")
            .to_string()
    }

    /// Regression guard for the runtime split. `SpawnOnce::start()` is invoked
    /// on the node-core runtime (as `IotaNode::start_async` does), but it
    /// must run the server *bind* on the serving runtime:
    /// `iota_http::Builder::serve` captures `Handle::current()` there for
    /// the accept loop and every request handler.
    ///
    /// The test records the runtime `start()` runs on and the runtime the bind
    /// runs on, and asserts the former is node-core and the latter is
    /// serving (so they differ). With the pre-fix inline bind the bind ran
    /// on the caller (node-core) runtime, and this test fails.
    #[test]
    fn spawn_once_binds_on_serving_not_the_core_runtime() {
        let node = runtime("node-core");
        let serving = runtime("serving");
        let (tx, rx) = mpsc::channel::<(&'static str, String)>();

        // The "bind" future records where it runs, then binds a minimal real
        // server (health service only) on an ephemeral port.
        let bind_tx = tx.clone();
        let bind_future = async move {
            let _ = bind_tx.send(("bind", current_pool()));
            let addr = "/ip4/127.0.0.1/tcp/0/http".parse().unwrap();
            let server = iota_network_stack::config::Config::new()
                .server_builder()
                .bind(&addr, None)
                .await
                .map_err(|e| anyhow!("bind failed: {e}"))?;
            Ok(server)
        };
        let once = SpawnOnce::new(bind_future, serving.handle().clone());

        // Drive start() ON the node-core runtime and record the runtime it runs on.
        let caller_tx = tx.clone();
        node.spawn(async move {
            let _ = caller_tx.send(("caller", current_pool()));
            let _ = once.start().await;
        });

        // Collect both readings.
        let (mut caller_pool, mut bind_pool) = (None, None);
        let deadline = Instant::now() + Duration::from_secs(10);
        while (caller_pool.is_none() || bind_pool.is_none()) && Instant::now() < deadline {
            if let Ok((which, pool)) = rx.recv_timeout(Duration::from_millis(200)) {
                match which {
                    "caller" => caller_pool = Some(pool),
                    "bind" => bind_pool = Some(pool),
                    _ => {}
                }
            }
        }
        let caller_pool = caller_pool.expect("start() never ran");
        let bind_pool = bind_pool.expect("bind future never ran");

        assert!(
            caller_pool.starts_with("node-core"),
            "start() should run on the node-core runtime, ran on: {caller_pool}"
        );
        assert!(
            bind_pool.starts_with("serving"),
            "server bind must run on the serving runtime, ran on: {bind_pool}"
        );
        assert_ne!(
            caller_pool, bind_pool,
            "the fix must move the bind off the caller (node-core) runtime onto serving"
        );
    }
}