// Copyright 2026 TiKV Project Authors. Licensed under Apache-2.0.

//! Kvengine online config manager.

use std::sync::Arc;

use online_config::{ConfigChange, ConfigManager, ConfigValue, OnlineConfig, Result as CfgResult};
use slog_global::warn;
use tikv_util::{
    config::{AbsoluteOrPercentSize, ReadableDuration, ReadableSize, VersionTrack},
    time::{monotonic_raw_now, timespec_to_ns},
};

use crate::{
    config::Config as KvEngineConfig,
    context::IaCtx,
    ia::{
        manager::{
            SMALL_QUEUE_MIN_SEGMENT_COUNT, small_queue_capacity_floor,
            validate_cache_capacity_ratio, validate_cache_capacity_shrink_step,
        },
        transition::TransitionLimiter,
    },
    options::Options,
};

pub struct KvEngineConfigManager {
    config: Arc<VersionTrack<KvEngineConfig>>,
    ia_ctx: IaCtx,
    opts: Arc<Options>,
}

#[derive(Default)]
struct IaSideEffects {
    cache_capacity: Option<CacheCapacityUpdate>,
    remote_read_max_throughput: Option<ReadableSize>,
    local_write_max_throughput: Option<ReadableSize>,
    transition_max_throughput: Option<ReadableSize>,
    transition_max_regions: Option<usize>,
    transition_wait_timeout: Option<ReadableDuration>,
}

#[derive(Clone, Copy)]
struct CacheCapacityUpdate {
    mem_cap: Option<AbsoluteOrPercentSize>,
    disk_cap: Option<AbsoluteOrPercentSize>,
    cache_cap_to_total_data_size_ratio: Option<f64>,
    shrink_step: Option<ReadableSize>,
}

impl KvEngineConfigManager {
    pub fn new(
        config: Arc<VersionTrack<KvEngineConfig>>,
        ia_ctx: IaCtx,
        opts: Arc<Options>,
    ) -> Self {
        Self {
            config,
            ia_ctx,
            opts,
        }
    }

    fn collect_ia_side_effects(
        change: &ConfigChange,
        new_config: &KvEngineConfig,
    ) -> CfgResult<Option<IaSideEffects>> {
        let Some(ConfigValue::Module(ia_change)) = change.get("ia") else {
            return Ok(None);
        };
        let mut ia_change = ia_change.clone();
        let mem_cap_changed = ia_change.remove("mem_cap").is_some();
        let disk_cap_changed = ia_change.remove("disk_cap").is_some();
        let cache_cap_to_total_data_size_ratio = ia_change
            .remove("cache_cap_to_total_data_size_ratio")
            .map(|_| new_config.ia.cache_cap_to_total_data_size_ratio);
        if let Some(ratio) = cache_cap_to_total_data_size_ratio {
            validate_cache_capacity_ratio(ratio)?;
        }
        let cache_capacity_shrink_step = ia_change
            .remove("cache_capacity_shrink_step")
            .map(|_| new_config.ia.cache_capacity_shrink_step);
        if let Some(step) = cache_capacity_shrink_step {
            validate_cache_capacity_shrink_step(step)?;
        }
        if mem_cap_changed {
            let mem_cap = new_config.ia.mem_cap.as_memory_size();
            let min_mem_cap = small_queue_capacity_floor(new_config.ia.segment_size);
            if min_mem_cap <= 0 || mem_cap < min_mem_cap as u64 {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::InvalidInput,
                    format!(
                        "kvengine.ia.mem-cap must be at least {SMALL_QUEUE_MIN_SEGMENT_COUNT} * kvengine.ia.segment-size for online updates, mem-cap: {mem_cap}, minimum: {min_mem_cap}"
                    ),
                )
                .into());
            }
        }
        let remote_read_max_throughput = ia_change
            .remove("remote_read_max_throughput")
            .map(|_| new_config.ia.remote_read_max_throughput);
        let local_write_max_throughput = ia_change
            .remove("local_write_max_throughput")
            .map(|_| new_config.ia.local_write_max_throughput);
        let transition_max_throughput = ia_change
            .remove("ia_to_standard_transition_max_throughput")
            .map(|_| new_config.ia.ia_to_standard_transition_max_throughput);
        let transition_max_regions = ia_change
            .remove("ia_to_standard_transition_max_regions")
            .map(|_| new_config.ia.ia_to_standard_transition_max_regions);
        let transition_wait_timeout = ia_change
            .remove("ia_to_standard_transition_wait_timeout")
            .map(|_| new_config.ia.ia_to_standard_transition_wait_timeout);
        if !ia_change.is_empty() {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidInput,
                format!("unsupported kvengine.ia online config: {ia_change:?}"),
            )
            .into());
        }
        if transition_max_regions == Some(0) {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidInput,
                "kvengine.ia.ia-to-standard-transition-max-regions should be >= 1",
            )
            .into());
        }
        if transition_wait_timeout
            .as_ref()
            .is_some_and(|timeout| timeout.0.is_zero())
        {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidInput,
                "kvengine.ia.ia-to-standard-transition-wait-timeout should be > 0",
            )
            .into());
        }
        let cache_capacity = (mem_cap_changed
            || disk_cap_changed
            || cache_cap_to_total_data_size_ratio.is_some()
            || cache_capacity_shrink_step.is_some())
        .then_some(CacheCapacityUpdate {
            mem_cap: mem_cap_changed.then_some(new_config.ia.mem_cap),
            disk_cap: disk_cap_changed.then_some(new_config.ia.disk_cap),
            cache_cap_to_total_data_size_ratio,
            shrink_step: cache_capacity_shrink_step,
        });
        Ok(Some(IaSideEffects {
            cache_capacity,
            remote_read_max_throughput,
            local_write_max_throughput,
            transition_max_throughput,
            transition_max_regions,
            transition_wait_timeout,
        }))
    }

    fn apply_ia_side_effects(&self, side_effects: IaSideEffects) -> CfgResult<()> {
        match &self.ia_ctx {
            IaCtx::Enabled(ia_mgr, _) => {
                if let Some(update) = side_effects.cache_capacity {
                    ia_mgr.update_cache_capacity_config(
                        update.mem_cap,
                        update.disk_cap,
                        update.cache_cap_to_total_data_size_ratio,
                        update.shrink_step,
                    )?;
                }
                if let Some(max_throughput) = side_effects.remote_read_max_throughput {
                    ia_mgr.update_remote_read_rate_limit(max_throughput);
                }
                if let Some(max_throughput) = side_effects.local_write_max_throughput {
                    ia_mgr.update_local_write_rate_limit(max_throughput);
                }
            }
            IaCtx::Disabled(_) => {
                if side_effects.cache_capacity.is_some() {
                    return Err(std::io::Error::new(
                        std::io::ErrorKind::InvalidInput,
                        "IA cache capacity can not be updated because IA is disabled",
                    )
                    .into());
                }
                if side_effects.remote_read_max_throughput.is_some()
                    || side_effects.local_write_max_throughput.is_some()
                {
                    warn!("skip updating IA IO rate limiter because IA is disabled");
                }
            }
        }

        let transition_limiter = self.ia_ctx.transition_limiter();
        Self::apply_transition_side_effects(&transition_limiter, &side_effects);
        Ok(())
    }

    fn apply_transition_side_effects(limiter: &TransitionLimiter, side_effects: &IaSideEffects) {
        if let Some(max_throughput) = side_effects.transition_max_throughput {
            limiter.update_max_throughput(max_throughput);
        }
        if let Some(max_regions) = side_effects.transition_max_regions {
            limiter.update_max_regions(max_regions);
        }
        if let Some(wait_timeout) = side_effects.transition_wait_timeout {
            limiter.update_wait_timeout(wait_timeout.0);
        }
    }
}

impl ConfigManager for KvEngineConfigManager {
    fn dispatch(&mut self, change: ConfigChange) -> CfgResult<()> {
        if change.is_empty() {
            return Ok(());
        }
        warn!(
            "kvengine online config changed";
            "change" => ?change,
        );

        let mut new_config = self.config.value().clone();
        new_config.update(change.clone())?;
        new_config.validate()?;

        let ia_side_effects = Self::collect_ia_side_effects(&change, &new_config)?;
        let update_compaction_old_data_opts = change.contains_key("compaction_old_data_ratio")
            || change.contains_key("compaction_old_data_size");
        let update_file_gc_ratio_opts = change.contains_key("file_gc_compaction_ratio")
            || change.contains_key("file_gc_compaction_max_overlap_ratio");
        let update_file_gc_trigger_interval =
            change.contains_key("file_gc_compaction_min_interval");
        let update_file_gc_allow_l3 = change.contains_key("file_gc_compaction_allow_l3");
        if let Some(side_effects) = ia_side_effects {
            self.apply_ia_side_effects(side_effects)?;
        }
        self.config.update(|config| -> CfgResult<()> {
            *config = new_config;
            Ok(())
        })?;
        if update_compaction_old_data_opts {
            let cfg = self.config.value();
            self.opts
                .set_compaction_old_data_ratio(cfg.compaction_old_data_ratio);
            self.opts
                .set_compaction_old_data_size(cfg.compaction_old_data_size.0);
        }
        if update_file_gc_ratio_opts {
            let cfg = self.config.value();
            self.opts
                .set_file_gc_compaction_ratio(cfg.file_gc_compaction_ratio);
            self.opts
                .set_file_gc_compaction_max_overlap_ratio(cfg.file_gc_compaction_max_overlap_ratio);
        }
        if update_file_gc_trigger_interval {
            let cfg = self.config.value();
            self.opts.set_file_gc_compaction_min_interval_ns(
                cfg.file_gc_compaction_min_interval.0.as_nanos() as u64,
            );
        }
        if update_file_gc_allow_l3 {
            let cfg = self.config.value();
            self.opts
                .set_file_gc_compaction_allow_l3(cfg.file_gc_compaction_allow_l3);
        }
        if update_file_gc_ratio_opts || update_file_gc_allow_l3 {
            self.opts
                .set_file_gc_compaction_config_changed_at(timespec_to_ns(monotonic_raw_now()));
        }
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use online_config::{ConfigChange, ConfigValue, OnlineConfig};
    use tikv_util::config::{GIB, MIB, ReadableDuration, ReadableSize};

    use super::*;
    use crate::{
        dfs::{FileType, InMemFs},
        ia::{
            manager::{IaManager, QueueOptions},
            types::FileSegmentIdent,
            util::{IaCapacity, IaManagerOptionsBuilder},
        },
    };

    fn new_manager(
        cfg: KvEngineConfig,
    ) -> (
        Arc<VersionTrack<KvEngineConfig>>,
        Arc<Options>,
        KvEngineConfigManager,
    ) {
        let ia_ctx = IaCtx::disabled(&cfg.ia);
        let track = Arc::new(VersionTrack::new(cfg));
        let opts = Arc::new(Options::default());
        let mgr = KvEngineConfigManager::new(track.clone(), ia_ctx, opts.clone());
        (track, opts, mgr)
    }

    fn change_ia(ia_change: ConfigChange) -> ConfigChange {
        let mut change = ConfigChange::new();
        change.insert("ia".to_owned(), ConfigValue::Module(ia_change));
        change
    }

    fn new_enabled_manager(
        cfg: KvEngineConfig,
        small_queue_cap: i64,
        main_queue_cap: i64,
    ) -> (
        Arc<VersionTrack<KvEngineConfig>>,
        KvEngineConfigManager,
        IaManager,
    ) {
        let runtime = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(1)
            .enable_all()
            .build()
            .unwrap();
        let options = IaManagerOptionsBuilder::default()
            .capacity(IaCapacity::Manual {
                small_queue: QueueOptions {
                    paths: vec![],
                    cap: small_queue_cap,
                },
                main_queue: QueueOptions {
                    paths: vec![],
                    cap: main_queue_cap,
                },
            })
            .segment_size(cfg.ia.segment_size)
            .dfs_concurrency(1, 1)
            .dynamic_capacity(cfg.ia.dynamic_capacity)
            .cache_cap_to_total_data_size_ratio(cfg.ia.cache_cap_to_total_data_size_ratio)
            .cache_capacity_shrink_step(cfg.ia.cache_capacity_shrink_step)
            .build()
            .unwrap();
        let ia_mgr =
            IaManager::new(options, Arc::new(InMemFs::new()), None, runtime.into()).unwrap();
        let ia_ctx = IaCtx::Enabled(ia_mgr.clone(), Arc::new(vec![]));
        let track = Arc::new(VersionTrack::new(cfg));
        let opts = Arc::new(Options::default());
        let manager = KvEngineConfigManager::new(track.clone(), ia_ctx, opts);
        (track, manager, ia_mgr)
    }

    #[test]
    fn test_collect_ia_side_effects_validates_only_changed_mem_cap() {
        let mut config = KvEngineConfig::default();
        config.ia.segment_size = 1;
        config.ia.mem_cap = 32_u64.into();

        let mut new_config = config.clone();
        new_config.ia.disk_cap = 900_u64.into();
        let disk_only_change = config.diff(&new_config);
        KvEngineConfigManager::collect_ia_side_effects(&disk_only_change, &new_config)
            .unwrap()
            .unwrap();

        new_config = config.clone();
        new_config.ia.mem_cap = 63_u64.into();
        let below_floor_change = config.diff(&new_config);
        let err = KvEngineConfigManager::collect_ia_side_effects(&below_floor_change, &new_config)
            .err()
            .unwrap();
        assert!(err.to_string().contains("64 * kvengine.ia.segment-size"));

        new_config.ia.mem_cap = 64_u64.into();
        let at_floor_change = config.diff(&new_config);
        KvEngineConfigManager::collect_ia_side_effects(&at_floor_change, &new_config)
            .unwrap()
            .unwrap();
    }

    #[test]
    fn test_dispatch_capacity_partial_updates_with_enabled_ia() {
        let mut cfg = KvEngineConfig::default();
        cfg.ia.segment_size = 1;
        cfg.ia.mem_cap = 32_u64.into();
        cfg.ia.disk_cap = 100_u64.into();
        let (track, mut manager, ia_mgr) = new_enabled_manager(cfg.clone(), 32, 100);

        let mut disk_only = cfg.clone();
        disk_only.ia.disk_cap = 200_u64.into();
        manager.dispatch(cfg.diff(&disk_only)).unwrap();
        assert_eq!(track.value().ia.mem_cap, 32_u64.into());
        assert_eq!(track.value().ia.disk_cap, 200_u64.into());
        assert_eq!(ia_mgr.cache_queue_capacities(), (32, 100));

        let snapshot_before = track.value().clone();
        let mut below_floor = snapshot_before.clone();
        below_floor.ia.mem_cap = 63_u64.into();
        let err = manager
            .dispatch(snapshot_before.diff(&below_floor))
            .unwrap_err();
        assert!(err.to_string().contains("64 * kvengine.ia.segment-size"));
        assert_eq!(*track.value(), snapshot_before);

        let mut at_floor = snapshot_before.clone();
        at_floor.ia.mem_cap = 64_u64.into();
        manager.dispatch(snapshot_before.diff(&at_floor)).unwrap();
        assert_eq!(track.value().ia.mem_cap, 64_u64.into());
        assert_eq!(track.value().ia.disk_cap, 200_u64.into());
        assert_eq!(ia_mgr.cache_queue_capacities(), (32, 100));

        ia_mgr.notify_total_data_size(10_000);
        assert_eq!(ia_mgr.cache_queue_capacities(), (64, 200));
        let ia_mgr_for_read = ia_mgr.clone();
        let err = ia_mgr
            .runtime_handle()
            .block_on(async move {
                ia_mgr_for_read
                    .get_segment_handle(FileSegmentIdent::new(1, 0, 65), FileType::Sst, None, true)
                    .await
            })
            .unwrap_err();
        assert!(format!("{err:?}").contains("capacity: 64"));
    }

    #[test]
    fn test_dispatch_defers_cache_capacity_snapshot_until_next_size_notification() {
        let mut cfg = KvEngineConfig::default();
        cfg.ia.segment_size = MIB as i64;
        cfg.ia.mem_cap = (100 * MIB).into();
        cfg.ia.disk_cap = (10 * GIB).into();
        cfg.ia.dynamic_capacity = true;
        cfg.ia.cache_cap_to_total_data_size_ratio = 1.0;
        cfg.ia.cache_capacity_shrink_step = ReadableSize::gb(10);
        let (track, mut manager, ia_mgr) =
            new_enabled_manager(cfg.clone(), 100 * MIB as i64, 10 * GIB as i64);

        ia_mgr.notify_total_data_size(4 * GIB);
        let mut updated = cfg.clone();
        updated.ia.mem_cap = (80 * MIB).into();
        updated.ia.disk_cap = (3 * GIB).into();
        updated.ia.cache_cap_to_total_data_size_ratio = 0.75;
        manager.dispatch(cfg.diff(&updated)).unwrap();

        assert_eq!(track.value().ia.mem_cap, (80 * MIB).into());
        assert_eq!(track.value().ia.disk_cap, (3 * GIB).into());
        assert_eq!(track.value().ia.cache_cap_to_total_data_size_ratio, 0.75);
        assert_eq!(
            ia_mgr.cache_queue_capacities(),
            (64 * MIB as i64, 4 * GIB as i64)
        );

        ia_mgr.notify_total_data_size(4 * GIB);
        assert_eq!(
            ia_mgr.cache_queue_capacities(),
            (80 * MIB as i64, 3 * GIB as i64)
        );
    }

    #[test]
    fn test_dispatch_ratio_is_inert_when_dynamic_capacity_is_disabled() {
        let mut cfg = KvEngineConfig::default();
        cfg.ia.segment_size = MIB as i64;
        cfg.ia.mem_cap = (100 * MIB).into();
        cfg.ia.disk_cap = (10 * GIB).into();
        cfg.ia.dynamic_capacity = false;
        cfg.ia.cache_cap_to_total_data_size_ratio = 1.0;
        let (track, mut manager, ia_mgr) =
            new_enabled_manager(cfg.clone(), 100 * MIB as i64, 10 * GIB as i64);

        let mut updated = cfg.clone();
        updated.ia.cache_cap_to_total_data_size_ratio = 0.0;
        manager.dispatch(cfg.diff(&updated)).unwrap();
        assert_eq!(track.value().ia.cache_cap_to_total_data_size_ratio, 0.0);

        ia_mgr.notify_total_data_size(0);
        assert_eq!(
            ia_mgr.cache_queue_capacities(),
            (100 * MIB as i64, 10 * GIB as i64)
        );
    }

    #[test]
    fn test_dispatch_rejects_invalid_cache_capacity_ratio_without_side_effects() {
        let mut cfg = KvEngineConfig::default();
        cfg.ia.segment_size = MIB as i64;
        cfg.ia.mem_cap = (100 * MIB).into();
        cfg.ia.disk_cap = (10 * GIB).into();
        cfg.ia.dynamic_capacity = true;
        cfg.ia.cache_cap_to_total_data_size_ratio = 1.0;
        cfg.ia.cache_capacity_shrink_step = ReadableSize::gb(10);
        let (track, mut manager, ia_mgr) =
            new_enabled_manager(cfg.clone(), 100 * MIB as i64, 10 * GIB as i64);
        ia_mgr.notify_total_data_size(4 * GIB);

        let mut invalid = cfg.clone();
        invalid.ia.mem_cap = (80 * MIB).into();
        invalid.ia.disk_cap = (8 * GIB).into();
        invalid.ia.cache_cap_to_total_data_size_ratio = -0.1;
        let err = manager.dispatch(cfg.diff(&invalid)).unwrap_err();
        assert!(
            err.to_string()
                .contains("cache-cap-to-total-data-size-ratio")
        );
        assert_eq!(*track.value(), cfg);
        assert_eq!(
            ia_mgr.cache_queue_capacities(),
            (64 * MIB as i64, 4 * GIB as i64)
        );

        // The runtime ratio and max caps remain unchanged after validation
        // fails.
        ia_mgr.notify_total_data_size(10 * GIB);
        assert_eq!(
            ia_mgr.cache_queue_capacities(),
            (100 * MIB as i64, 10 * GIB as i64)
        );
    }

    #[test]
    fn test_dispatch_updates_cache_capacity_shrink_step_for_next_report() {
        let mut cfg = KvEngineConfig::default();
        cfg.ia.segment_size = MIB as i64;
        cfg.ia.mem_cap = (5 * GIB).into();
        cfg.ia.disk_cap = (20 * GIB).into();
        cfg.ia.dynamic_capacity = true;
        cfg.ia.cache_cap_to_total_data_size_ratio = 1.0;
        let (track, mut manager, ia_mgr) =
            new_enabled_manager(cfg.clone(), 5 * GIB as i64, 20 * GIB as i64);

        ia_mgr.notify_total_data_size(100 * GIB);
        assert_eq!(
            ia_mgr.cache_queue_capacities(),
            (5 * GIB as i64, 20 * GIB as i64)
        );

        let mut updated = cfg.clone();
        updated.ia.cache_cap_to_total_data_size_ratio = 0.0;
        updated.ia.cache_capacity_shrink_step = ReadableSize::gb(2);
        manager.dispatch(cfg.diff(&updated)).unwrap();
        assert_eq!(
            track.value().ia.cache_capacity_shrink_step,
            ReadableSize::gb(2)
        );
        assert_eq!(
            ia_mgr.cache_queue_capacities(),
            (5 * GIB as i64, 20 * GIB as i64)
        );

        ia_mgr.notify_total_data_size(100 * GIB);
        assert_eq!(
            ia_mgr.cache_queue_capacities(),
            (9 * GIB as i64 / 2, 18 * GIB as i64)
        );

        let snapshot_before = track.value().clone();
        let mut invalid = snapshot_before.clone();
        invalid.ia.cache_capacity_shrink_step = ReadableSize(0);
        let err = manager
            .dispatch(snapshot_before.diff(&invalid))
            .unwrap_err();
        assert!(err.to_string().contains("cache-capacity-shrink-step"));
        assert_eq!(*track.value(), snapshot_before);
        assert_eq!(
            ia_mgr.cache_queue_capacities(),
            (9 * GIB as i64 / 2, 18 * GIB as i64)
        );
    }

    #[test]
    fn test_dispatch_applies_online_fields_to_snapshot() {
        let cfg = KvEngineConfig::default();
        let (track, _, mut mgr) = new_manager(cfg.clone());

        // Build a ConfigChange by diffing two configs. This catches drift
        // between the fields the OnlineConfig derive emits and what the
        // snapshot actually carries; the manager itself just forwards.
        let mut new_cfg = cfg.clone();
        new_cfg.remote_worker_addr = "http://worker:1".into();
        new_cfg.remote_coprocessor_addr = "http://cop:2".into();
        new_cfg.remote_coprocessor_min_blocks_size = 4096;
        new_cfg.remote_coprocessor_num_ranges = 8;
        new_cfg.remote_coprocessor_min_process_duration =
            cfg.remote_coprocessor_min_process_duration + ReadableDuration::millis(1);
        new_cfg.remote_coprocessor_topn_max_pushdown_rows = 12_345;
        let change: ConfigChange = cfg.diff(&new_cfg);
        assert!(!change.is_empty(), "diff must include the online fields");

        mgr.dispatch(change).unwrap();

        let live = track.value();
        assert_eq!(live.remote_worker_addr, "http://worker:1");
        assert_eq!(live.remote_coprocessor_addr, "http://cop:2");
        assert_eq!(live.remote_coprocessor_min_blocks_size, 4096);
        assert_eq!(live.remote_coprocessor_num_ranges, 8);
        assert_eq!(
            live.remote_coprocessor_min_process_duration,
            new_cfg.remote_coprocessor_min_process_duration
        );
        assert_eq!(live.remote_coprocessor_topn_max_pushdown_rows, 12_345);
    }

    #[test]
    fn test_dispatch_partial_change_leaves_others_untouched() {
        let mut cfg = KvEngineConfig::default();
        cfg.remote_worker_addr = "init".into();
        cfg.remote_coprocessor_addr = "init".into();
        cfg.remote_coprocessor_min_blocks_size = 1;
        cfg.remote_coprocessor_num_ranges = 2;
        let (track, _, mut mgr) = new_manager(cfg);

        let mut change = ConfigChange::new();
        change.insert(
            "remote_coprocessor_min_blocks_size".to_string(),
            ConfigValue::Usize(99),
        );
        mgr.dispatch(change).unwrap();

        let live = track.value();
        assert_eq!(live.remote_worker_addr, "init");
        assert_eq!(live.remote_coprocessor_addr, "init");
        assert_eq!(live.remote_coprocessor_min_blocks_size, 99);
        assert_eq!(live.remote_coprocessor_num_ranges, 2);
    }

    #[test]
    fn test_dispatch_empty_change_is_noop() {
        let cfg = KvEngineConfig::default();
        let (track, _, mut mgr) = new_manager(cfg.clone());

        mgr.dispatch(ConfigChange::new()).unwrap();
        assert_eq!(track.value().remote_worker_addr, cfg.remote_worker_addr);
    }

    #[test]
    fn test_dispatch_rejects_unknown_ia_subkey() {
        let cfg = KvEngineConfig::default();
        let (track, _, mut mgr) = new_manager(cfg.clone());
        let snapshot_before = track.value().clone();

        let mut ia_change = ConfigChange::new();
        ia_change.insert(
            "remote_read_max_throughput".to_string(),
            ConfigValue::Size(99),
        );
        ia_change.insert("unknown_field".to_string(), ConfigValue::Size(1));

        let err = mgr.dispatch(change_ia(ia_change)).unwrap_err();
        assert!(
            err.to_string().contains("unknown_field"),
            "expected reject for unknown ia subkey, got: {}",
            err
        );
        // Snapshot must not change when validation fails, otherwise IaManager
        // and snapshot can drift on partial-success paths.
        assert_eq!(*track.value(), snapshot_before);
    }

    #[test]
    fn test_dispatch_rejects_capacity_change_when_ia_disabled() {
        let cfg = KvEngineConfig::default();
        let (track, _, mut mgr) = new_manager(cfg.clone());
        let snapshot_before = track.value().clone();

        let mut ia_change = ConfigChange::new();
        ia_change.insert(
            "mem_cap".to_string(),
            ConfigValue::Size(ReadableSize::mb(64).0),
        );

        let err = mgr.dispatch(change_ia(ia_change)).unwrap_err();
        assert!(
            err.to_string()
                .contains("IA cache capacity can not be updated because IA is disabled"),
            "expected reject for disabled IA cache update, got: {}",
            err
        );
        assert_eq!(*track.value(), snapshot_before);

        let mut ia_change = ConfigChange::new();
        ia_change.insert(
            "cache_cap_to_total_data_size_ratio".to_string(),
            ConfigValue::F64(0.5),
        );
        let err = mgr.dispatch(change_ia(ia_change)).unwrap_err();
        assert!(
            err.to_string()
                .contains("IA cache capacity can not be updated because IA is disabled")
        );
        assert_eq!(*track.value(), snapshot_before);
    }

    #[test]
    fn test_dispatch_rejects_capacity_change_before_applying_other_ia_side_effects() {
        let cfg = KvEngineConfig::default();
        let ia_ctx = IaCtx::disabled(&cfg.ia);
        let limiter = ia_ctx.transition_limiter();
        let original_max_regions = limiter.max_regions();
        let track = Arc::new(VersionTrack::new(cfg));
        let opts = Arc::new(Options::default());
        let mut manager = KvEngineConfigManager::new(track.clone(), ia_ctx, opts);
        let snapshot_before = track.value().clone();

        let mut ia_change = ConfigChange::new();
        ia_change.insert(
            "mem_cap".to_owned(),
            ConfigValue::Size(ReadableSize::mb(64).0),
        );
        ia_change.insert(
            "ia_to_standard_transition_max_regions".to_owned(),
            ConfigValue::Usize(2),
        );

        let err = manager.dispatch(change_ia(ia_change)).unwrap_err();
        assert!(
            err.to_string()
                .contains("IA cache capacity can not be updated because IA is disabled")
        );
        assert_eq!(limiter.max_regions(), original_max_regions);
        assert_eq!(*track.value(), snapshot_before);
    }

    #[test]
    fn test_dispatch_updates_ia_to_standard_transition_config() {
        let cfg = KvEngineConfig::default();
        let ia_ctx = IaCtx::disabled(&cfg.ia);
        let limiter = ia_ctx.transition_limiter();
        let track = Arc::new(VersionTrack::new(cfg));
        let opts = Arc::new(Options::default());
        let mut manager = KvEngineConfigManager::new(track.clone(), ia_ctx, opts);
        let mut ia_change = ConfigChange::new();
        ia_change.insert(
            "ia_to_standard_transition_max_throughput".to_owned(),
            ConfigValue::Size(0),
        );
        ia_change.insert(
            "ia_to_standard_transition_max_regions".to_owned(),
            ConfigValue::Usize(2),
        );
        ia_change.insert(
            "ia_to_standard_transition_wait_timeout".to_owned(),
            ConfigValue::Duration(120_000),
        );

        manager.dispatch(change_ia(ia_change)).unwrap();

        assert_eq!(
            track.value().ia.ia_to_standard_transition_max_throughput,
            ReadableSize(0)
        );
        assert_eq!(track.value().ia.ia_to_standard_transition_max_regions, 2);
        assert_eq!(
            track.value().ia.ia_to_standard_transition_wait_timeout,
            ReadableDuration::secs(120)
        );
        assert_eq!(limiter.remote_read_speed_limit(), f64::INFINITY);
        assert_eq!(limiter.max_regions(), 2);
        assert_eq!(limiter.wait_timeout(), ReadableDuration::secs(120).0);
    }

    #[test]
    fn test_dispatch_rejects_invalid_ia_to_standard_transition_config() {
        let cfg = KvEngineConfig::default();
        let (track, _, mut manager) = new_manager(cfg.clone());
        let snapshot_before = track.value().clone();
        let mut ia_change = ConfigChange::new();
        ia_change.insert(
            "ia_to_standard_transition_max_regions".to_owned(),
            ConfigValue::Usize(0),
        );

        let err = manager
            .dispatch(change_ia(ia_change))
            .unwrap_err()
            .to_string();
        assert!(err.contains("ia-to-standard-transition-max-regions"));
        assert_eq!(*track.value(), snapshot_before);

        let mut ia_change = ConfigChange::new();
        ia_change.insert(
            "ia_to_standard_transition_wait_timeout".to_owned(),
            ConfigValue::Duration(0),
        );
        let err = manager
            .dispatch(change_ia(ia_change))
            .unwrap_err()
            .to_string();
        assert!(err.contains("ia-to-standard-transition-wait-timeout"));
        assert_eq!(*track.value(), snapshot_before);
    }

    #[test]
    fn test_dispatch_updates_compaction_old_data_runtime_options() {
        let cfg = KvEngineConfig::default();
        let (track, opts, mut manager) = new_manager(cfg.clone());
        let mut new_cfg = cfg.clone();
        new_cfg.compaction_old_data_ratio = 0.5;
        new_cfg.compaction_old_data_size = ReadableSize::mb(128);
        let change = cfg.diff(&new_cfg);

        manager.dispatch(change).unwrap();

        assert_eq!(track.value().compaction_old_data_ratio, 0.5);
        assert_eq!(
            track.value().compaction_old_data_size,
            ReadableSize::mb(128)
        );
        assert_eq!(opts.compaction_old_data_ratio(), 0.5);
        assert_eq!(opts.compaction_old_data_size(), ReadableSize::mb(128).0);
    }

    #[test]
    fn test_dispatch_rejects_invalid_compaction_old_data_ratio() {
        let cfg = KvEngineConfig::default();
        let (track, opts, mut manager) = new_manager(cfg.clone());
        let snapshot_before = track.value().clone();
        let ratio_before = opts.compaction_old_data_ratio();
        let mut change = ConfigChange::new();
        change.insert(
            "compaction_old_data_ratio".to_owned(),
            ConfigValue::F64(f64::NAN),
        );

        let err = manager.dispatch(change).unwrap_err().to_string();
        assert!(err.contains("compaction-old-data-ratio"));
        assert_eq!(*track.value(), snapshot_before);
        assert_eq!(opts.compaction_old_data_ratio(), ratio_before);
    }

    #[test]
    fn test_dispatch_updates_file_gc_ratio_runtime_options() {
        let cfg = KvEngineConfig::default();
        let (track, opts, mut manager) = new_manager(cfg.clone());
        let mut new_cfg = cfg.clone();
        new_cfg.file_gc_compaction_ratio = 0.5;
        new_cfg.file_gc_compaction_max_overlap_ratio = 10.0;
        let change = cfg.diff(&new_cfg);

        manager.dispatch(change).unwrap();

        assert_eq!(track.value().file_gc_compaction_ratio, 0.5);
        assert_eq!(track.value().file_gc_compaction_max_overlap_ratio, 10.0);
        assert_eq!(opts.file_gc_compaction_ratio(), 0.5);
        assert_eq!(opts.file_gc_compaction_max_overlap_ratio(), 10.0);
    }

    #[test]
    fn test_dispatch_updates_file_gc_trigger_interval_runtime_options() {
        let cfg = KvEngineConfig::default();
        let (track, opts, mut manager) = new_manager(cfg.clone());
        let mut new_cfg = cfg.clone();
        new_cfg.file_gc_compaction_min_interval = ReadableDuration::secs(7);
        let change = cfg.diff(&new_cfg);

        manager.dispatch(change).unwrap();

        assert_eq!(
            track.value().file_gc_compaction_min_interval,
            ReadableDuration::secs(7)
        );
        assert_eq!(
            opts.file_gc_compaction_min_interval_ns(),
            ReadableDuration::secs(7).0.as_nanos() as u64
        );
    }

    #[test]
    fn test_dispatch_stamps_changed_at_only_for_hint_knobs() {
        let cfg = KvEngineConfig::default();

        // A ratio change feeds the candidate hint, so it stamps the marker the
        // leader tick reads to refresh stale hints on write-idle shards.
        let (_, opts, mut manager) = new_manager(cfg.clone());
        assert_eq!(opts.file_gc_compaction_config_changed_at(), 0);
        let mut new_cfg = cfg.clone();
        new_cfg.file_gc_compaction_ratio = 0.5;
        manager.dispatch(cfg.diff(&new_cfg)).unwrap();
        assert_ne!(opts.file_gc_compaction_config_changed_at(), 0);

        // allow_l3 also feeds the hint.
        let (_, opts, mut manager) = new_manager(cfg.clone());
        let mut new_cfg = cfg.clone();
        new_cfg.file_gc_compaction_allow_l3 = !cfg.file_gc_compaction_allow_l3;
        manager.dispatch(cfg.diff(&new_cfg)).unwrap();
        assert_ne!(opts.file_gc_compaction_config_changed_at(), 0);

        // The trigger interval does not feed the hint, so it must not stamp.
        let (_, opts, mut manager) = new_manager(cfg.clone());
        let mut new_cfg = cfg.clone();
        new_cfg.file_gc_compaction_min_interval = ReadableDuration::secs(7);
        manager.dispatch(cfg.diff(&new_cfg)).unwrap();
        assert_eq!(opts.file_gc_compaction_config_changed_at(), 0);
    }

    #[test]
    fn test_dispatch_rejects_invalid_file_gc_compaction_ratio() {
        let cfg = KvEngineConfig::default();
        let (track, opts, mut manager) = new_manager(cfg.clone());
        let snapshot_before = track.value().clone();
        let ratio_before = opts.file_gc_compaction_ratio();
        let mut change = ConfigChange::new();
        change.insert(
            "file_gc_compaction_ratio".to_owned(),
            ConfigValue::F64(f64::NAN),
        );

        let err = manager.dispatch(change).unwrap_err().to_string();
        assert!(err.contains("file-gc-compaction-ratio"));
        assert_eq!(*track.value(), snapshot_before);
        assert_eq!(opts.file_gc_compaction_ratio(), ratio_before);
    }
}