iota_config/

node.rs

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

use std::{
    net::{IpAddr, Ipv4Addr, SocketAddr},
    num::{NonZeroU32, NonZeroUsize},
    path::{Path, PathBuf},
    sync::Arc,
    time::Duration,
};

use anyhow::Result;
use iota_keys::keypair_file::{read_authority_keypair_from_file, read_keypair_from_file};
use iota_metrics::MetricGroups;
use iota_multiaddr::Multiaddr;
use iota_names::config::IotaNamesConfig;
use iota_sdk_crypto::simple::SimpleKeypair;
use iota_sdk_types::Address;
use iota_types::{
    committee::EpochId,
    crypto::{
        AccountPrivateKey, AuthorityKeyPair, AuthorityPublicKeyBytes, KeypairTraits,
        NetworkPrivateKey, get_key_pair_from_rng,
    },
    messages_checkpoint::CheckpointSequenceNumber,
    supported_protocol_versions::{Chain, SupportedProtocolVersions},
    traffic_control::PolicyConfig,
};
use once_cell::sync::OnceCell;
use rand::{rand_core::UnwrapErr, rngs::SysRng};
use serde::{Deserialize, Serialize};
use starfish_config::Parameters as StarfishParameters;

use crate::{
    Config, certificate_deny_config::CertificateDenyConfig, genesis,
    migration_tx_data::MigrationTxData, object_storage_config::ObjectStoreConfig, p2p::P2pConfig,
    setting::Setting, transaction_deny_config::TransactionDenyConfig,
    verifier_signing_config::VerifierSigningConfig,
};

/// Default gas price of 1000 Nanos
pub const DEFAULT_VALIDATOR_GAS_PRICE: u64 = iota_types::transaction::DEFAULT_VALIDATOR_GAS_PRICE;

/// Default commission rate of 2%
pub const DEFAULT_COMMISSION_RATE: u64 = 200;

/// Default budget in MiB for the in-memory full-checkpoint-contents cache.
pub const DEFAULT_FULL_CHECKPOINT_CONTENTS_CACHE_SIZE_MB: usize = 1024;

#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "kebab-case")]
pub struct NodeConfig {
    /// The public key bytes corresponding to the private key that the validator
    /// holds to sign transactions.
    #[serde(default = "default_authority_key_pair")]
    pub authority_key_pair: AuthorityKeyPairWithPath,
    /// The public key bytes corresponding to the private key that the validator
    /// holds to sign consensus blocks.
    #[serde(default = "default_key_pair")]
    pub protocol_key_pair: KeyPairWithPath,
    #[serde(default = "default_key_pair")]
    pub account_key_pair: KeyPairWithPath,
    /// The public key bytes corresponding to the private key that the validator
    /// uses to establish TLS connections.
    #[serde(default = "default_key_pair")]
    pub network_key_pair: KeyPairWithPath,
    pub db_path: PathBuf,

    /// The network address for gRPC communication.
    ///
    /// Can be overwritten with args `listen-address` parameters.
    #[serde(default = "default_grpc_address")]
    pub network_address: Multiaddr,
    #[serde(default = "default_json_rpc_address")]
    pub json_rpc_address: SocketAddr,

    /// The address for Prometheus metrics.
    #[serde(default = "default_metrics_address")]
    pub metrics_address: SocketAddr,

    /// The address for the admin interface that is
    /// run in the metrics separate runtime and provides access to
    /// admin node commands such as logging and tracing options.
    #[serde(default = "default_admin_interface_address")]
    pub admin_interface_address: SocketAddr,

    /// Configuration struct for the consensus.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub consensus_config: Option<ConsensusConfig>,

    /// Flag to enable the JSON-RPC API. Default: `true`. When `false` a full
    /// node serves only the `/health` endpoint on `json_rpc_address`.
    #[serde(default = "default_enable_jsonrpc_api")]
    pub enable_jsonrpc_api: bool,

    /// Renamed to `enable_jsonrpc_api`; see [`NodeConfig::check_renamed_keys`]
    /// for which configs that still set it are refused. Never set this.
    /// Removal is tracked in <https://github.com/iotaledger/iota/issues/12885>.
    #[serde(default, skip_serializing)]
    pub enable_index_processing: Option<bool>,

    /// Flag to enable gRPC load shedding: requests over a service's
    /// concurrency limit are rejected immediately with `RESOURCE_EXHAUSTED`
    /// instead of waiting for a slot.
    #[serde(default)]
    pub grpc_load_shed: Option<bool>,

    /// Maximum number of concurrent in-flight requests per CPU core, applied
    /// to each service of the validator gRPC server separately (`Validator`,
    /// `ValidatorV2`, `ValidatorPeer`), so a flood of client transaction
    /// submissions cannot crowd validator-peer RPCs out of admission slots.
    /// The effective per-service limit is this value multiplied by the CPU
    /// cores of the machine at server startup, so the same config file scales
    /// with the hardware.
    ///
    /// Most request time is spent awaiting locks, I/O or consensus rather
    /// than on-CPU, so the default ceiling is generous: it bounds total
    /// in-flight work so a request flood cannot grow queues and memory
    /// without limit, it does not throttle normal load. Operators wanting
    /// hard load-shedding can lower it and set `grpc_load_shed`.
    ///
    /// A value of zero is rejected at config load: it would not disable the
    /// limit, it would block every request.
    #[serde(default = "default_grpc_concurrency_limit_per_core")]
    pub grpc_concurrency_limit_per_core: NonZeroUsize,

    /// Configuration struct for P2P.
    #[serde(default)]
    pub p2p_config: P2pConfig,

    /// Contains genesis location that might be `InPlace`
    /// for reading all genesis data to memory or `InFile`,
    /// and `OnceCell` pointer to a genesis struct.
    pub genesis: Genesis,

    /// Contains the path where to find the migration blob.
    pub migration_tx_data_path: Option<PathBuf>,

    /// Configuration for pruning of the authority store, to define when
    /// an old data is removed from the storage space.
    #[serde(default = "default_authority_store_pruning_config")]
    pub authority_store_pruning_config: AuthorityStorePruningConfig,

    /// Size of the broadcast channel used for notifying other systems of end of
    /// epoch.
    ///
    /// If unspecified, this will default to `128`.
    #[serde(default = "default_end_of_epoch_broadcast_channel_capacity")]
    pub end_of_epoch_broadcast_channel_capacity: usize,

    /// Configuration for the checkpoint executor for limiting
    /// the number of checkpoints to execute concurrently,
    /// and to allow for checkpoint post-processing.
    #[serde(default)]
    pub checkpoint_executor_config: CheckpointExecutorConfig,

    #[serde(skip_serializing_if = "Option::is_none")]
    pub metrics: Option<MetricsConfig>,

    /// In a `iota-node` binary, this is set to
    /// SupportedProtocolVersions::SYSTEM_DEFAULT in iota-node/src/main.rs.
    /// It is present in the config so that it can be changed by tests in
    /// order to test protocol upgrades.
    #[serde(skip)]
    pub supported_protocol_versions: Option<SupportedProtocolVersions>,

    /// Configuration for enabling/disabling expensive safety checks.
    #[serde(default)]
    pub expensive_safety_check_config: ExpensiveSafetyCheckConfig,

    /// Configuration to specify rules for denying transactions
    /// based on `objectsIDs`, `addresses`, or enable/disable many
    /// features such as publishing new packages or using shared objects.
    #[serde(default)]
    pub transaction_deny_config: TransactionDenyConfig,

    /// Config used to deny execution for certificate digests
    /// know for crashing or hanging validator nodes.
    ///
    /// Should be used for a fast temporary fixes and
    /// removed once the issue is fixed.
    #[serde(default)]
    pub certificate_deny_config: CertificateDenyConfig,

    /// Used to determine how state debug information is dumped
    /// when a node forks.
    #[serde(default)]
    pub state_debug_dump_config: StateDebugDumpConfig,

    #[serde(default)]
    pub checkpoint_archive_config: Option<CheckpointArchiveConfig>,

    /// Determines if snapshot should be uploaded to the remote storage.
    #[serde(default)]
    pub state_snapshot_write_config: StateSnapshotConfig,

    #[serde(default = "default_transaction_kv_store_config")]
    pub transaction_kv_store_read_config: TransactionKeyValueStoreReadConfig,

    // TODO: write config seem to be unused.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub transaction_kv_store_write_config: Option<TransactionKeyValueStoreWriteConfig>,

    /// Configuration for defining thresholds and settings
    /// for managing system overload conditions in a node.
    #[serde(default = "default_authority_overload_config")]
    pub authority_overload_config: AuthorityOverloadConfig,

    /// Specifies the ending epoch for a node for debugging purposes.
    ///
    ///  Ignored if set by config, can be configured only by cli arguments.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub run_with_range: Option<RunWithRange>,

    /// Traffic control policy. For killswitch use None.
    ///
    /// With the key absent the default denial-of-service protection policy
    /// applies, an explicit `null` turns traffic control off, and a value
    /// configures it.
    #[serde(
        skip_serializing_if = "is_default_traffic_controller_policy_config",
        default = "default_traffic_controller_policy_config"
    )]
    pub policy_config: Option<PolicyConfig>,

    #[serde(default)]
    pub execution_cache_config: ExecutionCacheConfig,

    /// Memory budget in MiB for the in-memory cache of full checkpoint
    /// contents, which serves the checkpoint executor's bulk transaction
    /// loads and checkpoint-contents requests from state-sync peers. When
    /// the budget is exceeded, the oldest checkpoints are evicted first.
    /// Set to 0 to disable the cache; consumers then fall back to
    /// reconstructing contents from the transaction and effects stores.
    ///
    /// The budget is accounted in serialized (BCS) bytes; the resident
    /// memory of a full cache is somewhat higher than the configured value
    /// due to in-memory representation overhead.
    #[serde(default = "default_full_checkpoint_contents_cache_size_mb")]
    pub full_checkpoint_contents_cache_size_mb: usize,

    #[serde(default = "bool_true")]
    pub enable_validator_tx_finalizer: bool,

    /// Enables the pre-consensus soft-locking mechanism used by the
    /// certificate-less (pcool) transaction flow (default: enabled).
    ///
    /// When disabled, post-consensus validation alone resolves owned-object
    /// conflicts. Has no effect unless the pcool flow is enabled.
    #[serde(default = "bool_true")]
    pub enable_soft_locking: bool,

    #[serde(default)]
    pub verifier_signing_config: VerifierSigningConfig,

    /// If a value is set, it determines if writes to DB can stall, which can
    /// halt the whole process. By default, write stall is enabled on
    /// validators but not on fullnodes.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub enable_db_write_stall: Option<bool>,

    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub iota_names_config: Option<IotaNamesConfig>,

    /// Flag to enable the gRPC API.
    #[serde(default)]
    pub enable_grpc_api: bool,
    /// Configuration of the gRPC API, read when `enable_grpc_api` is set.
    ///
    /// With the key absent the default configuration applies, an explicit
    /// `null` leaves the API unconfigured — a node with `enable-grpc-api` set
    /// then fails to start — and a value configures it.
    #[serde(
        default = "default_grpc_api_config",
        skip_serializing_if = "is_default_grpc_api_config"
    )]
    pub grpc_api_config: Option<GrpcApiConfig>,

    /// Allow overriding the chain for testing purposes. For instance, it allows
    /// you to create a test network that believes it is mainnet or testnet.
    /// Attempting to override this value on production networks will result
    /// in an error.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub chain_override_for_testing: Option<Chain>,

    /// Configuration for the validator client monitor that tracks
    /// client-observed performance metrics for validators.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub validator_client_monitor_config:
        Option<crate::validator_client_monitor_config::ValidatorClientMonitorConfig>,

    /// Limits of the validator's gRPC listener on `network_address`.
    #[serde(default, skip_serializing_if = "ListenerLimitsConfig::is_unset")]
    pub validator_grpc_listener_limits: ListenerLimitsConfig,

    /// Limits of the JSON-RPC listener on `json_rpc_address`.
    #[serde(default, skip_serializing_if = "ListenerLimitsConfig::is_unset")]
    pub json_rpc_listener_limits: ListenerLimitsConfig,

    /// Limits of the gRPC API listener on `grpc_api_config.address`.
    #[serde(default, skip_serializing_if = "ListenerLimitsConfig::is_unset")]
    pub grpc_api_listener_limits: ListenerLimitsConfig,
}

/// Limits a listener places on the clients that connect to it.
///
/// Each field names a setting of `iota_http::Config`, which documents what the
/// listener does with it. A field left out takes the listener's default, and a
/// field written as `null` turns that limit off.
#[derive(Clone, Debug, Default, PartialEq, Eq, Deserialize, Serialize)]
#[serde(rename_all = "kebab-case")]
pub struct ListenerLimitsConfig {
    /// How long an accepted connection may take to complete its TLS handshake.
    /// Can be turned off only together with `max-pending-connections`.
    #[serde(default, skip_serializing_if = "Setting::is_unset")]
    pub handshake_timeout: Setting<Duration>,

    /// How many accepted connections may be in their TLS handshake at the
    /// same time.
    #[serde(default, skip_serializing_if = "Setting::is_unset")]
    pub max_pending_connections: Setting<NonZeroUsize>,

    /// How many connections the listener may hold at once.
    #[serde(default, skip_serializing_if = "Setting::is_unset")]
    pub max_connections: Setting<NonZeroUsize>,

    /// How many connections one client address may hold at once.
    #[serde(default, skip_serializing_if = "Setting::is_unset")]
    pub max_connections_per_client_ip: Setting<NonZeroUsize>,

    /// How many leading bits of an IPv6 address identify one client.
    /// Must be between 32 and 128.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub ipv6_prefix_len: Option<u8>,

    /// How long a connection may serve no request before the listener closes
    /// it.
    #[serde(default, skip_serializing_if = "Setting::is_unset")]
    pub max_connection_idle: Setting<Duration>,

    /// How many HTTP/2 streams one connection may have open at once.
    #[serde(default, skip_serializing_if = "Setting::is_unset")]
    pub max_concurrent_streams: Setting<NonZeroU32>,

    /// How often the listener pings an HTTP/2 connection to find out whether
    /// its peer is still there.
    #[serde(default, skip_serializing_if = "Setting::is_unset")]
    pub http2_keepalive_interval: Setting<Duration>,

    /// How long an HTTP/1 connection may take to send a complete request
    /// header block.
    #[serde(default, skip_serializing_if = "Setting::is_unset")]
    pub http1_header_read_timeout: Setting<Duration>,
}

impl ListenerLimitsConfig {
    /// Whether every field was left out. Meant for `skip_serializing_if`.
    pub fn is_unset(&self) -> bool {
        *self == Self::default()
    }

    /// Rejects the limits the listener would refuse to start with. `key` is
    /// the section's key in `NodeConfig`, used in the error.
    fn validate(&self, key: &str) -> Result<()> {
        for (name, duration) in [
            ("handshake-timeout", self.handshake_timeout),
            ("max-connection-idle", self.max_connection_idle),
            ("http2-keepalive-interval", self.http2_keepalive_interval),
            ("http1-header-read-timeout", self.http1_header_read_timeout),
        ] {
            if duration == Setting::Value(Duration::ZERO) {
                anyhow::bail!(
                    "`{key}.{name}` is 0; set it greater than zero, or to `null` to turn it off"
                );
            }
        }
        // Every listener defaults both settings to on, so an absent
        // `max-pending-connections` is on too.
        if self.handshake_timeout == Setting::Disabled
            && self.max_pending_connections != Setting::Disabled
        {
            anyhow::bail!(
                "`{key}.handshake-timeout` is `null`, but `max-pending-connections` is on, and \
                 only the handshake timeout frees its slots; set a handshake timeout, or set \
                 `max-pending-connections` to `null` too"
            );
        }
        if let Some(len) = self
            .ipv6_prefix_len
            .filter(|len| !(MIN_IPV6_PREFIX_LEN..=128).contains(len))
        {
            anyhow::bail!(
                "`{key}.ipv6-prefix-len` is {len}; set it between {MIN_IPV6_PREFIX_LEN} and 128"
            );
        }
        Ok(())
    }
}

/// The shortest IPv6 prefix length a listener accepts.
const MIN_IPV6_PREFIX_LEN: u8 = 32;

#[derive(Clone, Debug, Default, serde::Deserialize, serde::Serialize)]
#[serde(rename_all = "kebab-case")]
pub struct TlsConfig {
    /// File path to a PEM formatted TLS certificate chain
    cert: String,
    /// File path to a PEM formatted TLS private key
    key: String,
}

impl TlsConfig {
    pub fn cert(&self) -> &str {
        &self.cert
    }

    pub fn key(&self) -> &str {
        &self.key
    }
}

/// Configuration for the gRPC API service
#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
#[serde(rename_all = "kebab-case")]
pub struct GrpcApiConfig {
    /// The address to bind the gRPC server to
    #[serde(default = "default_grpc_api_address")]
    pub address: SocketAddr,

    /// TLS configuration for the gRPC server.
    ///
    /// If not provided, the gRPC server will use plain TCP without TLS.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub tls: Option<TlsConfig>,

    /// Maximum message size for gRPC responses (in bytes)
    #[serde(default = "default_grpc_api_max_message_size_bytes")]
    pub max_message_size_bytes: u32,

    /// Buffer size for broadcast channels used for streaming
    #[serde(default = "default_grpc_api_broadcast_buffer_size")]
    pub broadcast_buffer_size: u32,

    /// Maximum number of concurrent subscribers to checkpoint streaming RPCs.
    /// Once the cap is reached, additional subscribe requests are rejected
    /// with `Unavailable` to protect the server from being overwhelmed by
    /// unbounded streaming clients. Values below 1 are clamped to 1 at
    /// server startup.
    #[serde(default = "default_grpc_api_max_concurrent_stream_subscribers")]
    pub max_concurrent_stream_subscribers: u32,

    /// Maximum size for Move values when rendering to JSON
    /// in bytes.
    #[serde(default = "default_grpc_api_max_json_move_value_size")]
    pub max_json_move_value_size: usize,

    /// Maximum number of transactions allowed in a single ExecuteTransactions
    /// batch request.
    #[serde(default = "default_grpc_api_max_execute_transaction_batch_size")]
    pub max_execute_transaction_batch_size: u32,

    /// Maximum number of transactions allowed in a single SimulateTransactions
    /// batch request.
    #[serde(default = "default_grpc_api_max_simulate_transaction_batch_size")]
    pub max_simulate_transaction_batch_size: u32,

    /// Maximum number of objects allowed in a single GetObjects batch request.
    #[serde(default = "default_grpc_api_max_get_objects_batch_size")]
    pub max_get_objects_batch_size: u32,

    /// Maximum number of transactions allowed in a single GetTransactions batch
    /// request.
    #[serde(default = "default_grpc_api_max_get_transactions_batch_size")]
    pub max_get_transactions_batch_size: u32,

    /// Maximum number of view function calls allowed in a single
    /// ViewFunctionCalls batch request.
    #[serde(default = "default_grpc_api_max_view_function_call_batch_size")]
    pub max_view_function_call_batch_size: u32,

    /// Maximum allowed timeout in milliseconds for waiting for checkpoint
    /// inclusion in ExecuteTransactions requests. Client-specified timeouts
    /// are clamped to this value.
    #[serde(default = "default_grpc_api_max_checkpoint_inclusion_timeout_ms")]
    pub max_checkpoint_inclusion_timeout_ms: u64,
}

fn default_grpc_api_address() -> SocketAddr {
    SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), 50051)
}

fn default_grpc_api_broadcast_buffer_size() -> u32 {
    100
}

fn default_grpc_api_max_concurrent_stream_subscribers() -> u32 {
    1024
}

fn default_grpc_api_max_message_size_bytes() -> u32 {
    128 * 1024 * 1024 // 128MB
}

fn default_grpc_api_max_json_move_value_size() -> usize {
    1024 * 1024 // 1 MB
}

fn default_grpc_api_max_execute_transaction_batch_size() -> u32 {
    20
}

fn default_grpc_api_max_simulate_transaction_batch_size() -> u32 {
    20
}

fn default_grpc_api_max_get_objects_batch_size() -> u32 {
    1000
}

fn default_grpc_api_max_get_transactions_batch_size() -> u32 {
    1000
}

fn default_grpc_api_max_view_function_call_batch_size() -> u32 {
    20
}

fn default_grpc_api_max_checkpoint_inclusion_timeout_ms() -> u64 {
    60_000 // 60 seconds
}

impl Default for GrpcApiConfig {
    fn default() -> Self {
        Self {
            address: default_grpc_api_address(),
            tls: None,
            max_message_size_bytes: default_grpc_api_max_message_size_bytes(),
            broadcast_buffer_size: default_grpc_api_broadcast_buffer_size(),
            max_concurrent_stream_subscribers: default_grpc_api_max_concurrent_stream_subscribers(),
            max_json_move_value_size: default_grpc_api_max_json_move_value_size(),
            max_execute_transaction_batch_size: default_grpc_api_max_execute_transaction_batch_size(
            ),
            max_simulate_transaction_batch_size:
                default_grpc_api_max_simulate_transaction_batch_size(),
            max_get_objects_batch_size: default_grpc_api_max_get_objects_batch_size(),
            max_get_transactions_batch_size: default_grpc_api_max_get_transactions_batch_size(),
            max_view_function_call_batch_size: default_grpc_api_max_view_function_call_batch_size(),
            max_checkpoint_inclusion_timeout_ms:
                default_grpc_api_max_checkpoint_inclusion_timeout_ms(),
        }
    }
}

impl GrpcApiConfig {
    // The default maximum uncompressed size in bytes for a message, based on
    // tonic's default.
    const GRPC_TONIC_DEFAULT_MAX_RECV_MESSAGE_SIZE: u32 = 4 * 1024 * 1024; // 4MB
    const GRPC_MIN_CLIENT_MAX_MESSAGE_SIZE_BYTES: u32 =
        Self::GRPC_TONIC_DEFAULT_MAX_RECV_MESSAGE_SIZE;

    pub fn tls_config(&self) -> Option<&TlsConfig> {
        self.tls.as_ref()
    }

    pub fn max_message_size_bytes(&self) -> u32 {
        // Ensure max message size is at least the minimum allowed
        self.max_message_size_bytes
            .max(Self::GRPC_MIN_CLIENT_MAX_MESSAGE_SIZE_BYTES)
    }

    /// Calculate the maximum size for a message that can be
    /// sent to a client, taking into account the client's max message size
    /// preference.
    pub fn max_message_size_client_bytes(&self, client_max_message_size_bytes: Option<u32>) -> u32 {
        client_max_message_size_bytes
            // if the client did not specify a max message size, use the tonic default receive
            // message size
            .unwrap_or(Self::GRPC_TONIC_DEFAULT_MAX_RECV_MESSAGE_SIZE)
            // clamp the value between the tonic default and the service max message size
            .clamp(
                Self::GRPC_MIN_CLIENT_MAX_MESSAGE_SIZE_BYTES,
                self.max_message_size_bytes(),
            )
    }
}

#[derive(Clone, Debug, Default, Deserialize, Serialize)]
#[serde(rename_all = "kebab-case")]
pub struct ExecutionCacheConfig {
    #[serde(default)]
    pub writeback_cache: WritebackCacheConfig,
}

#[derive(Clone, Debug, Default, Deserialize, Serialize)]
#[serde(rename_all = "kebab-case")]
pub struct WritebackCacheConfig {
    /// Maximum number of entries in each cache. (There are several
    /// different caches).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub max_cache_size: Option<u64>, // defaults to 100000

    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub package_cache_size: Option<u64>, // defaults to 1000

    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub object_cache_size: Option<u64>, // defaults to max_cache_size
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub marker_cache_size: Option<u64>, // defaults to object_cache_size
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub object_by_id_cache_size: Option<u64>, // defaults to object_cache_size

    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub transaction_cache_size: Option<u64>, // defaults to max_cache_size
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub executed_effect_cache_size: Option<u64>, // defaults to transaction_cache_size
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub effect_cache_size: Option<u64>, // defaults to executed_effect_cache_size

    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub events_cache_size: Option<u64>, // defaults to transaction_cache_size

    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub transaction_objects_cache_size: Option<u64>, // defaults to 1000

    /// Number of uncommitted transactions at which to pause consensus
    /// handler.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub backpressure_threshold: Option<u64>, // defaults to 100_000

    /// Number of uncommitted transactions at which to refuse new
    /// transaction submissions. Defaults to backpressure_threshold
    /// if unset.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub backpressure_threshold_for_rpc: Option<u64>, // defaults to backpressure_threshold

    /// Percentage of `backpressure_threshold` at which graduated load shedding
    /// based on writeback-cache pending transaction count begins. The
    /// locally-calculated shedding percentage increases linearly from 0% at
    /// `backpressure_threshold * backpressure_soft_limit_pct / 100` up to
    /// 100% at the `backpressure_threshold` if the cache size continues to
    /// increase. The calculated shedding percentage is broadcast to other
    /// validators for a coordinated response. Defaults to 50.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub backpressure_soft_limit_pct: Option<u32>,
}

impl WritebackCacheConfig {
    pub fn max_cache_size(&self) -> u64 {
        std::env::var("IOTA_CACHE_WRITEBACK_SIZE_MAX")
            .ok()
            .and_then(|s| s.parse().ok())
            .or(self.max_cache_size)
            .unwrap_or(100000)
    }

    pub fn package_cache_size(&self) -> u64 {
        std::env::var("IOTA_CACHE_WRITEBACK_SIZE_PACKAGE")
            .ok()
            .and_then(|s| s.parse().ok())
            .or(self.package_cache_size)
            .unwrap_or(1000)
    }

    pub fn object_cache_size(&self) -> u64 {
        std::env::var("IOTA_CACHE_WRITEBACK_SIZE_OBJECT")
            .ok()
            .and_then(|s| s.parse().ok())
            .or(self.object_cache_size)
            .unwrap_or_else(|| self.max_cache_size())
    }

    pub fn marker_cache_size(&self) -> u64 {
        std::env::var("IOTA_CACHE_WRITEBACK_SIZE_MARKER")
            .ok()
            .and_then(|s| s.parse().ok())
            .or(self.marker_cache_size)
            .unwrap_or_else(|| self.object_cache_size())
    }

    pub fn object_by_id_cache_size(&self) -> u64 {
        std::env::var("IOTA_CACHE_WRITEBACK_SIZE_OBJECT_BY_ID")
            .ok()
            .and_then(|s| s.parse().ok())
            .or(self.object_by_id_cache_size)
            .unwrap_or_else(|| self.object_cache_size())
    }

    pub fn transaction_cache_size(&self) -> u64 {
        std::env::var("IOTA_CACHE_WRITEBACK_SIZE_TRANSACTION")
            .ok()
            .and_then(|s| s.parse().ok())
            .or(self.transaction_cache_size)
            .unwrap_or_else(|| self.max_cache_size())
    }

    pub fn executed_effect_cache_size(&self) -> u64 {
        std::env::var("IOTA_CACHE_WRITEBACK_SIZE_EXECUTED_EFFECT")
            .ok()
            .and_then(|s| s.parse().ok())
            .or(self.executed_effect_cache_size)
            .unwrap_or_else(|| self.transaction_cache_size())
    }

    pub fn effect_cache_size(&self) -> u64 {
        std::env::var("IOTA_CACHE_WRITEBACK_SIZE_EFFECT")
            .ok()
            .and_then(|s| s.parse().ok())
            .or(self.effect_cache_size)
            .unwrap_or_else(|| self.executed_effect_cache_size())
    }

    pub fn events_cache_size(&self) -> u64 {
        std::env::var("IOTA_CACHE_WRITEBACK_SIZE_EVENTS")
            .ok()
            .and_then(|s| s.parse().ok())
            .or(self.events_cache_size)
            .unwrap_or_else(|| self.transaction_cache_size())
    }

    pub fn transaction_objects_cache_size(&self) -> u64 {
        std::env::var("IOTA_CACHE_WRITEBACK_SIZE_TRANSACTION_OBJECTS")
            .ok()
            .and_then(|s| s.parse().ok())
            .or(self.transaction_objects_cache_size)
            .unwrap_or(1000)
    }

    pub fn backpressure_threshold(&self) -> u64 {
        std::env::var("IOTA_CACHE_WRITEBACK_BACKPRESSURE_THRESHOLD")
            .ok()
            .and_then(|s| s.parse().ok())
            .or(self.backpressure_threshold)
            .unwrap_or(100_000)
    }

    pub fn backpressure_threshold_for_rpc(&self) -> u64 {
        std::env::var("IOTA_CACHE_WRITEBACK_BACKPRESSURE_THRESHOLD_FOR_RPC")
            .ok()
            .and_then(|s| s.parse().ok())
            .or(self.backpressure_threshold_for_rpc)
            .unwrap_or(self.backpressure_threshold())
    }

    pub fn backpressure_soft_limit_pct(&self) -> u32 {
        std::env::var("IOTA_CACHE_WRITEBACK_BACKPRESSURE_SOFT_LIMIT_PCT")
            .ok()
            .and_then(|s| s.parse().ok())
            .or(self.backpressure_soft_limit_pct)
            .unwrap_or(50)
            .min(100)
    }
}

#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "kebab-case")]
pub struct TransactionKeyValueStoreReadConfig {
    #[serde(default = "default_base_url")]
    pub base_url: String,

    #[serde(default = "default_cache_size")]
    pub cache_size: u64,
}

impl Default for TransactionKeyValueStoreReadConfig {
    fn default() -> Self {
        Self {
            base_url: default_base_url(),
            cache_size: default_cache_size(),
        }
    }
}

fn default_base_url() -> String {
    "".to_string()
}

fn default_cache_size() -> u64 {
    100_000
}

fn default_transaction_kv_store_config() -> TransactionKeyValueStoreReadConfig {
    TransactionKeyValueStoreReadConfig::default()
}

fn default_authority_store_pruning_config() -> AuthorityStorePruningConfig {
    AuthorityStorePruningConfig::default()
}

pub fn default_enable_jsonrpc_api() -> bool {
    true
}

fn default_grpc_address() -> Multiaddr {
    "/ip4/0.0.0.0/tcp/8080".parse().unwrap()
}
fn default_authority_key_pair() -> AuthorityKeyPairWithPath {
    AuthorityKeyPairWithPath::new(
        get_key_pair_from_rng::<AuthorityKeyPair, _>(&mut UnwrapErr(SysRng)).1,
    )
}

fn default_key_pair() -> KeyPairWithPath {
    KeyPairWithPath::new(AccountPrivateKey::random().into())
}

fn default_metrics_address() -> SocketAddr {
    SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), 9184)
}

pub fn default_admin_interface_address() -> SocketAddr {
    SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 1337)
}

pub fn default_json_rpc_address() -> SocketAddr {
    SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), 9000)
}

pub fn default_grpc_api_config() -> Option<GrpcApiConfig> {
    Some(GrpcApiConfig::default())
}

fn is_default_grpc_api_config(grpc_api_config: &Option<GrpcApiConfig>) -> bool {
    serializes_like(grpc_api_config, &default_grpc_api_config())
}

/// Returns whether `value` and `default` serialize to the same YAML.
///
/// Meant for `skip_serializing_if` predicates, which cannot report an error: a
/// value that fails to serialize is reported as unlike the default, so the
/// field is kept and the failure surfaces when serializing the field itself.
fn serializes_like<T: Serialize>(value: &T, default: &T) -> bool {
    match (serde_yaml::to_string(value), serde_yaml::to_string(default)) {
        (Ok(value), Ok(default)) => value == default,
        _ => false,
    }
}

pub fn default_grpc_concurrency_limit_per_core() -> NonZeroUsize {
    NonZeroUsize::new(1000).unwrap()
}

pub fn default_end_of_epoch_broadcast_channel_capacity() -> usize {
    128
}

pub fn default_full_checkpoint_contents_cache_size_mb() -> usize {
    DEFAULT_FULL_CHECKPOINT_CONTENTS_CACHE_SIZE_MB
}

pub fn bool_true() -> bool {
    true
}

impl Config for NodeConfig {}

impl NodeConfig {
    /// Fails if the config file still sets a key that has been renamed, unless
    /// the new key already does what the old value asked for; that case only
    /// logs a warning. Call this before doing any work.
    ///
    /// `enable-index-processing` is accepted on a validator, which never built
    /// the JSON-RPC index whatever its value, and as `true` while
    /// `enable-jsonrpc-api` is on. `false` on a full node is refused: it served
    /// the JSON-RPC API without its index, which `enable-jsonrpc-api` cannot
    /// express.
    pub fn check_renamed_keys(&self) -> Result<()> {
        let Some(enable_index_processing) = self.enable_index_processing else {
            return Ok(());
        };
        if self.is_validator() || (enable_index_processing && self.enable_jsonrpc_api) {
            tracing::warn!(
                "`enable-index-processing` was renamed to `enable-jsonrpc-api` and has no \
                 effect; remove it from the config"
            );
            return Ok(());
        }
        anyhow::bail!(
            "`enable-index-processing` was renamed to `enable-jsonrpc-api` (default true), which \
             turns the JSON-RPC API and its index on or off together; remove the old key and \
             set `enable-jsonrpc-api` to the value you want."
        );
    }

    pub fn authority_key_pair(&self) -> &AuthorityKeyPair {
        self.authority_key_pair.authority_keypair()
    }

    pub fn protocol_private_key(&self) -> NetworkPrivateKey {
        self.protocol_key_pair.network_private_key()
    }

    pub fn network_private_key(&self) -> NetworkPrivateKey {
        self.network_key_pair.network_private_key()
    }

    pub fn authority_public_key(&self) -> AuthorityPublicKeyBytes {
        self.authority_key_pair().public().into()
    }

    pub fn db_path(&self) -> PathBuf {
        self.db_path.join("live")
    }

    pub fn snapshot_path(&self) -> PathBuf {
        self.db_path.join("snapshot")
    }

    pub fn network_address(&self) -> &Multiaddr {
        &self.network_address
    }

    pub fn consensus_config(&self) -> Option<&ConsensusConfig> {
        self.consensus_config.as_ref()
    }

    /// Whether this node builds the RPC indexes: only a fullnode with the
    /// JSON-RPC or gRPC API enabled does.
    pub fn maintains_rpc_indexes(&self) -> bool {
        self.consensus_config().is_none() && (self.enable_jsonrpc_api || self.enable_grpc_api)
    }

    pub fn genesis(&self) -> Result<&genesis::Genesis> {
        self.genesis.genesis()
    }

    pub fn load_migration_tx_data(&self) -> Result<MigrationTxData> {
        let Some(location) = &self.migration_tx_data_path else {
            anyhow::bail!("no file location set");
        };

        // Load from file
        let migration_tx_data = MigrationTxData::load(location)?;

        // Validate migration content in order to avoid corrupted or malicious data
        migration_tx_data.validate_from_genesis(self.genesis.genesis()?)?;
        Ok(migration_tx_data)
    }

    pub fn iota_address(&self) -> Address {
        self.account_key_pair
            .keypair()
            .public_key()
            .derive_address()
    }

    pub fn checkpoint_archive_config(&self) -> Option<&CheckpointArchiveConfig> {
        self.checkpoint_archive_config.as_ref()
    }

    /// A config selects the validator role exactly when it has a consensus
    /// config.
    pub fn is_validator(&self) -> bool {
        self.consensus_config.is_some()
    }

    /// Validate the node config. The checks reject a config a node could
    /// not start with. They also reject the known cases where a node would
    /// start and ignore part of the config.
    pub fn validate(&self) -> Result<()> {
        // The settings a validator must not carry: it never starts the gRPC
        // server, and snapshot publication is a fullnode role.
        if self.is_validator() {
            if self.enable_grpc_api {
                anyhow::bail!(
                    "`enable-grpc-api` is set, but validators do not expose the gRPC API; turn \
                     it off, or move the API to a fullnode"
                );
            }
            if self
                .state_snapshot_write_config
                .object_store_config
                .is_some()
            {
                anyhow::bail!(
                    "`state-snapshot-write-config.object-store-config` is set, but snapshot \
                     upload is only supported on fullnodes; remove the setting or move the \
                     upload to a fullnode"
                );
            }
        }
        // In a file, an absent key deserializes to the default config, and a
        // config built in code carries it too. `None` here therefore means an
        // explicit `null`.
        if self.enable_grpc_api && self.grpc_api_config.is_none() {
            anyhow::bail!(
                "`enable-grpc-api` is set but `grpc-api-config` is `null`; give it a value, \
                 remove the key to take the default config, or turn off `enable-grpc-api`"
            );
        }
        // The uploader builds the store at startup, which needs a backend.
        if self
            .state_snapshot_write_config
            .object_store_config
            .as_ref()
            .is_some_and(|store| store.object_store.is_none())
        {
            anyhow::bail!(
                "`state-snapshot-write-config.object-store-config` has no `object-store`; \
                 snapshot upload needs a storage backend"
            );
        }
        // Right after an epoch boundary, peers still sync the end of the
        // previous epoch, and a validator still decides whether a transaction
        // re-proposed from it was already executed, from its checkpoints.
        if self
            .authority_store_pruning_config
            .num_epochs_to_retain_for_checkpoints()
            == Some(0)
        {
            anyhow::bail!(
                "`num-epochs-to-retain-for-checkpoints` is 0, but the previous epoch's \
                 checkpoints are still needed after an epoch boundary; set it to 1 or more"
            );
        }
        // Checked on every node: a fullnode starts the validator listener
        // only when it joins the committee, at an epoch change.
        for (key, limits) in [
            (
                "validator-grpc-listener-limits",
                &self.validator_grpc_listener_limits,
            ),
            ("json-rpc-listener-limits", &self.json_rpc_listener_limits),
            ("grpc-api-listener-limits", &self.grpc_api_listener_limits),
        ] {
            limits.validate(key)?;
        }
        Ok(())
    }
}

#[derive(Debug, Clone, Deserialize, Serialize)]
#[serde(rename_all = "kebab-case")]
pub struct ConsensusConfig {
    /// Base consensus DB path for all epochs.
    pub db_path: PathBuf,

    /// The number of epochs for which to retain the consensus DBs.
    /// Setting it to 0 will make a consensus DB getting dropped
    /// as soon as system is switched to a new epoch.
    pub db_retention_epochs: Option<u64>,

    /// Pruner will run on every epoch change but it will also check
    /// periodically on every `db_pruner_period_secs` seconds to see
    /// if there are any epoch DBs to remove.
    pub db_pruner_period_secs: Option<u64>,

    /// Hard limit on the number of pending transactions to submit to
    /// consensus, including those in submission wait. Used as the upper
    /// bound for graduated pre-consensus load shedding
    /// (`graduated_load_shedding_soft_limit_pct`) in the certificate-less
    /// (P-COOL) mode, and as the threshold for the binary
    /// cutoff in `ConsensusAdapter::check_consensus_overload()` in both
    /// certificate-less and certificate-based flows.
    ///
    /// Default to 20_000 inflight limit, assuming 20_000 txn tps * 1 sec
    /// consensus latency.
    pub max_pending_transactions: Option<usize>,

    /// When defined caps the calculated submission position to the
    /// max_submit_position.
    ///
    /// Even if the is elected to submit from a higher
    /// position than this, it will "reset" to the max_submit_position.
    pub max_submit_position: Option<usize>,

    /// The submit delay step to consensus defined in milliseconds.
    ///
    /// When provided it will override the current back off logic otherwise the
    /// default backoff logic will be applied based on consensus latency
    /// estimates.
    pub submit_delay_step_override_millis: Option<u64>,

    /// Parameters for Starfish consensus
    #[serde(skip_serializing_if = "Option::is_none", alias = "starfish_parameters")]
    pub parameters: Option<StarfishParameters>,

    /// Percentage of `max_pending_transactions` (hard limit) defining the soft
    /// limit at which graduated pre-consensus load shedding begins. When
    /// in-flight transactions are at or below the soft limit, no shedding
    /// occurs; above it, the shedding rate scales linearly from 0% to 100% at
    /// `max_pending_transactions`. Used in the certificate-less (P-COOL) mode.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub graduated_load_shedding_soft_limit_pct: Option<u32>,
}

impl ConsensusConfig {
    pub fn db_path(&self) -> &Path {
        &self.db_path
    }

    /// Returns the hard limit on the number of pending transactions to submit
    /// to consensus, including those in submission wait. Defaults to 20_000
    /// inflight limit, assuming 20_000 txn tps * 1 sec consensus latency.
    pub fn max_pending_transactions(&self) -> usize {
        self.max_pending_transactions.unwrap_or(20_000)
    }

    /// Returns the percentage of `max_pending_transactions` (hard limit)
    /// defining the soft limit at which graduated pre-consensus load
    /// shedding begins. Defaults to 50%. Used in the certificate-less
    /// (P-COOL) mode.
    pub fn graduated_load_shedding_soft_limit_pct(&self) -> u32 {
        self.graduated_load_shedding_soft_limit_pct
            .unwrap_or(50)
            .min(100)
    }

    pub fn submit_delay_step_override(&self) -> Option<Duration> {
        self.submit_delay_step_override_millis
            .map(Duration::from_millis)
    }

    pub fn db_retention_epochs(&self) -> u64 {
        self.db_retention_epochs.unwrap_or(0)
    }

    pub fn db_pruner_period(&self) -> Duration {
        // Default to 1 hour
        self.db_pruner_period_secs
            .map(Duration::from_secs)
            .unwrap_or(Duration::from_secs(3_600))
    }
}

#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "kebab-case")]
pub struct CheckpointExecutorConfig {
    /// Upper bound on the number of checkpoints that can be concurrently
    /// executed.
    ///
    /// If unspecified, this will default to `200`
    #[serde(default = "default_checkpoint_execution_max_concurrency")]
    pub checkpoint_execution_max_concurrency: usize,

    /// Number of seconds to wait for effects of a batch of transactions
    /// before logging a warning. Note that we will continue to retry
    /// indefinitely.
    ///
    /// If unspecified, this will default to `10`.
    #[serde(default = "default_local_execution_timeout_sec")]
    pub local_execution_timeout_sec: u64,

    /// Optional directory used for data ingestion pipeline.
    ///
    /// When specified, each executed checkpoint will be saved in a local
    /// directory for post-processing
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub data_ingestion_dir: Option<PathBuf>,
}

#[derive(Clone, Debug, Default, Deserialize, Serialize)]
#[serde(rename_all = "kebab-case")]
pub struct ExpensiveSafetyCheckConfig {
    /// If enabled, at epoch boundary, we will check that the storage
    /// fund balance is always identical to the sum of the storage
    /// rebate of all live objects, and that the total IOTA in the network
    /// remains the same.
    #[serde(default)]
    enable_epoch_iota_conservation_check: bool,

    /// If enabled, we will check that the total IOTA in all input objects of a
    /// tx (both the Move part and the storage rebate) matches the total IOTA
    /// in all output objects of the tx + gas fees.
    #[serde(default)]
    enable_deep_per_tx_iota_conservation_check: bool,

    /// Disable epoch IOTA conservation check even when we are running in debug
    /// mode.
    #[serde(default)]
    force_disable_epoch_iota_conservation_check: bool,

    /// If enabled, at epoch boundary, we will check that the accumulated
    /// live object state matches the end of epoch root state digest.
    #[serde(default)]
    enable_state_consistency_check: bool,

    /// Disable state consistency check even when we are running in debug mode.
    #[serde(default)]
    force_disable_state_consistency_check: bool,
}

impl ExpensiveSafetyCheckConfig {
    pub fn new_enable_all() -> Self {
        Self {
            enable_epoch_iota_conservation_check: true,
            enable_deep_per_tx_iota_conservation_check: true,
            force_disable_epoch_iota_conservation_check: false,
            enable_state_consistency_check: true,
            force_disable_state_consistency_check: false,
        }
    }

    pub fn new_disable_all() -> Self {
        Self {
            enable_epoch_iota_conservation_check: false,
            enable_deep_per_tx_iota_conservation_check: false,
            force_disable_epoch_iota_conservation_check: true,
            enable_state_consistency_check: false,
            force_disable_state_consistency_check: true,
        }
    }

    pub fn force_disable_epoch_iota_conservation_check(&mut self) {
        self.force_disable_epoch_iota_conservation_check = true;
    }

    pub fn enable_epoch_iota_conservation_check(&self) -> bool {
        (self.enable_epoch_iota_conservation_check || cfg!(debug_assertions))
            && !self.force_disable_epoch_iota_conservation_check
    }

    pub fn force_disable_state_consistency_check(&mut self) {
        self.force_disable_state_consistency_check = true;
    }

    pub fn enable_state_consistency_check(&self) -> bool {
        (self.enable_state_consistency_check || cfg!(debug_assertions))
            && !self.force_disable_state_consistency_check
    }

    pub fn enable_deep_per_tx_iota_conservation_check(&self) -> bool {
        self.enable_deep_per_tx_iota_conservation_check || cfg!(debug_assertions)
    }
}

fn default_checkpoint_execution_max_concurrency() -> usize {
    4
}

fn default_local_execution_timeout_sec() -> u64 {
    30
}

impl Default for CheckpointExecutorConfig {
    fn default() -> Self {
        Self {
            checkpoint_execution_max_concurrency: default_checkpoint_execution_max_concurrency(),
            local_execution_timeout_sec: default_local_execution_timeout_sec(),
            data_ingestion_dir: None,
        }
    }
}

#[derive(Debug, Clone, Deserialize, Serialize)]
#[serde(rename_all = "kebab-case")]
pub struct AuthorityStorePruningConfig {
    /// number of the latest epoch dbs to retain
    #[serde(default = "default_num_latest_epoch_dbs_to_retain")]
    pub num_latest_epoch_dbs_to_retain: usize,
    /// Number of historic epochs of object versions to keep, on top of the
    /// epoch the node is in. Read through [`Self::num_epochs_to_retain`].
    ///   0        — keep the epoch the node is in only (default).
    ///   N        — keep the epoch the node is in plus the N epochs before it.
    ///   u64::MAX — keep every epoch's bucket; object pruning is off.
    #[serde(default)]
    pub num_epochs_to_retain: u64,
    /// enables periodic background compaction for old SST files whose last
    /// modified time is older than `periodic_compaction_threshold_days`
    /// days. That ensures that all sst files eventually go through the
    /// compaction process
    ///
    /// With the key absent files older than a day are compacted, an explicit
    /// `null` turns periodic compaction off, and a value sets the threshold in
    /// days.
    #[serde(
        default = "default_periodic_compaction_threshold_days",
        skip_serializing_if = "is_default_periodic_compaction_threshold_days"
    )]
    pub periodic_compaction_threshold_days: Option<usize>,
    /// Number of historic epochs of transactions, effects, events, and
    /// checkpoint data to keep, on top of the epoch the node is in. Controls
    /// transaction pruning. Read through
    /// [`Self::num_epochs_to_retain_for_checkpoints`].
    ///   0        — refused by [`NodeConfig::validate`].
    ///   N        — keep the epoch the node is in plus the N epochs before it.
    ///   u64::MAX — keep every epoch's bucket; transaction pruning is off.
    ///   None     — [`DEFAULT_NUM_EPOCHS_TO_RETAIN_FOR_CHECKPOINTS`].
    #[serde(skip_serializing_if = "Option::is_none")]
    pub num_epochs_to_retain_for_checkpoints: Option<u64>,
    /// Number of historic epochs of RPC index history (the transaction and
    /// event lookups behind `iotax_queryTransactionBlocks`, `iotax_queryEvents`
    /// and their gRPC equivalents) to keep, on top of the epoch the node is
    /// in. Index tables for live objects are not affected. Read through
    /// [`Self::num_epochs_to_retain_for_indexes`].
    ///   0        — keep the epoch the node is in only.
    ///   N        — keep the epoch the node is in plus the N epochs before it.
    ///   u64::MAX — keep every epoch's bucket; index pruning is off.
    ///   None     — follow `num_epochs_to_retain_for_checkpoints`.
    ///
    /// Must not exceed `num_epochs_to_retain_for_checkpoints`; the node
    /// refuses to start otherwise.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub num_epochs_to_retain_for_indexes: Option<u64>,
}

/// The historic epochs of transactions, effects, events and checkpoint data
/// kept when `num-epochs-to-retain-for-checkpoints` is not set: about a month
/// at the networks' one-day epochs.
pub const DEFAULT_NUM_EPOCHS_TO_RETAIN_FOR_CHECKPOINTS: u64 = 30;

fn default_num_latest_epoch_dbs_to_retain() -> usize {
    3
}

fn default_periodic_compaction_threshold_days() -> Option<usize> {
    Some(1)
}

fn is_default_periodic_compaction_threshold_days(days: &Option<usize>) -> bool {
    *days == default_periodic_compaction_threshold_days()
}

impl Default for AuthorityStorePruningConfig {
    fn default() -> Self {
        Self {
            num_latest_epoch_dbs_to_retain: default_num_latest_epoch_dbs_to_retain(),
            num_epochs_to_retain: 0,
            periodic_compaction_threshold_days: default_periodic_compaction_threshold_days(),
            num_epochs_to_retain_for_checkpoints: if cfg!(msim) { Some(2) } else { None },
            num_epochs_to_retain_for_indexes: None,
        }
    }
}

impl AuthorityStorePruningConfig {
    pub fn set_num_epochs_to_retain(&mut self, num_epochs_to_retain: u64) {
        self.num_epochs_to_retain = num_epochs_to_retain;
    }

    pub fn set_num_epochs_to_retain_for_checkpoints(&mut self, num_epochs_to_retain: Option<u64>) {
        self.num_epochs_to_retain_for_checkpoints = num_epochs_to_retain;
    }

    /// Historic epochs of object versions to keep on top of the epoch the
    /// node is in, `None` when object pruning is off.
    pub fn num_epochs_to_retain(&self) -> Option<u64> {
        (self.num_epochs_to_retain != u64::MAX).then_some(self.num_epochs_to_retain)
    }

    /// Historic epochs of transactions, effects, events and checkpoint data
    /// to keep on top of the epoch the node is in, `None` when transaction
    /// pruning is off. Unset is
    /// [`DEFAULT_NUM_EPOCHS_TO_RETAIN_FOR_CHECKPOINTS`].
    pub fn num_epochs_to_retain_for_checkpoints(&self) -> Option<u64> {
        match self.num_epochs_to_retain_for_checkpoints {
            Some(u64::MAX) => None,
            Some(n) => Some(n),
            None => Some(DEFAULT_NUM_EPOCHS_TO_RETAIN_FOR_CHECKPOINTS),
        }
    }

    /// Historic epochs of RPC index history to keep on top of the epoch the
    /// node is in, `None` when index pruning is off. Unset follows the
    /// transaction retention.
    pub fn num_epochs_to_retain_for_indexes(&self) -> Option<u64> {
        match self.num_epochs_to_retain_for_indexes {
            Some(n) if n == u64::MAX => None,
            Some(n) => Some(n),
            None => self.num_epochs_to_retain_for_checkpoints(),
        }
    }

    /// Fails if the index retention exceeds the transaction retention, which
    /// would let `iotax_queryTransactionBlocks` return digests the node has
    /// already pruned.
    pub fn check_index_retention_within_ledger(&self) -> Result<()> {
        let Some(ledger) = self.num_epochs_to_retain_for_checkpoints() else {
            // An unbounded ledger outlives every index retention.
            return Ok(());
        };
        let index = self.num_epochs_to_retain_for_indexes();
        if index.is_none_or(|index| index > ledger) {
            let index = index.map_or("unlimited".to_owned(), |index| index.to_string());
            anyhow::bail!(
                "num-epochs-to-retain-for-indexes ({index}) must not exceed \
                 num-epochs-to-retain-for-checkpoints ({ledger}): a longer index retention \
                 lets iotax_queryTransactionBlocks return a transaction digest whose ledger \
                 data has already been pruned. Lower num-epochs-to-retain-for-indexes or \
                 raise num-epochs-to-retain-for-checkpoints."
            );
        }
        Ok(())
    }
}

#[derive(Debug, Clone, Deserialize, Serialize)]
#[serde(rename_all = "kebab-case")]
pub struct MetricsConfig {
    #[serde(skip_serializing_if = "Option::is_none")]
    pub push_interval_seconds: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub push_url: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub groups: Option<MetricGroups>,
}

fn default_checkpoint_archive_download_concurrency() -> NonZeroUsize {
    NonZeroUsize::new(10).unwrap()
}

fn default_checkpoint_archive_verify_concurrency() -> NonZeroUsize {
    std::thread::available_parallelism().unwrap_or(NonZeroUsize::new(4).unwrap())
}

/// Configuration for backfilling checkpoint contents from the
/// checkpoint archive when peers no longer serve the required range.
#[derive(Debug, Clone, Deserialize, Serialize)]
#[serde(rename_all = "kebab-case")]
pub struct CheckpointArchiveConfig {
    /// URL of the checkpoint archive to backfill from.
    pub url: String,
    /// Non-zero number of checkpoints to download in parallel.
    #[serde(default = "default_checkpoint_archive_download_concurrency")]
    pub download_concurrency: NonZeroUsize,
    /// Non-zero number of downloaded checkpoints to verify in parallel.
    /// Defaults to the number of CPU cores.
    #[serde(default = "default_checkpoint_archive_verify_concurrency")]
    pub verify_concurrency: NonZeroUsize,
}

/// Configuration for the per-epoch state-snapshot publisher.
///
/// **Operator note (V2 snapshot publishing).** A node configured to publish
/// V2 snapshots must have a perpetual store containing no pre-V2
/// (`StoreObjectV1`) rows. This means a snapshot-publishing node must have
/// either synced from genesis under V2 or been restored from a V2 snapshot.
/// There is no on-disk backfill: a fresh sync is the only supported path.
#[derive(Default, Debug, Clone, Deserialize, Serialize)]
#[serde(rename_all = "kebab-case")]
pub struct StateSnapshotConfig {
    #[serde(skip_serializing_if = "Option::is_none")]
    pub object_store_config: Option<ObjectStoreConfig>,
    pub concurrency: usize,
}

#[derive(Default, Debug, Clone, Deserialize, Serialize)]
#[serde(rename_all = "kebab-case")]
pub struct TransactionKeyValueStoreWriteConfig {
    pub aws_access_key_id: String,
    pub aws_secret_access_key: String,
    pub aws_region: String,
    pub table_name: String,
    pub bucket_name: String,
    pub concurrency: usize,
}

/// Configuration for the threshold(s) at which we consider the system
/// to be overloaded. When one of the threshold is passed, the node may
/// stop processing new transactions and/or certificates until the congestion
/// resolves.
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "kebab-case")]
pub struct AuthorityOverloadConfig {
    /// Maximum time a transaction can wait in the transaction manager execution
    /// queue before it triggers an overload detection on the object it depends
    /// on.
    #[serde(default = "default_max_txn_age_in_queue")]
    pub max_txn_age_in_queue: Duration,

    /// The interval of checking overload signal.
    #[serde(default = "default_overload_monitor_interval")]
    pub overload_monitor_interval: Duration,

    /// The execution queueing latency when entering load shedding mode.
    #[serde(default = "default_execution_queue_latency_soft_limit")]
    pub execution_queue_latency_soft_limit: Duration,

    /// The execution queueing latency when entering aggressive load shedding
    /// mode.
    #[serde(default = "default_execution_queue_latency_hard_limit")]
    pub execution_queue_latency_hard_limit: Duration,

    /// The maximum percentage of transactions to shed in load shedding mode.
    #[serde(default = "default_max_load_shedding_percentage")]
    pub max_load_shedding_percentage: u32,

    /// When in aggressive load shedding mode, the minimum percentage of
    /// transactions to shed.
    #[serde(default = "default_min_load_shedding_percentage_above_hard_limit")]
    pub min_load_shedding_percentage_above_hard_limit: u32,

    /// If transaction ready rate is below this rate, we consider the validator
    /// is well under used, and will not enter load shedding mode.
    #[serde(default = "default_safe_transaction_ready_rate")]
    pub safe_transaction_ready_rate: u32,

    /// When set to true, transaction signing may be rejected when the validator
    /// is overloaded.
    #[serde(default = "default_check_system_overload_at_signing")]
    pub check_system_overload_at_signing: bool,

    /// When set to true, transaction execution may be rejected when the
    /// validator is overloaded.
    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
    pub check_system_overload_at_execution: bool,

    /// Reject a transaction if transaction manager queue length is above this
    /// threshold. 100_000 = 10k TPS * 5s resident time in transaction
    /// manager (pending + executing) * 2.
    #[serde(default = "default_max_transaction_manager_queue_length")]
    pub max_transaction_manager_queue_length: usize,

    /// Reject a transaction if the number of pending transactions depending on
    /// the object is above the threshold.
    #[serde(default = "default_max_transaction_manager_per_object_queue_length")]
    pub max_transaction_manager_per_object_queue_length: usize,

    /// Percentage of `max_transaction_manager_queue_length` at which graduated
    /// load shedding begins in the certificate-less (P-COOL) mode. Read
    /// via the same-named accessor, which clamps the value to <=100.
    #[serde(default = "default_max_transaction_manager_queue_length_soft_limit_pct")]
    pub max_transaction_manager_queue_length_soft_limit_pct: u32,
}

impl AuthorityOverloadConfig {
    /// Returns the soft-limit percentage, clamped to <=100 to guard against
    /// out-of-range operator-supplied values.
    pub fn max_transaction_manager_queue_length_soft_limit_pct(&self) -> u32 {
        self.max_transaction_manager_queue_length_soft_limit_pct
            .min(100)
    }
}

fn default_max_txn_age_in_queue() -> Duration {
    Duration::from_millis(500)
}

fn default_overload_monitor_interval() -> Duration {
    Duration::from_secs(10)
}

fn default_execution_queue_latency_soft_limit() -> Duration {
    Duration::from_secs(1)
}

fn default_execution_queue_latency_hard_limit() -> Duration {
    Duration::from_secs(10)
}

fn default_max_load_shedding_percentage() -> u32 {
    95
}

fn default_min_load_shedding_percentage_above_hard_limit() -> u32 {
    50
}

fn default_safe_transaction_ready_rate() -> u32 {
    100
}

fn default_check_system_overload_at_signing() -> bool {
    true
}

fn default_max_transaction_manager_queue_length() -> usize {
    100_000
}

fn default_max_transaction_manager_queue_length_soft_limit_pct() -> u32 {
    50
}

fn default_max_transaction_manager_per_object_queue_length() -> usize {
    20
}

impl Default for AuthorityOverloadConfig {
    fn default() -> Self {
        Self {
            max_txn_age_in_queue: default_max_txn_age_in_queue(),
            overload_monitor_interval: default_overload_monitor_interval(),
            execution_queue_latency_soft_limit: default_execution_queue_latency_soft_limit(),
            execution_queue_latency_hard_limit: default_execution_queue_latency_hard_limit(),
            max_load_shedding_percentage: default_max_load_shedding_percentage(),
            min_load_shedding_percentage_above_hard_limit:
                default_min_load_shedding_percentage_above_hard_limit(),
            safe_transaction_ready_rate: default_safe_transaction_ready_rate(),
            check_system_overload_at_signing: true,
            check_system_overload_at_execution: false,
            max_transaction_manager_queue_length: default_max_transaction_manager_queue_length(),
            max_transaction_manager_queue_length_soft_limit_pct:
                default_max_transaction_manager_queue_length_soft_limit_pct(),
            max_transaction_manager_per_object_queue_length:
                default_max_transaction_manager_per_object_queue_length(),
        }
    }
}

fn default_authority_overload_config() -> AuthorityOverloadConfig {
    AuthorityOverloadConfig::default()
}

fn default_traffic_controller_policy_config() -> Option<PolicyConfig> {
    Some(PolicyConfig::default_dos_protection_policy())
}

fn is_default_traffic_controller_policy_config(policy_config: &Option<PolicyConfig>) -> bool {
    serializes_like(policy_config, &default_traffic_controller_policy_config())
}

#[derive(Debug, Clone, PartialEq, Deserialize, Serialize, Eq)]
pub struct Genesis {
    #[serde(flatten)]
    location: Option<GenesisLocation>,

    #[serde(skip)]
    genesis: once_cell::sync::OnceCell<genesis::Genesis>,
}

impl Genesis {
    pub fn new(genesis: genesis::Genesis) -> Self {
        Self {
            location: Some(GenesisLocation::InPlace {
                genesis: Box::new(genesis),
            }),
            genesis: Default::default(),
        }
    }

    pub fn new_from_file<P: Into<PathBuf>>(path: P) -> Self {
        Self {
            location: Some(GenesisLocation::File {
                genesis_file_location: path.into(),
            }),
            genesis: Default::default(),
        }
    }

    pub fn new_empty() -> Self {
        Self {
            location: None,
            genesis: Default::default(),
        }
    }

    pub fn genesis(&self) -> Result<&genesis::Genesis> {
        match &self.location {
            Some(GenesisLocation::InPlace { genesis }) => Ok(genesis),
            Some(GenesisLocation::File {
                genesis_file_location,
            }) => self
                .genesis
                .get_or_try_init(|| genesis::Genesis::load(genesis_file_location)),
            None => anyhow::bail!("no genesis location set"),
        }
    }
}

#[derive(Debug, Clone, PartialEq, Deserialize, Serialize, Eq)]
#[serde(untagged)]
enum GenesisLocation {
    InPlace {
        genesis: Box<genesis::Genesis>,
    },
    File {
        #[serde(rename = "genesis-file-location")]
        genesis_file_location: PathBuf,
    },
}

/// Wrapper struct for SimpleKeypair that can be deserialized from a file path.
/// Used by network, worker, and account keypair.
#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct KeyPairWithPath {
    #[serde(flatten)]
    location: KeyPairLocation,

    #[serde(skip)]
    keypair: OnceCell<Arc<SimpleKeypair>>,
}

impl PartialEq for KeyPairWithPath {
    fn eq(&self, other: &Self) -> bool {
        self.location == other.location
    }
}

impl Eq for KeyPairWithPath {}

#[derive(Debug, Clone, Deserialize, Serialize)]
#[serde(untagged)]
enum KeyPairLocation {
    InPlace {
        #[serde(with = "bech32_formatted_keypair")]
        value: Arc<SimpleKeypair>,
    },
    File {
        path: PathBuf,
    },
}

impl PartialEq for KeyPairLocation {
    fn eq(&self, other: &Self) -> bool {
        match (self, other) {
            (Self::InPlace { value: a }, Self::InPlace { value: b }) => {
                a.to_bytes() == b.to_bytes()
            }
            (Self::File { path: a }, Self::File { path: b }) => a == b,
            _ => false,
        }
    }
}

impl Eq for KeyPairLocation {}

impl KeyPairWithPath {
    pub fn new(kp: SimpleKeypair) -> Self {
        let cell: OnceCell<Arc<SimpleKeypair>> = OnceCell::new();
        let arc_kp = Arc::new(kp);
        // OK to unwrap panic because authority should not start without all keypairs
        // loaded.
        cell.set(arc_kp.clone()).expect("failed to set keypair");
        Self {
            location: KeyPairLocation::InPlace { value: arc_kp },
            keypair: cell,
        }
    }

    pub fn new_from_path(path: PathBuf) -> Self {
        let cell: OnceCell<Arc<SimpleKeypair>> = OnceCell::new();
        // OK to unwrap panic because authority should not start without all keypairs
        // loaded.
        cell.set(Arc::new(read_keypair_from_file(&path).unwrap_or_else(
            |e| panic!("invalid keypair file at path {path:?}: {e}"),
        )))
        .expect("failed to set keypair");
        Self {
            location: KeyPairLocation::File { path },
            keypair: cell,
        }
    }

    pub fn keypair(&self) -> &SimpleKeypair {
        self.keypair
            .get_or_init(|| match &self.location {
                KeyPairLocation::InPlace { value } => value.clone(),
                KeyPairLocation::File { path } => {
                    // OK to unwrap panic because authority should not start without all keypairs
                    // loaded.
                    Arc::new(
                        read_keypair_from_file(path).unwrap_or_else(|e| {
                            panic!("invalid keypair file at path {path:?}: {e}")
                        }),
                    )
                }
            })
            .as_ref()
    }

    /// The stored keypair as an ed25519 network private key, for the network
    /// stacks that consume that type directly. Panics if the stored keypair
    /// is not ed25519.
    pub fn network_private_key(&self) -> NetworkPrivateKey {
        self.keypair()
            .as_opt_ed25519()
            .cloned()
            .expect("only Ed25519 network keys are allowed")
    }
}

/// Wrapper struct for AuthorityKeyPair that can be deserialized from a file
/// path.
#[derive(Clone, Debug, PartialEq, Eq, Deserialize, Serialize)]
pub struct AuthorityKeyPairWithPath {
    #[serde(flatten)]
    location: AuthorityKeyPairLocation,

    #[serde(skip)]
    keypair: OnceCell<Arc<AuthorityKeyPair>>,
}

#[derive(Debug, Clone, PartialEq, Deserialize, Serialize, Eq)]
#[serde(untagged)]
enum AuthorityKeyPairLocation {
    InPlace { value: Arc<AuthorityKeyPair> },
    File { path: PathBuf },
}

impl AuthorityKeyPairWithPath {
    pub fn new(kp: AuthorityKeyPair) -> Self {
        let cell: OnceCell<Arc<AuthorityKeyPair>> = OnceCell::new();
        let arc_kp = Arc::new(kp);
        // OK to unwrap panic because authority should not start without all keypairs
        // loaded.
        cell.set(arc_kp.clone())
            .expect("failed to set authority keypair");
        Self {
            location: AuthorityKeyPairLocation::InPlace { value: arc_kp },
            keypair: cell,
        }
    }

    pub fn new_from_path(path: PathBuf) -> Self {
        let cell: OnceCell<Arc<AuthorityKeyPair>> = OnceCell::new();
        // OK to unwrap panic because authority should not start without all keypairs
        // loaded.
        cell.set(Arc::new(
            read_authority_keypair_from_file(&path)
                .unwrap_or_else(|_| panic!("invalid authority keypair file at path {path:?}")),
        ))
        .expect("failed to set authority keypair");
        Self {
            location: AuthorityKeyPairLocation::File { path },
            keypair: cell,
        }
    }

    pub fn authority_keypair(&self) -> &AuthorityKeyPair {
        self.keypair
            .get_or_init(|| match &self.location {
                AuthorityKeyPairLocation::InPlace { value } => value.clone(),
                AuthorityKeyPairLocation::File { path } => {
                    // OK to unwrap panic because authority should not start without all keypairs
                    // loaded.
                    Arc::new(
                        read_authority_keypair_from_file(path)
                            .unwrap_or_else(|_| panic!("invalid authority keypair file {path:?}")),
                    )
                }
            })
            .as_ref()
    }
}

/// Configurations which determine how we dump state debug info.
/// Debug info is dumped when a node forks.
#[derive(Clone, Debug, Deserialize, Serialize, Default)]
#[serde(rename_all = "kebab-case")]
pub struct StateDebugDumpConfig {
    #[serde(skip_serializing_if = "Option::is_none")]
    pub dump_file_directory: Option<PathBuf>,
}

#[cfg(test)]
mod tests {
    use std::{
        num::{NonZeroU32, NonZeroUsize},
        path::PathBuf,
        time::Duration,
    };

    use fastcrypto::traits::KeyPair;
    use iota_keys::keypair_file::{write_authority_keypair_to_file, write_keypair_to_file};
    use iota_sdk_crypto::simple::SimpleKeypair;
    use iota_types::{
        crypto::{AuthorityKeyPair, NetworkPrivateKey, get_key_pair_from_rng},
        traffic_control::PolicyConfig,
    };
    use rand::{SeedableRng, rngs::StdRng};
    use serde::Serialize;
    use serde_yaml::Value;

    use super::{
        AuthorityStorePruningConfig, DEFAULT_NUM_EPOCHS_TO_RETAIN_FOR_CHECKPOINTS, Genesis,
        GrpcApiConfig, ListenerLimitsConfig, ObjectStoreConfig, default_grpc_api_config,
        default_periodic_compaction_threshold_days, default_traffic_controller_policy_config,
    };
    use crate::{
        ConsensusConfig, NodeConfig, object_storage_config::ObjectStoreType, setting::Setting,
    };

    const TEMPLATE: &str = include_str!("../data/fullnode-template.yaml");

    const POLICY_CONFIG: &[&str] = &["policy-config"];
    const GRPC_API_CONFIG: &[&str] = &["grpc-api-config"];
    const COMPACTION_THRESHOLD: &[&str] = &[
        "authority-store-pruning-config",
        "periodic-compaction-threshold-days",
    ];

    fn template_config() -> NodeConfig {
        serde_yaml::from_str(TEMPLATE).unwrap()
    }

    fn consensus_config() -> super::ConsensusConfig {
        serde_yaml::from_str("db-path: /opt/iota/consensus-db").unwrap()
    }

    fn object_store_config() -> ObjectStoreConfig {
        ObjectStoreConfig {
            object_store: Some(ObjectStoreType::File),
            directory: Some(PathBuf::from("/opt/iota/snapshots")),
            ..Default::default()
        }
    }

    fn round_trip(config: &NodeConfig) -> NodeConfig {
        serde_yaml::from_str(&serde_yaml::to_string(config).unwrap()).unwrap()
    }

    fn as_yaml<T: Serialize>(value: &T) -> String {
        serde_yaml::to_string(value).unwrap()
    }

    /// Reads `path` out of a serialized value, returning `None` when the last
    /// key is absent.
    fn written_at(value: &Value, path: &[&str]) -> Option<Value> {
        let (last, parents) = path.split_last().unwrap();
        let mut current = value;
        for name in parents {
            current = current
                .as_mapping()
                .unwrap()
                .get(&Value::String((*name).to_owned()))
                .unwrap();
        }
        current
            .as_mapping()
            .unwrap()
            .get(&Value::String((*last).to_owned()))
            .cloned()
    }

    #[test]
    fn a_retention_below_two_is_no_longer_rounded_up() {
        let config = AuthorityStorePruningConfig {
            num_epochs_to_retain_for_checkpoints: Some(1),
            ..Default::default()
        };
        assert_eq!(config.num_epochs_to_retain_for_checkpoints(), Some(1));
    }

    #[test]
    fn index_retention_within_ledger_retention_is_accepted() {
        let config = AuthorityStorePruningConfig {
            num_epochs_to_retain_for_checkpoints: Some(2),
            num_epochs_to_retain_for_indexes: Some(2),
            ..Default::default()
        };
        assert!(config.check_index_retention_within_ledger().is_ok());
    }

    #[test]
    fn index_retention_beyond_ledger_retention_is_refused() {
        let config = AuthorityStorePruningConfig {
            num_epochs_to_retain_for_checkpoints: Some(2),
            num_epochs_to_retain_for_indexes: Some(3),
            ..Default::default()
        };
        let err = config
            .check_index_retention_within_ledger()
            .unwrap_err()
            .to_string();
        assert!(err.contains("num-epochs-to-retain-for-indexes"), "{err}");
        assert!(
            err.contains("num-epochs-to-retain-for-checkpoints"),
            "{err}"
        );
        assert!(err.contains('3'), "{err}");
        assert!(err.contains('2'), "{err}");
    }

    #[test]
    fn the_narrowest_ledger_retention_refuses_a_longer_index_retention() {
        let config = AuthorityStorePruningConfig {
            num_epochs_to_retain_for_checkpoints: Some(1),
            num_epochs_to_retain_for_indexes: Some(2),
            ..Default::default()
        };
        assert!(config.check_index_retention_within_ledger().is_err());
    }

    #[test]
    fn equal_narrowest_retentions_are_accepted() {
        let config = AuthorityStorePruningConfig {
            num_epochs_to_retain_for_checkpoints: Some(1),
            num_epochs_to_retain_for_indexes: Some(1),
            ..Default::default()
        };
        assert!(config.check_index_retention_within_ledger().is_ok());
    }

    #[test]
    fn unbounded_ledger_retention_accepts_any_index_retention() {
        let config = AuthorityStorePruningConfig {
            num_epochs_to_retain_for_checkpoints: Some(u64::MAX),
            num_epochs_to_retain_for_indexes: Some(1000),
            ..Default::default()
        };
        assert!(config.check_index_retention_within_ledger().is_ok());
    }

    #[test]
    fn unset_ledger_retention_keeps_the_default_and_the_index_follows_it() {
        let config = AuthorityStorePruningConfig {
            num_epochs_to_retain_for_checkpoints: None,
            num_epochs_to_retain_for_indexes: None,
            ..Default::default()
        };
        assert_eq!(
            config.num_epochs_to_retain_for_checkpoints(),
            Some(DEFAULT_NUM_EPOCHS_TO_RETAIN_FOR_CHECKPOINTS)
        );
        assert_eq!(
            config.num_epochs_to_retain_for_indexes(),
            Some(DEFAULT_NUM_EPOCHS_TO_RETAIN_FOR_CHECKPOINTS)
        );
        assert!(config.check_index_retention_within_ledger().is_ok());
    }

    #[test]
    fn unset_index_retention_follows_the_ledger() {
        let config = AuthorityStorePruningConfig {
            num_epochs_to_retain_for_checkpoints: Some(3),
            num_epochs_to_retain_for_indexes: None,
            ..Default::default()
        };
        assert_eq!(config.num_epochs_to_retain_for_indexes(), Some(3));
        assert!(config.check_index_retention_within_ledger().is_ok());
    }

    #[test]
    fn index_pruning_turned_off_against_a_bounded_ledger_is_refused() {
        // `u64::MAX` is index pruning off, which outlives any bounded ledger.
        let config = AuthorityStorePruningConfig {
            num_epochs_to_retain_for_checkpoints: Some(2),
            num_epochs_to_retain_for_indexes: Some(u64::MAX),
            ..Default::default()
        };
        let err = config
            .check_index_retention_within_ledger()
            .unwrap_err()
            .to_string();
        assert!(err.contains("unlimited"), "{err}");
    }

    #[test]
    fn max_retention_means_pruning_is_off_for_every_knob() {
        let config = AuthorityStorePruningConfig {
            num_epochs_to_retain: u64::MAX,
            num_epochs_to_retain_for_checkpoints: Some(u64::MAX),
            num_epochs_to_retain_for_indexes: Some(u64::MAX),
            ..Default::default()
        };
        assert_eq!(config.num_epochs_to_retain(), None);
        assert_eq!(config.num_epochs_to_retain_for_checkpoints(), None);
        assert_eq!(config.num_epochs_to_retain_for_indexes(), None);
    }

    #[test]
    fn a_fullnode_serving_an_api_builds_the_indexes() {
        const TEMPLATE: &str = include_str!("../data/fullnode-template.yaml");

        let config: NodeConfig = serde_yaml::from_str(TEMPLATE).unwrap();
        assert!(config.maintains_rpc_indexes());
    }

    #[test]
    fn a_fullnode_with_both_apis_off_builds_no_indexes() {
        const TEMPLATE: &str = include_str!("../data/fullnode-template.yaml");

        let mut config: NodeConfig = serde_yaml::from_str(TEMPLATE).unwrap();
        config.enable_jsonrpc_api = false;
        config.enable_grpc_api = false;
        assert!(!config.maintains_rpc_indexes());
    }

    #[test]
    fn a_validator_builds_no_indexes_even_with_an_api_enabled() {
        const TEMPLATE: &str = include_str!("../data/fullnode-template.yaml");

        let mut config: NodeConfig = serde_yaml::from_str(TEMPLATE).unwrap();
        config.enable_jsonrpc_api = true;
        config.consensus_config = Some(ConsensusConfig {
            db_path: PathBuf::from("/tmp/consensus"),
            db_retention_epochs: None,
            db_pruner_period_secs: None,
            max_pending_transactions: None,
            max_submit_position: None,
            submit_delay_step_override_millis: None,
            parameters: None,
            graduated_load_shedding_soft_limit_pct: None,
        });
        assert!(!config.maintains_rpc_indexes());
    }

    #[test]
    fn serialize_genesis_from_file() {
        let g = Genesis::new_from_file("path/to/file");

        let s = serde_yaml::to_string(&g).unwrap();
        assert_eq!("---\ngenesis-file-location: path/to/file\n", s);
        let loaded_genesis: Genesis = serde_yaml::from_str(&s).unwrap();
        assert_eq!(g, loaded_genesis);
    }

    #[test]
    fn fullnode_template() {
        const TEMPLATE: &str = include_str!("../data/fullnode-template.yaml");

        let _template: NodeConfig = serde_yaml::from_str(TEMPLATE).unwrap();
    }

    #[test]
    fn validate_requires_a_grpc_config_when_the_api_is_enabled() {
        // An absent key gives the default config, so only an explicit `null`
        // reaches this rule.
        let mut config = template_config();
        config.grpc_api_config = None;
        config.validate().unwrap();

        config.enable_grpc_api = true;
        let err = config.validate().unwrap_err().to_string();
        assert!(err.contains("`grpc-api-config` is `null`"), "{err}");

        config.grpc_api_config = Some(GrpcApiConfig::default());
        config.validate().unwrap();

        // Validators do not expose the gRPC API, so the config check does
        // not apply to them and the API itself is rejected instead.
        config.grpc_api_config = None;
        config.consensus_config = Some(consensus_config());
        let err = config.validate().unwrap_err().to_string();
        assert!(err.contains("validators do not expose"), "{err}");
    }

    #[test]
    fn validate_rejects_the_grpc_api_on_a_validator() {
        let mut config = template_config();
        config.consensus_config = Some(consensus_config());
        config.validate().unwrap();

        config.enable_grpc_api = true;
        let err = config.validate().unwrap_err().to_string();
        assert!(err.contains("validators do not expose"), "{err}");

        // The same config is valid for a fullnode.
        config.consensus_config = None;
        config.validate().unwrap();
    }

    #[test]
    fn validate_rejects_snapshot_upload_on_a_validator() {
        let mut config = template_config();
        config.state_snapshot_write_config.object_store_config = Some(object_store_config());
        config.validate().unwrap();

        config.consensus_config = Some(consensus_config());
        let err = config.validate().unwrap_err().to_string();
        assert!(err.contains("snapshot upload"), "{err}");
    }

    #[test]
    fn validate_rejects_a_snapshot_store_without_a_backend() {
        let mut config = template_config();
        config.state_snapshot_write_config.object_store_config = Some(ObjectStoreConfig::default());

        let err = config.validate().unwrap_err().to_string();
        assert!(err.contains("storage backend"), "{err}");
    }

    #[test]
    fn validate_rejects_a_zero_checkpoint_retention() {
        let mut config = template_config();
        config
            .authority_store_pruning_config
            .set_num_epochs_to_retain_for_checkpoints(Some(0));
        let err = config.validate().unwrap_err().to_string();
        assert!(
            err.contains("num-epochs-to-retain-for-checkpoints"),
            "{err}"
        );

        config
            .authority_store_pruning_config
            .set_num_epochs_to_retain_for_checkpoints(Some(1));
        config.validate().unwrap();
    }

    /// The keys of the listeners' limits sections in a node config.
    const LISTENER_LIMITS_KEYS: [&str; 3] = [
        "validator-grpc-listener-limits",
        "json-rpc-listener-limits",
        "grpc-api-listener-limits",
    ];

    /// The template config with `limits`, a YAML mapping, under `key`.
    fn config_with_listener_limits(key: &str, limits: &str) -> NodeConfig {
        serde_yaml::from_str(&format!("{TEMPLATE}\n{key}: {limits}\n")).unwrap()
    }

    /// The pending-connection limit is on unless turned off, and only the
    /// handshake timeout frees its slots, so the timeout can be turned off
    /// only together with it. The template is a fullnode's, so this also
    /// covers the validator listener it starts only once it joins the
    /// committee.
    #[test]
    fn validate_rejects_a_handshake_timeout_turned_off_alone() {
        for key in LISTENER_LIMITS_KEYS {
            for limits in [
                "{handshake-timeout: null}",
                "{handshake-timeout: null, max-pending-connections: 1}",
            ] {
                let err = config_with_listener_limits(key, limits)
                    .validate()
                    .unwrap_err()
                    .to_string();
                assert!(
                    err.contains(&format!("`{key}.handshake-timeout` is `null`")),
                    "{limits}: {err}"
                );
            }
            config_with_listener_limits(
                key,
                "{handshake-timeout: null, max-pending-connections: null}",
            )
            .validate()
            .unwrap();
        }
    }

    /// A zero deadline is reached at once and a zero interval never pauses,
    /// so the listener refuses either.
    #[test]
    fn validate_rejects_a_zero_listener_duration() {
        for key in LISTENER_LIMITS_KEYS {
            for name in [
                "handshake-timeout",
                "max-connection-idle",
                "http2-keepalive-interval",
                "http1-header-read-timeout",
            ] {
                let err =
                    config_with_listener_limits(key, &format!("{{{name}: {{secs: 0, nanos: 0}}}}"))
                        .validate()
                        .unwrap_err()
                        .to_string();
                assert!(err.contains(&format!("`{key}.{name}` is 0")), "{err}");

                config_with_listener_limits(key, &format!("{{{name}: {{secs: 1, nanos: 0}}}}"))
                    .validate()
                    .unwrap();
            }
        }
    }

    /// The listener accepts IPv6 prefix lengths from 32 to 128 bits.
    #[test]
    fn validate_rejects_an_ipv6_prefix_len_out_of_range() {
        for key in LISTENER_LIMITS_KEYS {
            for len in [0, 31, 129] {
                let err = config_with_listener_limits(key, &format!("{{ipv6-prefix-len: {len}}}"))
                    .validate()
                    .unwrap_err()
                    .to_string();
                assert!(
                    err.contains(&format!("`{key}.ipv6-prefix-len` is {len}")),
                    "{err}"
                );
            }
            for len in [32, 128] {
                config_with_listener_limits(key, &format!("{{ipv6-prefix-len: {len}}}"))
                    .validate()
                    .unwrap();
            }
        }
    }

    #[test]
    fn enable_soft_locking_defaults_to_enabled() {
        // The template omits `enable-soft-locking`, so this exercises the serde
        // default and pins the documented "default: enabled" contract.
        const TEMPLATE: &str = include_str!("../data/fullnode-template.yaml");

        let config: NodeConfig = serde_yaml::from_str(TEMPLATE).unwrap();
        assert!(config.enable_soft_locking);
    }

    /// The old key is refused where `enable-jsonrpc-api` cannot be assumed to
    /// do what its value asked for, and accepted elsewhere, so that the configs
    /// earlier releases wrote out keep starting.
    #[test]
    fn renamed_enable_index_processing_key() {
        const TEMPLATE: &str = include_str!("../data/fullnode-template.yaml");
        let with_old_key = |value: bool| {
            let mut template: serde_yaml::Value = serde_yaml::from_str(TEMPLATE).unwrap();
            template
                .as_mapping_mut()
                .unwrap()
                .insert("enable-index-processing".into(), value.into());
            serde_yaml::from_value::<NodeConfig>(template).unwrap()
        };

        // `false` on a full node served the JSON-RPC API without its index.
        let mut config = with_old_key(false);
        assert_eq!(config.enable_index_processing, Some(false));
        assert!(config.enable_jsonrpc_api);
        let err = config.check_renamed_keys().unwrap_err().to_string();
        assert!(err.contains("enable-index-processing"), "{err}");
        assert!(err.contains("enable-jsonrpc-api"), "{err}");

        // A validator never built the index, whatever the value.
        config.consensus_config = Some(consensus_config());
        assert!(config.check_renamed_keys().is_ok());

        // `true` asks for what `enable-jsonrpc-api` does by default.
        let mut config = with_old_key(true);
        assert!(config.check_renamed_keys().is_ok());
        config.enable_jsonrpc_api = false;
        assert!(config.check_renamed_keys().is_err());

        // Never serialized, so rewriting the config cannot bring the key back.
        let serialized = serde_yaml::to_string(&config).unwrap();
        assert!(
            !serialized.contains("enable-index-processing"),
            "{serialized}"
        );

        let config: NodeConfig = serde_yaml::from_str(TEMPLATE).unwrap();
        assert!(config.check_renamed_keys().is_ok());
    }

    #[test]
    fn load_key_pairs_to_node_config() {
        let authority_key_pair: AuthorityKeyPair =
            get_key_pair_from_rng(&mut StdRng::from_seed([0; 32])).1;
        let protocol_private_key: NetworkPrivateKey =
            get_key_pair_from_rng(&mut StdRng::from_seed([0; 32])).1;
        let network_private_key: NetworkPrivateKey =
            get_key_pair_from_rng(&mut StdRng::from_seed([0; 32])).1;

        write_authority_keypair_to_file(&authority_key_pair, PathBuf::from("authority.key"))
            .unwrap();
        write_keypair_to_file(
            &SimpleKeypair::from(protocol_private_key.clone()),
            PathBuf::from("protocol.key"),
        )
        .unwrap();
        write_keypair_to_file(
            &SimpleKeypair::from(network_private_key.clone()),
            PathBuf::from("network.key"),
        )
        .unwrap();

        const TEMPLATE: &str = include_str!("../data/fullnode-template-with-path.yaml");
        let template: NodeConfig = serde_yaml::from_str(TEMPLATE).unwrap();
        assert_eq!(
            template.authority_key_pair().public(),
            authority_key_pair.public()
        );
        assert_eq!(
            template.network_private_key().public_key(),
            network_private_key.public_key()
        );
        assert_eq!(
            template.protocol_private_key().public_key(),
            protocol_private_key.public_key()
        );
    }

    #[test]
    fn a_policy_config_survives_a_round_trip_in_all_three_states() {
        let mut config = template_config();

        config.policy_config = None;
        assert!(round_trip(&config).policy_config.is_none());

        config.policy_config = default_traffic_controller_policy_config();
        assert_eq!(
            as_yaml(&round_trip(&config).policy_config),
            as_yaml(&default_traffic_controller_policy_config())
        );

        let configured = PolicyConfig {
            dry_run: !PolicyConfig::default_dos_protection_policy().dry_run,
            ..PolicyConfig::default_dos_protection_policy()
        };
        config.policy_config = Some(configured.clone());
        assert_eq!(
            as_yaml(&round_trip(&config).policy_config),
            as_yaml(&Some(configured))
        );
    }

    #[test]
    fn a_grpc_api_config_survives_a_round_trip_in_all_three_states() {
        let mut config = template_config();

        config.grpc_api_config = None;
        assert!(round_trip(&config).grpc_api_config.is_none());

        config.grpc_api_config = default_grpc_api_config();
        assert_eq!(
            as_yaml(&round_trip(&config).grpc_api_config),
            as_yaml(&default_grpc_api_config())
        );

        let configured = GrpcApiConfig {
            max_message_size_bytes: 1234,
            ..GrpcApiConfig::default()
        };
        config.grpc_api_config = Some(configured.clone());
        assert_eq!(
            as_yaml(&round_trip(&config).grpc_api_config),
            as_yaml(&Some(configured))
        );
    }

    #[test]
    fn the_default_pruning_config_agrees_with_the_serde_default() {
        assert_eq!(
            super::AuthorityStorePruningConfig::default().periodic_compaction_threshold_days,
            default_periodic_compaction_threshold_days()
        );
    }

    #[test]
    fn a_compaction_threshold_survives_a_round_trip_in_all_three_states() {
        let mut config = template_config();

        for state in [None, default_periodic_compaction_threshold_days(), Some(7)] {
            config
                .authority_store_pruning_config
                .periodic_compaction_threshold_days = state;
            assert_eq!(
                round_trip(&config)
                    .authority_store_pruning_config
                    .periodic_compaction_threshold_days,
                state
            );
        }
    }

    #[test]
    fn a_default_value_is_omitted_and_a_disabled_one_is_written_as_null() {
        let mut config = template_config();
        config.policy_config = default_traffic_controller_policy_config();
        config.grpc_api_config = default_grpc_api_config();
        config
            .authority_store_pruning_config
            .periodic_compaction_threshold_days = default_periodic_compaction_threshold_days();

        let written = serde_yaml::to_value(&config).unwrap();
        assert_eq!(written_at(&written, POLICY_CONFIG), None);
        assert_eq!(written_at(&written, GRPC_API_CONFIG), None);
        assert_eq!(written_at(&written, COMPACTION_THRESHOLD), None);

        config.policy_config = None;
        config.grpc_api_config = None;
        config
            .authority_store_pruning_config
            .periodic_compaction_threshold_days = None;

        let written = serde_yaml::to_value(&config).unwrap();
        assert_eq!(written_at(&written, POLICY_CONFIG), Some(Value::Null));
        assert_eq!(written_at(&written, GRPC_API_CONFIG), Some(Value::Null));
        assert_eq!(
            written_at(&written, COMPACTION_THRESHOLD),
            Some(Value::Null)
        );
    }

    fn limits(yaml: &str) -> ListenerLimitsConfig {
        serde_yaml::from_str(yaml).unwrap()
    }

    #[test]
    fn a_node_config_without_listener_limits_leaves_every_limit_unset() {
        let config = template_config();
        assert_eq!(
            config.validator_grpc_listener_limits,
            ListenerLimitsConfig::default()
        );
        assert_eq!(
            config.json_rpc_listener_limits,
            ListenerLimitsConfig::default()
        );
        assert_eq!(
            config.grpc_api_listener_limits,
            ListenerLimitsConfig::default()
        );
    }

    #[test]
    fn a_listener_limit_left_out_is_unset_and_null_disables_it() {
        assert_eq!(
            limits("max-connections: 30\nmax-connection-idle: null"),
            ListenerLimitsConfig {
                max_connections: Setting::Value(NonZeroUsize::new(30).unwrap()),
                max_connection_idle: Setting::Disabled,
                ..Default::default()
            }
        );
    }

    /// Every value is distinct, so a key read into the wrong field fails here.
    #[test]
    fn every_listener_limit_is_read_from_its_own_key() {
        let read = limits(
            "
            handshake-timeout: {secs: 1, nanos: 0}
            max-pending-connections: 2
            max-connections: 3
            max-connections-per-client-ip: 4
            ipv6-prefix-len: 48
            max-connection-idle: {secs: 5, nanos: 0}
            max-concurrent-streams: 6
            http2-keepalive-interval: {secs: 7, nanos: 0}
            http1-header-read-timeout: {secs: 8, nanos: 0}
            ",
        );
        assert_eq!(
            read,
            ListenerLimitsConfig {
                handshake_timeout: Setting::Value(Duration::from_secs(1)),
                max_pending_connections: Setting::Value(NonZeroUsize::new(2).unwrap()),
                max_connections: Setting::Value(NonZeroUsize::new(3).unwrap()),
                max_connections_per_client_ip: Setting::Value(NonZeroUsize::new(4).unwrap()),
                ipv6_prefix_len: Some(48),
                max_connection_idle: Setting::Value(Duration::from_secs(5)),
                max_concurrent_streams: Setting::Value(NonZeroU32::new(6).unwrap()),
                http2_keepalive_interval: Setting::Value(Duration::from_secs(7)),
                http1_header_read_timeout: Setting::Value(Duration::from_secs(8)),
            }
        );
    }

    #[test]
    fn every_listener_limit_can_be_disabled() {
        let read = limits(
            "
            handshake-timeout: null
            max-pending-connections: null
            max-connections: null
            max-connections-per-client-ip: null
            ipv6-prefix-len: null
            max-connection-idle: null
            max-concurrent-streams: null
            http2-keepalive-interval: null
            http1-header-read-timeout: null
            ",
        );
        assert_eq!(
            read,
            ListenerLimitsConfig {
                handshake_timeout: Setting::Disabled,
                max_pending_connections: Setting::Disabled,
                max_connections: Setting::Disabled,
                max_connections_per_client_ip: Setting::Disabled,
                ipv6_prefix_len: None,
                max_connection_idle: Setting::Disabled,
                max_concurrent_streams: Setting::Disabled,
                http2_keepalive_interval: Setting::Disabled,
                http1_header_read_timeout: Setting::Disabled,
            }
        );
    }

    #[test]
    fn listener_limits_are_written_only_when_set() {
        let mut config = template_config();
        let written = serde_yaml::to_value(&config).unwrap();
        for key in [
            "validator-grpc-listener-limits",
            "json-rpc-listener-limits",
            "grpc-api-listener-limits",
        ] {
            assert_eq!(written_at(&written, &[key]), None, "{key} was written");
        }

        config.json_rpc_listener_limits.max_connections = Setting::Value(NonZeroUsize::MIN);
        config.grpc_api_listener_limits.max_connection_idle = Setting::Disabled;
        let read_back = round_trip(&config);
        assert_eq!(
            read_back.json_rpc_listener_limits,
            config.json_rpc_listener_limits
        );
        assert_eq!(
            read_back.grpc_api_listener_limits,
            config.grpc_api_listener_limits
        );
        assert!(read_back.validator_grpc_listener_limits.is_unset());
    }
}

// RunWithRange is used to specify the ending epoch/checkpoint to process.
// this is intended for use with disaster recovery debugging and verification
// workflows, never in normal operations
#[derive(Clone, Copy, PartialEq, Debug, Serialize, Deserialize)]
pub enum RunWithRange {
    Epoch(EpochId),
    Checkpoint(CheckpointSequenceNumber),
}

impl RunWithRange {
    // is epoch_id > RunWithRange::Epoch
    pub fn is_epoch_gt(&self, epoch_id: EpochId) -> bool {
        matches!(self, RunWithRange::Epoch(e) if epoch_id > *e)
    }

    pub fn matches_checkpoint(&self, seq_num: CheckpointSequenceNumber) -> bool {
        matches!(self, RunWithRange::Checkpoint(seq) if *seq == seq_num)
    }

    pub fn into_checkpoint_bound(self) -> Option<CheckpointSequenceNumber> {
        match self {
            RunWithRange::Epoch(_) => None,
            RunWithRange::Checkpoint(seq) => Some(seq),
        }
    }
}

/// A serde helper module used with #[serde(with = "...")] to change the
/// de/serialization format of an `SimpleKeypair` to Bech32 when written to or
/// read from a node config.
mod bech32_formatted_keypair {
    use std::ops::Deref;

    use fastcrypto::encoding::{Base64, Encoding};
    use iota_sdk_crypto::{ToFromBech32, simple::SimpleKeypair};
    use serde::{Deserialize, Deserializer, Serializer};

    pub fn serialize<S, T>(kp: &T, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
        T: Deref<Target = SimpleKeypair>,
    {
        use serde::ser::Error;

        // Serialize the keypair to a Bech32 string
        let s = kp.to_bech32().map_err(Error::custom)?;

        serializer.serialize_str(&s)
    }

    pub fn deserialize<'de, D, T>(deserializer: D) -> Result<T, D::Error>
    where
        D: Deserializer<'de>,
        T: From<SimpleKeypair>,
    {
        use serde::de::Error;

        let s = String::deserialize(deserializer)?;

        // Try to deserialize the keypair from a Bech32 formatted string
        SimpleKeypair::from_bech32(&s)
            .map_err(Error::custom)
            .or_else(|_: D::Error| {
                // For backwards compatibility try Base64 if Bech32 failed
                let bytes = Base64::decode(&s).map_err(Error::custom)?;
                SimpleKeypair::from_bytes(&bytes).map_err(Error::custom)
            })
            .map(Into::into)
    }
}