// Copyright 2025 PingCAP, Inc.

use std::{error::Error, sync::Arc, time::Duration};

use online_config::{ConfigChange, OnlineConfig};
use serde_derive::{Deserialize, Serialize};
use tikv_util::{
    config::{ReadableDuration, ReadableSize},
    info,
    sys::SysQuota,
    warn,
};

use crate::{ResourceEvent, ResourcePublisher, SeverityThreshold, limiter::MAX_WAIT_TIME};

#[derive(Clone, Serialize, Deserialize, PartialEq, Debug, OnlineConfig)]
#[serde(default)]
#[serde(rename_all = "kebab-case")]
pub struct CpuThrottleSettings {
    pub enabled: bool,
    pub keyspace_cpu_quotas: String,
    pub tenant_max_cpu_ratio: f64,
    pub tenant_cap_idle_ttl: ReadableDuration,
    pub estimated_cpu_per_request_us: u64,
    pub low_priority_dag_global_enabled: bool,
    pub low_priority_dag_global_ratio: f64,
    pub keyspace_estimated_cpu_per_request_us: String,
    pub enable_adaptive_estimated_cpu_per_request_us: bool,
    /// Token refill cadence. This controls how frequently new CPU budget
    /// becomes available, independently of bucket capacity.
    pub refill_interval: ReadableDuration,
    /// Maximum duration of CPU tokens retained by a bucket. The 100ms default
    /// preserves the effective bucket capacity from before refill cadence and
    /// capacity were decoupled; increasing it permits proportionally larger
    /// bursts.
    pub burst_window: ReadableDuration,
    /// Enable dynamic refill-rate adjustment for the legacy admission path.
    /// The keyspace-fair global CPU gate disables this to avoid applying a
    /// second feedback controller to the same global bucket.
    pub enable_dynamic_adjustment: bool,
    pub high_watermark: f64,
    pub low_watermark: f64,
    /// Enable proportional burst-share enforcement in the legacy throttle
    /// path. This is independent of keyspace fairness in the read pool.
    pub enable_fair_allocation: bool,
    pub fair_allocation_threshold: f64,
    pub enable_burst: bool,
    pub burst_threshold: f64,
    pub keyspace_burst_enabled: String,
    pub enable_runtime_token_management: bool,
    pub runtime_check_interval_us: u64,
    pub additional_allocation_threshold: f64,
    pub per_allocation_us: u64,
    pub enforce_global_limit: bool,
    pub enable_pd_service_limit: bool,
    pub pd_service_limit_fetch_interval: ReadableDuration,
    pub pd_service_limit_request_timeout: ReadableDuration,
    pub pd_service_limit_ttl: ReadableDuration,
    pub pd_service_limit_consecutive_404: u64,
    pub pd_service_limit_update_threshold: f64,
    pub pd_override_config: bool,
}

impl Default for CpuThrottleSettings {
    fn default() -> CpuThrottleSettings {
        CpuThrottleSettings {
            enabled: false,
            keyspace_cpu_quotas: String::new(),
            tenant_max_cpu_ratio: 0.0,
            tenant_cap_idle_ttl: ReadableDuration::minutes(10),
            estimated_cpu_per_request_us: 1000,
            low_priority_dag_global_enabled: false,
            low_priority_dag_global_ratio: 0.6,
            keyspace_estimated_cpu_per_request_us: String::new(),
            enable_adaptive_estimated_cpu_per_request_us: true,
            refill_interval: ReadableDuration::millis(10),
            burst_window: ReadableDuration::millis(100),
            enable_dynamic_adjustment: false,
            high_watermark: 0.8,
            low_watermark: 0.5,
            enable_fair_allocation: false,
            fair_allocation_threshold: 0.7,
            enable_burst: false,
            burst_threshold: 0.9,
            keyspace_burst_enabled: String::new(),
            enable_runtime_token_management: true,
            runtime_check_interval_us: 1000,
            additional_allocation_threshold: 0.5,
            per_allocation_us: 2000,
            enforce_global_limit: false,
            enable_pd_service_limit: true,
            pd_service_limit_fetch_interval: ReadableDuration::secs(30),
            pd_service_limit_request_timeout: ReadableDuration::secs(5),
            pd_service_limit_ttl: ReadableDuration::minutes(10),
            pd_service_limit_consecutive_404: 3,
            pd_service_limit_update_threshold: 0.01,
            pd_override_config: true,
        }
    }
}

#[derive(Clone, Serialize, Deserialize, PartialEq, Debug, OnlineConfig)]
#[serde(default)]
#[serde(rename_all = "kebab-case")]
pub struct Config {
    pub enabled: bool,
    pub dry_run: bool,
    pub debug: bool,
    pub window_size: ReadableDuration,
    pub stats_interval: ReadableDuration,
    pub limiter_timeout: ReadableDuration,
    pub limiter_stats_interval: ReadableDuration,
    pub smoothing_factor: f64,
    pub active_quota_ratio: f64,
    pub min_quota_ratio: f64,
    pub severity_stressed_factor: f64,
    pub severity_critical_factor: f64,
    pub severity_exhausted_factor: f64,
    pub severity_threshold_stressed: f64,
    pub severity_threshold_critical: f64,
    pub severity_threshold_exhausted: f64,
    pub whitelist_keyspace_ids: String,
    pub ignore_keyspace_ids: String,
    /// Enables the legacy adaptive per-keyspace QPS/bytes scan limiter.
    ///
    /// This switch controls a separate path from the newer CPU-token throttle
    /// and the read-pool keyspace-fairness scheduler. Managed deployments must
    /// enable it only for Starter and keep it disabled for RC-enabled tiers.
    pub read_enabled: bool,
    /// Maximum CPU ratio for read operations.
    ///
    /// The legacy read limiter can update this ratio online while keyspace
    /// fairness is disabled, matching the pre-fairness behavior. With
    /// keyspace fairness enabled, the ratio sizes the startup global token
    /// bucket and online changes are deferred until restart. The keyspace-fair
    /// read pool has a separate online override at
    /// `readpool.unified.keyspace-fairness.max-read-cpu-ratio`.
    pub max_read_cpu_ratio: f64,
    pub max_read_bytes_factor: f64,
    pub max_read_bytes_per_core_per_second: ReadableSize,
    /// Online maximum wait for the legacy scan limiter. Startup intentionally
    /// retains the historical [`crate::limiter::MAX_WAIT_TIME`] value.
    pub max_read_wait_time: ReadableDuration,
    // transfer leader
    pub transfer_leader_enabled: bool,
    pub max_transfer_leader_per_second: u64,
    /// Enables adaptive QPS/bytes limit calculation after `read_enabled`
    /// activates the legacy scan-limiter path.
    pub enable_qps_bytes_limiter: bool,
    #[online_config(submodule)]
    pub cpu_throttle: CpuThrottleSettings,
}

impl Default for Config {
    fn default() -> Config {
        let severity_threshold = SeverityThreshold::default();
        Config {
            enabled: false,
            dry_run: false,
            debug: false,
            window_size: ReadableDuration::minutes(1),
            stats_interval: ReadableDuration::secs(1),
            limiter_timeout: ReadableDuration::secs(2),
            limiter_stats_interval: ReadableDuration::secs(1),
            smoothing_factor: 0.9,
            active_quota_ratio: 0.8,
            min_quota_ratio: 0.01,
            severity_stressed_factor: 1.0,
            severity_critical_factor: 1.0,
            severity_exhausted_factor: 1.0,
            severity_threshold_stressed: severity_threshold.stressed,
            severity_threshold_critical: severity_threshold.critical,
            severity_threshold_exhausted: severity_threshold.exhausted,
            whitelist_keyspace_ids: "[]".to_string(),
            ignore_keyspace_ids: "[]".to_string(),
            read_enabled: false,
            max_read_cpu_ratio: 0.6,
            max_read_bytes_factor: 1.0,
            max_read_bytes_per_core_per_second: ReadableSize::mb(64),
            max_read_wait_time: ReadableDuration::millis(MAX_WAIT_TIME.as_millis() as u64),
            transfer_leader_enabled: false,
            max_transfer_leader_per_second: 64,
            enable_qps_bytes_limiter: true,
            cpu_throttle: CpuThrottleSettings::default(),
        }
    }
}

impl Config {
    /// Startup global CPU budget in microseconds per refill interval.
    ///
    /// This is the token-space budget the global bucket is built from. It is
    /// intentionally derived from `max_read_cpu_ratio` only: the keyspace-fair
    /// read-pool override never resizes the bucket, it scales each charge.
    pub fn global_cpu_us_per_interval(&self, refill_interval_ms: u64) -> u64 {
        Self::global_cpu_us_per_interval_for(self.max_read_cpu_ratio, refill_interval_ms)
    }

    /// Shared formula used by both startup and online paths so they cannot
    /// drift apart.
    pub fn global_cpu_us_per_interval_for(ratio: f64, refill_interval_ms: u64) -> u64 {
        let cpu_cores = SysQuota::cpu_cores_quota().max(1.0);
        (cpu_cores * ratio * refill_interval_ms as f64 * 1000.0) as u64
    }

    /// Check whether the configuration is legal.
    pub fn validate(&self) -> Result<(), Box<dyn Error>> {
        if self.read_enabled && self.cpu_throttle.enabled {
            warn!(
                "legacy read limiter and CPU throttle are both enabled; requests may be throttled by both mechanisms"
            );
        }
        if self.window_size.0 < Duration::from_millis(1) {
            return Err("window-size cannot be less than 1 milliseconds".into());
        }
        if self.stats_interval.0 < Duration::from_millis(1) {
            return Err("stats-interval cannot be less than 1 milliseconds".into());
        }
        if self.limiter_stats_interval.0 < Duration::from_millis(1) {
            return Err("limiter-stats-interval cannot be less than 1 milliseconds".into());
        }
        // These fields only configure the Starter scan limiter. Keep them
        // dormant when that path is disabled so malformed legacy values do
        // not prevent an RC-only configuration from starting or updating.
        if self.read_enabled {
            if self.limiter_timeout.0 < Duration::from_millis(1) {
                return Err("limiter-timeout cannot be less than 1 milliseconds".into());
            }
            if self.smoothing_factor <= 0.0 || self.smoothing_factor >= 1.0 {
                return Err("smoothing-factor must be a float in the range (0.0, 1.0).".into());
            }
            if self.active_quota_ratio <= 0.0 || self.active_quota_ratio >= 1.0 {
                return Err("active-quota-ratio must be a float in the range (0.0, 1.0).".into());
            }
            if self.min_quota_ratio <= 0.0 || self.min_quota_ratio >= 1.0 {
                return Err("min-quota-ratio must be a float in the range (0.0, 1.0).".into());
            }
            if self.max_read_wait_time.0 < Duration::from_millis(1) {
                return Err("max-read-wait-time cannot be less than 1 milliseconds".into());
            }
        }
        if self.severity_threshold_stressed <= 0.0
            || self.severity_threshold_stressed >= self.severity_threshold_critical
        {
            return Err("severity-threshold-stressed must be a float in the range (0.0, severity-threshold-critical).".into());
        }
        if self.severity_threshold_critical <= self.severity_threshold_stressed
            || self.severity_threshold_critical >= self.severity_threshold_exhausted
        {
            return Err("severity-threshold-critical must be a float in the range (severity-threshold-stressed, severity-threshold-exhausted).".into());
        }
        if self.severity_threshold_exhausted <= self.severity_threshold_critical
            || self.severity_threshold_exhausted >= 1.0
        {
            return Err("severity-threshold-exhausted must be a float in the range (severity-threshold-critical, 1.0).".into());
        }
        if self.max_read_cpu_ratio <= 0.0 || self.max_read_cpu_ratio >= 1.0 {
            return Err("max-read-cpu-ratio must be a float in the range (0.0, 1.0).".into());
        }
        if self.max_read_bytes_per_core_per_second.0 == 0 {
            return Err("max-read-bytes-per-core-per-second cannot be 0".into());
        }
        if self.max_transfer_leader_per_second == 0 {
            return Err("max-transfer-leader-per-second cannot be 0".into());
        }
        // CPU throttle validation
        if self.cpu_throttle.enabled {
            if self.cpu_throttle.refill_interval.0 < Duration::from_millis(1) {
                return Err(
                    "cpu-throttle.refill-interval cannot be less than 1 milliseconds".into(),
                );
            }
            if self.cpu_throttle.burst_window.0 < self.cpu_throttle.refill_interval.0 {
                return Err(
                    "cpu-throttle.burst-window cannot be shorter than refill-interval".into(),
                );
            }
            if self.cpu_throttle.high_watermark <= self.cpu_throttle.low_watermark {
                return Err(
                    "cpu-throttle.high-watermark must be greater than cpu-throttle.low-watermark"
                        .into(),
                );
            }
            if self.cpu_throttle.high_watermark > 1.0 || self.cpu_throttle.low_watermark < 0.0 {
                return Err("cpu-throttle watermarks must be in range [0.0, 1.0]".into());
            }
            if self.cpu_throttle.fair_allocation_threshold > 1.0
                || self.cpu_throttle.fair_allocation_threshold < 0.0
            {
                return Err(
                    "cpu-throttle.fair-allocation-threshold must be in range [0.0, 1.0]".into(),
                );
            }
            if self.cpu_throttle.burst_threshold > 1.0 || self.cpu_throttle.burst_threshold < 0.0 {
                return Err("cpu-throttle.burst-threshold must be in range [0.0, 1.0]".into());
            }
            if self.cpu_throttle.estimated_cpu_per_request_us == 0 {
                return Err("cpu-throttle.estimated-cpu-per-request-us cannot be 0".into());
            }
            if !(0.0..=1.0).contains(&self.cpu_throttle.tenant_max_cpu_ratio) {
                return Err("cpu-throttle.tenant-max-cpu-ratio must be in range [0.0, 1.0]".into());
            }
            if self.cpu_throttle.tenant_cap_idle_ttl.0 < Duration::from_millis(1) {
                return Err(
                    "cpu-throttle.tenant-cap-idle-ttl cannot be less than 1 milliseconds".into(),
                );
            }
            if self.cpu_throttle.tenant_max_cpu_ratio > 0.0 {
                let tenant_cap_quota_us =
                    crate::CpuThrottleConfig::compute_tenant_cap_quota_us_for(
                        self.cpu_throttle.tenant_max_cpu_ratio,
                        self.cpu_throttle.refill_interval.as_millis(),
                    )
                    .unwrap_or(0);
                if tenant_cap_quota_us == 0 {
                    return Err(
                        "cpu-throttle.tenant-max-cpu-ratio is too small, resulting in a 0us quota"
                            .into(),
                    );
                }
                if !self.cpu_throttle.enforce_global_limit {
                    return Err(
                        "cpu-throttle.enforce-global-limit must be true when tenant-max-cpu-ratio is enabled"
                            .into(),
                    );
                }
                if !self.cpu_throttle.keyspace_cpu_quotas.is_empty() {
                    return Err(
                        "cpu-throttle.keyspace-cpu-quotas must be empty when tenant-max-cpu-ratio is enabled"
                            .into(),
                    );
                }
                if self.cpu_throttle.enable_pd_service_limit {
                    return Err(
                        "cpu-throttle.enable-pd-service-limit must be false when tenant-max-cpu-ratio is enabled"
                            .into(),
                    );
                }
                if self.cpu_throttle.enable_fair_allocation {
                    return Err(
                        "cpu-throttle.enable-fair-allocation must be false when tenant-max-cpu-ratio is enabled"
                            .into(),
                    );
                }
                if self.cpu_throttle.enable_dynamic_adjustment {
                    return Err(
                        "cpu-throttle.enable-dynamic-adjustment must be false when tenant-max-cpu-ratio is enabled"
                            .into(),
                    );
                }
                if self.cpu_throttle.enable_burst {
                    return Err(
                        "cpu-throttle.enable-burst must be false when tenant-max-cpu-ratio is enabled"
                            .into(),
                    );
                }
                if self.cpu_throttle.low_priority_dag_global_enabled {
                    return Err(
                        "cpu-throttle.low-priority-dag-global-enabled must be false when tenant-max-cpu-ratio is enabled"
                            .into(),
                    );
                }
            }
            if !(0.0..=1.0).contains(&self.cpu_throttle.low_priority_dag_global_ratio) {
                return Err(
                    "cpu-throttle.low-priority-dag-global-ratio must be in range [0.0, 1.0]".into(),
                );
            }
            if self.cpu_throttle.enable_pd_service_limit {
                if self.cpu_throttle.pd_service_limit_fetch_interval.0 < Duration::from_millis(1) {
                    return Err(
                        "cpu-throttle.pd-service-limit-fetch-interval cannot be less than 1 milliseconds"
                            .into(),
                    );
                }
                if self.cpu_throttle.pd_service_limit_request_timeout.0 < Duration::from_millis(1) {
                    return Err(
                        "cpu-throttle.pd-service-limit-request-timeout cannot be less than 1 milliseconds"
                            .into(),
                    );
                }
                if self.cpu_throttle.pd_service_limit_ttl.0 < Duration::from_millis(1) {
                    return Err(
                        "cpu-throttle.pd-service-limit-ttl cannot be less than 1 milliseconds"
                            .into(),
                    );
                }
                if self.cpu_throttle.pd_service_limit_consecutive_404 == 0 {
                    return Err("cpu-throttle.pd-service-limit-consecutive-404 cannot be 0".into());
                }
                if !(0.0..=1.0).contains(&self.cpu_throttle.pd_service_limit_update_threshold) {
                    return Err(
                        "cpu-throttle.pd-service-limit-update-threshold must be in range [0.0, 1.0]"
                            .into(),
                    );
                }
            }
        }
        Ok(())
    }

    pub fn get_read_enabled(&self) -> bool {
        self.enabled && self.read_enabled
    }

    pub fn get_transfer_leader_enabled(&self) -> bool {
        self.enabled && self.transfer_leader_enabled
    }

    pub fn get_cpu_throttle_enabled(&self) -> bool {
        self.enabled && self.cpu_throttle.enabled
    }

    pub fn to_cpu_throttle_config(&self) -> crate::CpuThrottleConfig {
        crate::CpuThrottleConfig {
            enabled: self.get_cpu_throttle_enabled(),
            keyspace_cpu_quotas: self.cpu_throttle.keyspace_cpu_quotas.clone(),
            tenant_max_cpu_ratio: self.cpu_throttle.tenant_max_cpu_ratio,
            tenant_cap_idle_ttl_ms: self.cpu_throttle.tenant_cap_idle_ttl.as_millis(),
            estimated_cpu_per_request_us: self.cpu_throttle.estimated_cpu_per_request_us,
            low_priority_dag_global_enabled: self.cpu_throttle.low_priority_dag_global_enabled,
            low_priority_dag_global_ratio: self.cpu_throttle.low_priority_dag_global_ratio,
            keyspace_estimated_cpu_per_request_us: self
                .cpu_throttle
                .keyspace_estimated_cpu_per_request_us
                .clone(),
            enable_adaptive_estimated_cpu_per_request_us: self
                .cpu_throttle
                .enable_adaptive_estimated_cpu_per_request_us,
            stats_interval_ms: self.stats_interval.as_millis(),
            refill_interval_ms: self.cpu_throttle.refill_interval.as_millis(),
            burst_window_ms: self.cpu_throttle.burst_window.as_millis(),
            enable_dynamic_adjustment: self.cpu_throttle.enable_dynamic_adjustment,
            high_watermark: self.cpu_throttle.high_watermark,
            low_watermark: self.cpu_throttle.low_watermark,
            enable_fair_allocation: self.cpu_throttle.enable_fair_allocation,
            fair_allocation_threshold: self.cpu_throttle.fair_allocation_threshold,
            enable_burst: self.cpu_throttle.enable_burst,
            burst_threshold: self.cpu_throttle.burst_threshold,
            keyspace_burst_enabled: self.cpu_throttle.keyspace_burst_enabled.clone(),
            enable_runtime_token_management: self.cpu_throttle.enable_runtime_token_management,
            runtime_check_interval_us: self.cpu_throttle.runtime_check_interval_us,
            additional_allocation_threshold: self.cpu_throttle.additional_allocation_threshold,
            per_allocation_us: self.cpu_throttle.per_allocation_us,
            debug: self.debug,
            enforce_global_limit: self.cpu_throttle.enforce_global_limit,
            enable_pd_service_limit: self.cpu_throttle.enable_pd_service_limit,
            pd_service_limit_fetch_interval_ms: self
                .cpu_throttle
                .pd_service_limit_fetch_interval
                .as_millis(),
            pd_service_limit_request_timeout_ms: self
                .cpu_throttle
                .pd_service_limit_request_timeout
                .as_millis(),
            pd_service_limit_ttl_ms: self.cpu_throttle.pd_service_limit_ttl.as_millis(),
            pd_service_limit_consecutive_404: self.cpu_throttle.pd_service_limit_consecutive_404,
            pd_service_limit_update_threshold: self.cpu_throttle.pd_service_limit_update_threshold,
            pd_override_config: self.cpu_throttle.pd_override_config,
        }
    }
}

pub struct ConfigManager {
    pub(crate) event_publisher: Arc<dyn ResourcePublisher>,
    pub(crate) current_config: Config,
    legacy_read_limiter_allowed: bool,
}

impl ConfigManager {
    pub fn new(event_publisher: Arc<dyn ResourcePublisher>, config: Config) -> Self {
        Self::new_with_legacy_read_limiter(event_publisher, config, true)
    }

    pub fn new_with_legacy_read_limiter(
        event_publisher: Arc<dyn ResourcePublisher>,
        config: Config,
        legacy_read_limiter_allowed: bool,
    ) -> Self {
        info!(
            "new config manager for resource control {:?}",
            config;
            "legacy_read_limiter_allowed" => legacy_read_limiter_allowed,
        );
        Self {
            event_publisher,
            current_config: config,
            legacy_read_limiter_allowed,
        }
    }
}

impl online_config::ConfigManager for ConfigManager {
    fn dispatch(&mut self, change: ConfigChange) -> online_config::Result<()> {
        info!("resource control config change {:?}", change);
        let mut new_config = self.current_config.clone();
        new_config.update(change)?;
        new_config.validate()?;
        if new_config.max_read_cpu_ratio != self.current_config.max_read_cpu_ratio
            && !self.legacy_read_limiter_allowed
        {
            warn!(
                "resource-control.max-read-cpu-ratio is not applied online while keyspace fairness is enabled; restart is required";
                "current" => self.current_config.max_read_cpu_ratio,
                "requested" => new_config.max_read_cpu_ratio,
            );
            new_config.max_read_cpu_ratio = self.current_config.max_read_cpu_ratio;
        }
        self.event_publisher
            .publish(ResourceEvent::UpdateConfig(new_config.clone()));
        self.current_config = new_config;
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use online_config::{ConfigManager as _, ConfigValue};

    use super::*;

    struct NoopPublisher;

    impl ResourcePublisher for NoopPublisher {
        fn publish(&self, _event: ResourceEvent) {}
    }

    #[test]
    fn test_max_read_cpu_ratio_online_config_metadata() {
        let config = Config::default();
        let mut changed = config.clone();
        changed.max_read_cpu_ratio = 0.9;

        let diff = config.diff(&changed);
        assert_eq!(diff.get("max_read_cpu_ratio"), Some(&ConfigValue::F64(0.9)));
        assert_eq!(
            config.typed().get("max_read_cpu_ratio"),
            Some(&ConfigValue::F64(config.max_read_cpu_ratio))
        );
    }

    #[test]
    fn test_online_update_allows_max_read_cpu_ratio_without_fairness() {
        let current = Config {
            enabled: true,
            read_enabled: true,
            cpu_throttle: CpuThrottleSettings {
                enabled: true,
                ..Default::default()
            },
            ..Config::default()
        };
        let mut changed = current.clone();
        changed.max_read_cpu_ratio = 0.9;
        let change = current.diff(&changed);
        let mut manager = ConfigManager::new(Arc::new(NoopPublisher), current);

        manager.dispatch(change).unwrap();
        assert_eq!(manager.current_config.max_read_cpu_ratio, 0.9);
    }

    #[test]
    fn test_online_update_defers_max_read_cpu_ratio_with_fairness() {
        let current = Config {
            enabled: true,
            cpu_throttle: CpuThrottleSettings {
                enabled: true,
                ..Default::default()
            },
            ..Config::default()
        };
        let mut changed = current.clone();
        changed.max_read_cpu_ratio = 0.9;
        changed.dry_run = true;
        let change = current.diff(&changed);
        let mut manager =
            ConfigManager::new_with_legacy_read_limiter(Arc::new(NoopPublisher), current, false);

        manager.dispatch(change).unwrap();
        assert_eq!(manager.current_config.max_read_cpu_ratio, 0.6);
        assert!(manager.current_config.dry_run);
    }

    #[test]
    fn test_online_update_defers_ratio_when_enabling_cpu_throttle_with_fairness() {
        let current = Config {
            enabled: true,
            ..Config::default()
        };
        let mut changed = current.clone();
        changed.max_read_cpu_ratio = 0.9;
        changed.cpu_throttle.enabled = true;
        let change = current.diff(&changed);
        let mut manager =
            ConfigManager::new_with_legacy_read_limiter(Arc::new(NoopPublisher), current, false);

        manager.dispatch(change).unwrap();
        assert_eq!(manager.current_config.max_read_cpu_ratio, 0.6);
        assert!(manager.current_config.cpu_throttle.enabled);
    }

    #[test]
    fn test_read_limiter_config_round_trip() {
        let config: Config = serde_json::from_value(serde_json::json!({
            "limiter-timeout": "3s",
            "smoothing-factor": 0.8,
            "active-quota-ratio": 0.7,
            "min-quota-ratio": 0.05,
            "severity-stressed-factor": 1.1,
            "severity-critical-factor": 0.9,
            "severity-exhausted-factor": 0.5,
            "read-enabled": true,
            "max-read-bytes-factor": 2.0,
            "max-read-wait-time": "10ms",
            "enable-qps-bytes-limiter": true
        }))
        .unwrap();
        assert!(config.read_enabled);
        assert!(config.enable_qps_bytes_limiter);

        let serialized = serde_json::to_value(config).unwrap();
        for key in [
            "limiter-timeout",
            "smoothing-factor",
            "active-quota-ratio",
            "min-quota-ratio",
            "severity-stressed-factor",
            "severity-critical-factor",
            "severity-exhausted-factor",
            "read-enabled",
            "max-read-bytes-factor",
            "max-read-wait-time",
            "enable-qps-bytes-limiter",
        ] {
            assert!(serialized.get(key).is_some(), "{key} should be serialized");
        }
    }

    #[test]
    fn test_starter_and_rc_postures_use_separate_switches() {
        let starter = Config {
            enabled: true,
            read_enabled: true,
            cpu_throttle: CpuThrottleSettings {
                enabled: false,
                ..CpuThrottleSettings::default()
            },
            ..Config::default()
        };
        assert!(starter.get_read_enabled());
        assert!(!starter.get_cpu_throttle_enabled());

        let rc = Config {
            enabled: true,
            read_enabled: false,
            cpu_throttle: CpuThrottleSettings {
                enabled: true,
                ..CpuThrottleSettings::default()
            },
            ..Config::default()
        };
        assert!(!rc.get_read_enabled());
        assert!(rc.get_cpu_throttle_enabled());
    }

    #[test]
    fn test_validate_accepts_legacy_read_limiter_with_cpu_throttle() {
        let config = Config {
            enabled: true,
            read_enabled: true,
            cpu_throttle: CpuThrottleSettings {
                enabled: true,
                ..CpuThrottleSettings::default()
            },
            ..Config::default()
        };

        config.validate().unwrap();
    }

    #[test]
    fn test_online_update_accepts_legacy_read_limiter_with_cpu_throttle() {
        let current = Config {
            enabled: true,
            read_enabled: true,
            ..Config::default()
        };
        let mut changed = current.clone();
        changed.cpu_throttle.enabled = true;
        let change = current.diff(&changed);
        let mut manager = ConfigManager::new(Arc::new(NoopPublisher), current);

        manager.dispatch(change).unwrap();
        assert!(manager.current_config.cpu_throttle.enabled);
    }

    #[test]
    fn test_validate_ignores_dormant_legacy_read_limiter_settings() {
        let mut config = Config {
            enabled: true,
            read_enabled: false,
            limiter_timeout: ReadableDuration::ZERO,
            smoothing_factor: 0.0,
            active_quota_ratio: 0.0,
            min_quota_ratio: 0.0,
            max_read_wait_time: ReadableDuration::ZERO,
            ..Config::default()
        };
        config.validate().unwrap();

        config.read_enabled = true;
        assert_eq!(
            config.validate().unwrap_err().to_string(),
            "limiter-timeout cannot be less than 1 milliseconds"
        );
    }

    #[test]
    fn test_validate_rejects_tiny_positive_tenant_cap_ratio() {
        let mut config = Config::default();
        config.enabled = true;
        config.cpu_throttle.enabled = true;
        config.cpu_throttle.tenant_max_cpu_ratio = f64::MIN_POSITIVE;
        config.cpu_throttle.enforce_global_limit = true;
        config.cpu_throttle.enable_pd_service_limit = false;

        let err = config.validate().unwrap_err().to_string();
        assert!(err.contains("0us quota"));
    }
}