// Copyright 2024 PingCAP, Inc.

use std::{
    convert::TryFrom,
    io::{Read, Write},
    ops,
    ops::Deref,
    path::{Path, PathBuf},
    sync::{
        Arc,
        atomic::{AtomicI64, Ordering},
    },
    time::Duration,
};

use arc_swap::ArcSwap;
use bytes::{Buf, BufMut, Bytes};
use dashmap::DashMap;
use tikv_util::{
    codec::number::{I64_SIZE, U8_SIZE},
    config::{AbsoluteOrPercentSize, GIB, ReadableSize},
    deadline::Deadline,
    time::Instant,
    worker_pool::{WorkerPool, WorkerPoolHandle},
};
use tokio::sync::{OwnedSemaphorePermit, Semaphore};
use tracker::with_tls_tracker;

use crate::{
    dfs,
    dfs::{Dfs, FileType},
    ia::{
        flow_control::IaIoRateLimiter,
        queue::S3FifoHandle,
        transition::TransitionLimiter,
        types::{
            FileSegmentData, FileSegmentIdent, FileSegmentPosition, GuardMap, LocalSegmentMap,
            SegmentHandle, TableMetaInfo,
        },
        util::{LocalStore, new_local_store},
    },
    metrics::{
        ENGINE_IA_MAIN_QUEUE_CAPACITY, ENGINE_IA_MANAGER_SEGMENTS_MEMORY_CAPACITY,
        ENGINE_IA_MANAGER_SEGMENTS_MEMORY_WAIT_DURATION, ENGINE_IA_REMOTE_READ_SEGMENT_CACHE_HIT,
        ENGINE_IA_REMOTE_READ_SEGMENT_CACHE_MISS, ENGINE_IA_REMOTE_READ_SEGMENT_DURATION_HISTOGRAM,
        ENGINE_IA_SMALL_QUEUE_CAPACITY,
    },
    table::{Error, Result, file::FdCache},
    try_some,
};

const MANIFEST_PERSIST_INTERVAL: Duration = Duration::from_secs(60);
const CACHE_CAPACITY_GRANULARITY: i64 = GIB as i64;
// Normalize the decimal config value before integer arithmetic. Fifteen
// decimal places preserve meaningful f64 precision while preventing values
// such as 0.7 from landing one byte below an exact GiB boundary. Truncation
// keeps the normalization from intentionally increasing the configured ratio.
const CACHE_CAPACITY_RATIO_SCALE: u128 = 1_000_000_000_000_000;

pub(crate) const SMALL_QUEUE_MIN_SEGMENT_COUNT: i64 = 64;
const MAIN_QUEUE_MIN_SEGMENT_COUNT: i64 = 256;
const REMOTE_READ_MEMORY_BUFFER_SEGMENT_FACTOR: i64 = 2;

/// Return the dynamic small-queue capacity floor for the given segment size.
#[inline]
pub(crate) fn small_queue_capacity_floor(segment_size: i64) -> i64 {
    segment_size * SMALL_QUEUE_MIN_SEGMENT_COUNT
}

pub(crate) fn validate_cache_capacity_ratio(ratio: f64) -> Result<()> {
    if !ratio.is_finite() || ratio < 0.0 {
        return Err(Error::IaMgr(format!(
            "ia.cache-cap-to-total-data-size-ratio must be finite and non-negative, got {ratio}"
        )));
    }
    Ok(())
}

pub(crate) fn validate_cache_capacity_shrink_step(step: ReadableSize) -> Result<()> {
    if step.0 == 0 {
        return Err(Error::IaMgr(
            "ia.cache-capacity-shrink-step must be positive".to_string(),
        ));
    }
    Ok(())
}

/// `buf` can be empty, which means to request the specified range of data but
/// do not actually read it. Used to prefetch segments from remote.
pub(crate) struct ReadAt<'a> {
    buf: &'a mut [u8],
    offset: u64,

    /// Get handle to keep underlying data available, even the segment is
    /// evicted.
    need_segment_handle: bool,
    segment_handle: Option<SegmentHandle>,
}

impl<'a> ReadAt<'a> {
    pub(crate) fn new(buf: &'a mut [u8], offset: u64, need_segment_handle: bool) -> Self {
        Self {
            buf,
            offset,
            need_segment_handle,
            segment_handle: None,
        }
    }

    pub(crate) fn start_off(&self) -> u64 {
        self.offset
    }

    pub(crate) fn end_off(&self) -> u64 {
        self.offset + self.buf.len() as u64
    }

    fn read_from_segment_bytes(&mut self, ident: FileSegmentIdent, seg_data: &Bytes) {
        let (start_off, end_off) = (self.start_off(), self.end_off());
        debug_assert!(
            start_off <= end_off && ident.start_off <= start_off && end_off <= ident.end_off
        );
        if self.need_segment_handle {
            let handle = SegmentHandle::from_bytes(ident.file_id, seg_data.clone());
            self.set_handle(handle);
        }
        if !self.buf.is_empty() {
            let seg_slice = seg_data.slice(
                (start_off - ident.start_off) as usize..(end_off - ident.start_off) as usize,
            );
            self.buf.copy_from_slice(&seg_slice);
        }
    }

    fn set_handle(&mut self, handle: SegmentHandle) {
        let prev = self.segment_handle.replace(handle);
        debug_assert!(prev.is_none());
    }
}

#[must_use]
struct ReadSegmentStatsCollector {
    ident: FileSegmentIdent,
    start_time: Instant,
    cache_hit: Option<bool>,
}

impl ReadSegmentStatsCollector {
    fn new(ident: FileSegmentIdent) -> Self {
        Self {
            ident,
            start_time: Instant::now_coarse(),
            cache_hit: None,
        }
    }

    fn hit(mut self) {
        self.cache_hit = Some(true);
    }

    fn miss(mut self) {
        self.cache_hit = Some(false);
    }
}

impl Drop for ReadSegmentStatsCollector {
    fn drop(&mut self) {
        let Some(cache_hit) = self.cache_hit else {
            return;
        };

        let elapsed = self.start_time.saturating_elapsed();
        if cache_hit {
            ENGINE_IA_REMOTE_READ_SEGMENT_CACHE_HIT.inc();
        } else {
            ENGINE_IA_REMOTE_READ_SEGMENT_DURATION_HISTOGRAM.observe(elapsed.as_secs_f64());
            ENGINE_IA_REMOTE_READ_SEGMENT_CACHE_MISS.inc();
        }
        debug!(
            "read segment finished";
            "ident" => %self.ident,
            "elapsed" => ?elapsed,
            "cache_hit" => cache_hit
        );
        let remote_read_segment_nanos = elapsed.as_nanos().min(u64::MAX as u128) as u64;
        with_tls_tracker(|tracker| {
            if cache_hit {
                // Count request-visible cache accesses even when the segment was
                // populated by an earlier prefetch in the same request. The
                // remote-read counters below track only actual remote fetches.
                tracker.metrics.ia_cache_hit_count =
                    tracker.metrics.ia_cache_hit_count.saturating_add(1);
            } else {
                tracker.metrics.ia_remote_read_segment_count = tracker
                    .metrics
                    .ia_remote_read_segment_count
                    .saturating_add(1);
                tracker.metrics.ia_remote_read_segment_bytes = tracker
                    .metrics
                    .ia_remote_read_segment_bytes
                    .saturating_add(self.ident.size());
                tracker.metrics.ia_remote_read_segment_nanos = tracker
                    .metrics
                    .ia_remote_read_segment_nanos
                    .saturating_add(remote_read_segment_nanos);
            }
        });
    }
}

/// Note that the capacity is not strictly limited for performance. So some
/// additional buffer (maybe 10%) should be reserved.
#[derive(Default, Clone, Debug)]
pub struct QueueOptions {
    /// It means in memory when `path` is `None`.
    pub paths: Vec<PathBuf>,
    pub cap: i64,
}

#[derive(Default, Clone, Debug)]
pub struct IaManagerOptions {
    pub small_queue: QueueOptions,
    pub main_queue: QueueOptions,
    pub segment_size: i64,

    /// The minimum interval to update "freq" counter in queue.
    ///
    /// Used to handle the scene that a single request touch multiple slice of a
    /// segment and increase the freq unexpectedly.
    pub freq_update_interval: Duration,

    pub dfs_concurrency: usize,
    pub dfs_keyspace_concurrency: usize,

    // The capacity of the file descriptor cache.
    pub fd_cache_capacity: usize,

    pub table_meta_mtime_interval: Duration,

    /// Whether periodic IA data-size reports dynamically adjust queue
    /// capacities. When disabled, the configured memory and disk capacities
    /// are the queue-capacity targets.
    pub dynamic_capacity: bool,

    /// The ratio of cache capacity to total data size.
    ///
    /// Used to dynamically adjust cache capacity when the total IA data size
    /// changes and dynamic capacity is enabled.
    pub cache_cap_to_total_data_size_ratio: f64,

    /// Maximum main-queue capacity decrease per periodic reconciliation.
    /// The small queue uses a proportional step.
    pub cache_capacity_shrink_step: ReadableSize,

    pub disable_sync_read: bool,
    pub sync_read_concurrency: usize,
    pub sync_read_timeout: Duration,

    pub remote_read_max_throughput: ReadableSize,
    pub local_write_max_throughput: ReadableSize,
    pub io_rate_limiter_refill_period: Duration,
    pub transition_max_throughput: ReadableSize,
    pub transition_max_regions: usize,
    pub transition_wait_timeout: Duration,
}

impl IaManagerOptions {
    pub fn total_capacity(&self) -> i64 {
        self.small_queue.cap + self.main_queue.cap
    }

    #[inline]
    pub fn small_queue_min_cap(&self) -> i64 {
        if self.dynamic_capacity {
            small_queue_capacity_floor(self.segment_size).min(self.small_queue.cap)
        } else {
            self.small_queue.cap
        }
    }

    #[inline]
    pub fn main_queue_min_cap(&self) -> i64 {
        if self.dynamic_capacity {
            (self.segment_size * MAIN_QUEUE_MIN_SEGMENT_COUNT).min(self.main_queue.cap)
        } else {
            self.main_queue.cap
        }
    }
}

#[derive(Clone, Copy, Debug)]
struct CacheCapacityConfig {
    small_queue_max_cap: i64,
    main_queue_max_cap: i64,
    cache_cap_to_total_data_size_ratio: f64,
    shrink_step: ReadableSize,
}

#[derive(Clone)]
pub struct IaManager {
    core: Arc<IaManagerCore>,
}

impl ops::Deref for IaManager {
    type Target = IaManagerCore;

    fn deref(&self) -> &Self::Target {
        &self.core
    }
}

impl IaManager {
    pub fn new(
        opts: IaManagerOptions,
        fs: Arc<dyn dfs::Dfs>,
        meta_fd_cache: Option<FdCache>,
        runtime: WorkerPool,
    ) -> Result<Self> {
        validate_cache_capacity_ratio(opts.cache_cap_to_total_data_size_ratio)?;
        validate_cache_capacity_shrink_step(opts.cache_capacity_shrink_step)?;
        assert!(
            opts.small_queue.paths.is_empty(),
            "small queue must be in memory"
        );

        info!("create IA manager"; "opts" => ?opts);
        let io_limiter = IaIoRateLimiter::new(
            opts.remote_read_max_throughput,
            opts.local_write_max_throughput,
            opts.io_rate_limiter_refill_period,
        );
        let transition_limiter = TransitionLimiter::new(
            opts.transition_max_throughput,
            opts.transition_max_regions,
            opts.transition_wait_timeout,
        );
        let main_store = new_local_store(
            opts.main_queue.paths.clone(),
            opts.fd_cache_capacity,
            io_limiter.clone(),
        );
        let segments = Arc::new(LocalSegmentMap::default());
        let segment_data_ctx = SegmentDataContext {
            segments: segments.clone(),
            main_store: main_store.clone(),
        };

        let small_queue_cap = Arc::new(AtomicI64::new(opts.small_queue_min_cap()));
        let main_queue_cap = Arc::new(AtomicI64::new(opts.main_queue_min_cap()));
        let fifo = S3FifoHandle::new(
            small_queue_cap.clone(),
            main_queue_cap.clone(),
            opts.segment_size,
            opts.freq_update_interval,
            runtime.handle(),
            segment_data_ctx,
        );

        let dfs_concurrency = Arc::new(Semaphore::new(opts.dfs_concurrency));
        let local_segment_memory_cap = IaManagerCore::calc_local_segment_memory_cap(&opts)?;
        let local_segment_memory_limiter =
            Arc::new(Semaphore::new(local_segment_memory_cap as usize));
        let small_queue_max_cap = opts.small_queue.cap;
        let main_queue_max_cap = opts.main_queue.cap;
        let cache_capacity_config = CacheCapacityConfig {
            small_queue_max_cap,
            main_queue_max_cap,
            cache_cap_to_total_data_size_ratio: opts.cache_cap_to_total_data_size_ratio,
            shrink_step: opts.cache_capacity_shrink_step,
        };

        let core = Arc::new(IaManagerCore {
            opts,
            fs,
            runtime,
            main_store,
            meta_fd_cache,
            loading_segments: Default::default(),
            segments,
            table_metas: Default::default(),
            cache_capacity_config: ArcSwap::from_pointee(cache_capacity_config),
            small_queue_cap,
            main_queue_cap,
            fifo,
            dfs_concurrency,
            dfs_keyspace_concurrency: Default::default(),
            local_segment_memory_limiter,
            local_segment_memory_cap: AtomicI64::new(local_segment_memory_cap),
            io_limiter,
            transition_limiter,
            last_persist_manifest_time: AtomicI64::new(Instant::now_coarse().second()),
            sync_read_counter: Arc::new(()),
        });

        let mgr = Self { core };
        mgr.init()?;
        Ok(mgr)
    }

    /// Publish IA cache queue capacity limits for the next data-size report.
    pub fn update_cache_capacity(
        &self,
        mem_cap: AbsoluteOrPercentSize,
        disk_cap: AbsoluteOrPercentSize,
    ) -> Result<()> {
        self.core.update_cache_capacity(mem_cap, disk_cap)
    }

    pub(crate) fn update_cache_capacity_config(
        &self,
        mem_cap: Option<AbsoluteOrPercentSize>,
        disk_cap: Option<AbsoluteOrPercentSize>,
        cache_cap_to_total_data_size_ratio: Option<f64>,
        shrink_step: Option<ReadableSize>,
    ) -> Result<()> {
        self.core.update_cache_capacity_config(
            mem_cap,
            disk_cap,
            cache_cap_to_total_data_size_ratio,
            shrink_step,
        )
    }

    #[cfg(test)]
    pub(crate) fn cache_queue_capacities(&self) -> (i64, i64) {
        (
            self.core.small_queue_cap.load(Ordering::Relaxed),
            self.core.main_queue_cap.load(Ordering::Relaxed),
        )
    }
}

pub struct IaManagerCore {
    opts: IaManagerOptions,
    fs: Arc<dyn dfs::Dfs>,
    runtime: WorkerPool,
    main_store: Arc<dyn LocalStore>,
    meta_fd_cache: Option<FdCache>,

    loading_segments: GuardMap<FileSegmentIdent, ()>,
    segments: Arc<LocalSegmentMap>,

    table_metas: Arc<DashMap<u64 /* file_id */, TableMetaInfo>>,

    // Online updates publish a complete desired-capacity snapshot. The
    // periodic data-size reporter is the sole runtime writer of live caps.
    cache_capacity_config: ArcSwap<CacheCapacityConfig>,

    // Live S3-FIFO queue capacities after periodic reconciliation.
    small_queue_cap: Arc<AtomicI64>,
    main_queue_cap: Arc<AtomicI64>,
    fifo: S3FifoHandle,

    dfs_concurrency: Arc<Semaphore>,
    dfs_keyspace_concurrency: Arc<DashMap<u32 /* keyspace_id */, Arc<Semaphore>>>,
    local_segment_memory_limiter: Arc<Semaphore>,
    // Monotonic high-water mark of the memory admission limiter. Queue trims
    // enforce lower live caps without trying to revoke permits already in use.
    local_segment_memory_cap: AtomicI64,
    io_limiter: IaIoRateLimiter,
    transition_limiter: TransitionLimiter,

    last_persist_manifest_time: AtomicI64,

    sync_read_counter: Arc<()>,
}

impl IaManagerCore {
    pub fn enter_runtime(&self) -> tokio::runtime::EnterGuard<'_> {
        self.runtime.enter()
    }

    pub(crate) fn runtime_handle(&self) -> WorkerPoolHandle {
        self.runtime.handle()
    }

    pub fn get_dfs(&self) -> &dyn Dfs {
        self.fs.deref()
    }

    pub fn get_meta_fd_cache(&self) -> Option<FdCache> {
        self.meta_fd_cache.clone()
    }

    pub(crate) fn segment_size(&self) -> i64 {
        self.opts.segment_size
    }

    fn init(&self) -> Result<()> {
        if let Some(path) = self.main_store.main_paths().first() {
            match Manifest::read_from_path(path) {
                Ok(Some(manifest)) => {
                    self.init_from_manifest(&manifest);
                }
                Ok(None) => {}
                Err(err) => warn!("open manifest failed: {:?}", err),
            }
        }

        self.main_store.init()?;
        let mut entries = self.main_store.scan()?;
        if let Some(segments) = entries.remove("seg") {
            self.init_segments(segments)?;
        }

        ENGINE_IA_MAIN_QUEUE_CAPACITY.set(self.main_queue_cap.load(Ordering::Relaxed));
        ENGINE_IA_SMALL_QUEUE_CAPACITY.set(self.small_queue_cap.load(Ordering::Relaxed));
        ENGINE_IA_MANAGER_SEGMENTS_MEMORY_CAPACITY.set(self.local_segment_memory_cap());

        Ok(())
    }

    fn init_from_manifest(&self, manifest: &Manifest) {
        info!("init from manifest: {:?}", manifest);
        if self.opts.dynamic_capacity {
            self.adjust_queue_cap(manifest.main_queue_cap);
        }
    }

    // TODO: take snapshot for queue & restore from it.
    fn init_segments(&self, keys: Vec<String>) -> Result<()> {
        for k in keys {
            if let Some(ident) = FileSegmentIdent::parse_local_filename(&k) {
                self.segments
                    .set_segment_data(ident, FileSegmentData::InStore);
                self.fifo.read(ident, true)?;
            }
        }
        Ok(())
    }

    /// Offset in `read_at` are absolute offsets of the file.
    pub(crate) async fn read_segment(
        &self,
        ident: FileSegmentIdent,
        ftype: FileType,
        keyspace_id: Option<u32>,
        deadline: Option<Deadline>,
        read_at: &mut ReadAt<'_>,
        dfs: Option<&dyn Dfs>,
        charge_ia_remote_read: bool,
    ) -> Result<()> {
        let (start_off, end_off) = (read_at.start_off(), read_at.end_off());
        if start_off > end_off || start_off < ident.start_off || ident.end_off < end_off {
            debug_assert!(
                false,
                "invalid range, segment {:?}, range: {}-{}",
                ident, start_off, end_off
            );
            return Err(Error::IaMgr(format!(
                "invalid range, ident: {:?}, range: {}-{}",
                ident, start_off, end_off
            )));
        }

        let collect_ia_remote_read_stats = ReadSegmentStatsCollector::new(ident);
        debug!("read segment"; "ident" => %ident, "start_off" => start_off, "end_off" => end_off);
        if let Some(()) = self.read_segment_from_cache(ident, read_at)? {
            collect_ia_remote_read_stats.hit();
            return Ok(());
        }
        // This is the admission boundary for a new remote read. Once it
        // passes, an online shrink does not cancel it; S3-FIFO will trim the
        // fetched item against the latest live cap after insertion.
        self.validate_new_segment_size(ident)?;

        let data = {
            let _loading_guard = self.loading_segments.get_locked(ident).await;

            // Check cache again. Another thread may have filled the cache.
            if let Some(()) = self.read_segment_from_cache(ident, read_at)? {
                collect_ia_remote_read_stats.hit();
                // Not necessary to remove ident from `loading_segments`, as there must be
                // another concurrent request read the segment from remote.
                return Ok(());
            }

            let (data, memory_permit) = self
                .read_segment_from_remote(
                    ident,
                    ftype,
                    keyspace_id,
                    deadline,
                    dfs,
                    charge_ia_remote_read,
                )
                .await?;
            let admission_result = self.admit_remote_segment(ident, data.clone(), memory_permit);
            self.loading_segments.remove(&ident);
            admission_result?;

            data
        };

        read_at.read_from_segment_bytes(ident, &data);
        collect_ia_remote_read_stats.miss();
        Ok(())
    }

    fn validate_new_segment_size(&self, ident: FileSegmentIdent) -> Result<()> {
        let small_queue_cap = self.small_queue_cap.load(Ordering::Relaxed);
        if small_queue_cap > 0 && ident.size() > small_queue_cap as u64 {
            return Err(Error::IaMgr(format!(
                "IA cache item exceeds live memory capacity: ident: {ident:?}, size: {}, capacity: {small_queue_cap}",
                ident.size()
            )));
        }
        Ok(())
    }

    fn admit_remote_segment(
        &self,
        ident: FileSegmentIdent,
        data: Bytes,
        memory_permit: OwnedSemaphorePermit,
    ) -> Result<()> {
        self.segments
            .set_segment_data(ident, FileSegmentData::in_memory(data, memory_permit));
        if let Err(err) = self.fifo.read(ident, false) {
            self.segments.remove(ident);
            return Err(err);
        }
        Ok(())
    }

    async fn read_segment_from_remote(
        &self,
        ident: FileSegmentIdent,
        ftype: FileType,
        keyspace_id: Option<u32>,
        deadline: Option<Deadline>,
        dfs: Option<&dyn Dfs>,
        charge_ia_remote_read: bool,
    ) -> Result<(Bytes, OwnedSemaphorePermit)> {
        let _permit = self.acquire_concurrency_permit(keyspace_id).await;
        if deadline.is_some_and(|d| d.check().is_err()) {
            return Err(Error::DeadlineExceeded(format!(
                "acquire concurrency permit timeout: {ident}"
            )));
        }

        // Throttle before reserving remote-read memory so rate-limit waits do
        // not hold memory admission permits.
        let remote_read_charge = if charge_ia_remote_read {
            let charge = self.io_limiter.throttle_remote_read(ident.size()).await;
            if deadline.is_some_and(|d| d.check().is_err()) {
                return Err(Error::DeadlineExceeded(format!(
                    "throttle remote read timeout: {ident}"
                )));
            }
            Some(charge)
        } else {
            None
        };

        // Reserve memory before remote IO starts. The permit is moved into the
        // InMem segment, so concurrent reads cannot accumulate unbounded Bytes
        // while waiting for S3-FIFO to move data into the main store.
        let memory_permit = self.acquire_remote_read_memory(ident).await?;
        // Keep the same deadline semantics as `acquire_concurrency_permit`: the
        // wait itself is not wrapped in a timeout; callers observe deadline
        // expiration at the next checkpoint after the permit is acquired.
        if deadline.is_some_and(|d| d.check().is_err()) {
            return Err(Error::DeadlineExceeded(format!(
                "acquire remote read memory timeout: {ident}"
            )));
        }
        let opts = dfs::Options::default()
            .with_type(ftype)
            .with_start_off(ident.start_off)
            .with_end_off(ident.end_off);
        let data = if let Some(dfs) = dfs {
            dfs.read_file(ident.file_id, opts).await?
        } else {
            let dfs = self.fs.clone();
            let runtime = dfs.get_runtime().handle().clone();
            runtime
                .spawn(async move { dfs.read_file(ident.file_id, opts).await })
                .await
                .map_err(|err| Error::Other(err.to_string()))??
        };
        if let Some(remote_read_charge) = remote_read_charge {
            remote_read_charge.commit();
        }
        Ok((data, memory_permit))
    }

    async fn acquire_remote_read_memory(
        &self,
        ident: FileSegmentIdent,
    ) -> Result<OwnedSemaphorePermit> {
        let segment_bytes = ident.size();
        let permits = self.local_segment_memory_permits_for_bytes(segment_bytes)?;

        let wait_start = Instant::now_coarse();
        let permit = self
            .local_segment_memory_limiter
            .clone()
            .acquire_many_owned(permits)
            .await
            .map_err(|err| Error::IaMgr(format!("IA segment memory limiter closed: {err:?}")))?;

        ENGINE_IA_MANAGER_SEGMENTS_MEMORY_WAIT_DURATION
            .observe(wait_start.saturating_elapsed().as_secs_f64());
        Ok(permit)
    }

    fn calc_local_segment_memory_cap(opts: &IaManagerOptions) -> Result<i64> {
        Self::calc_local_segment_memory_cap_for_small_queue(opts, opts.small_queue.cap)
    }

    fn calc_local_segment_memory_cap_for_small_queue(
        opts: &IaManagerOptions,
        small_queue_cap: i64,
    ) -> Result<i64> {
        // Small-queue entries are memory-backed. Keep enough headroom for at
        // least one largest legal item even when the queue is full; otherwise
        // that item could wait forever because only its eventual FIFO insert
        // would trigger eviction. The fixed concurrency buffer remains useful
        // for ordinary segment-sized reads when it is larger than one queue.
        let small_queue_cap = i128::from(small_queue_cap);
        let fixed_buffer = i128::from(opts.segment_size)
            * opts.dfs_concurrency as i128
            * i128::from(REMOTE_READ_MEMORY_BUFFER_SEGMENT_FACTOR);
        let cap = small_queue_cap + small_queue_cap.max(fixed_buffer);
        if cap <= 0 || cap > i128::from(i64::MAX) || cap > Semaphore::MAX_PERMITS as i128 {
            return Err(Error::IaMgr(format!(
                "IA segment memory admission capacity is out of range: {cap}, limit: {}",
                Semaphore::MAX_PERMITS
            )));
        }
        Ok(cap as i64)
    }

    #[inline]
    fn local_segment_memory_cap(&self) -> i64 {
        self.local_segment_memory_cap.load(Ordering::Relaxed)
    }

    // Tokio semaphores support adding permits. Shrinking would need to revoke
    // permits that may already be held, so the admission limit intentionally
    // stays at its high-water mark while the queue trim enforces the live cap.
    // TODO: Replace this grow-only semaphore with a resizable weighted limiter
    // so online shrink can also lower the memory admission limit.
    fn grow_local_segment_memory_cap(&self, small_queue_cap: i64) -> Result<()> {
        let new_cap =
            Self::calc_local_segment_memory_cap_for_small_queue(&self.opts, small_queue_cap)?;
        let old_cap = self.local_segment_memory_cap();
        if new_cap <= old_cap {
            return Ok(());
        }

        self.local_segment_memory_limiter
            .add_permits((new_cap - old_cap) as usize);
        self.local_segment_memory_cap
            .store(new_cap, Ordering::Relaxed);
        ENGINE_IA_MANAGER_SEGMENTS_MEMORY_CAPACITY.set(new_cap);
        info!(
            "grow IA segment memory admission capacity: {}->{}",
            old_cap, new_cap
        );
        Ok(())
    }

    fn local_segment_memory_permits_for_bytes(&self, bytes: u64) -> Result<u32> {
        let local_segment_memory_cap = self.local_segment_memory_cap();
        if bytes > local_segment_memory_cap as u64 {
            return Err(Error::IaMgr(format!(
                "IA segment exceeds memory admission cap: size: {bytes}, cap: {}",
                local_segment_memory_cap
            )));
        }
        u32::try_from(bytes).map_err(|_| {
            Error::IaMgr(format!(
                "IA segment exceeds memory admission permit limit: size: {bytes}, permit_limit: {}",
                u32::MAX
            ))
        })
    }

    fn read_segment_from_cache(
        &self,
        ident: FileSegmentIdent,
        read_at: &mut ReadAt<'_>,
    ) -> Result<Option<()>> {
        let segment_data = try_some!(self.segments.get_segment(ident));
        let res = match &segment_data {
            FileSegmentData::InMem(data) => {
                read_at.read_from_segment_bytes(ident, data.bytes());
                Some(())
            }
            FileSegmentData::InStore => self.read_segment_from_local_store(ident, read_at)?,
        };

        if res.is_some() {
            self.fifo.read(ident, false)?;
        }
        Ok(res)
    }

    // Note: when `read_at.buf` is empty, the existence of segment is not checked.
    fn read_segment_from_local_store(
        &self,
        ident: FileSegmentIdent,
        read_at: &mut ReadAt<'_>,
    ) -> Result<Option<()>> {
        debug!("read segment from local"; "ident" => %ident);
        let local_filename = ident.local_filename();
        if read_at.need_segment_handle {
            match self.main_store.handle(ident.file_id, &local_filename)? {
                Some(handle) => read_at.set_handle(handle),
                None => return Ok(None),
            }
        }
        if read_at.buf.is_empty() {
            return Ok(Some(()));
        }
        self.main_store.read_at(
            ident.file_id,
            &local_filename,
            read_at.buf,
            read_at.offset - ident.start_off,
        )
    }

    // Note: It's possible that the segment is of `SegmentData::InStore` but
    // actually not existed. In such condition this method will still return
    // true.
    pub fn is_segment_cached(&self, ident: FileSegmentIdent) -> bool {
        self.segments.get_segment(ident).is_some()
    }

    pub fn segment_cached_position(&self, ident: FileSegmentIdent) -> FileSegmentPosition {
        match self.segments.get_segment(ident) {
            Some(FileSegmentData::InMem(_)) => FileSegmentPosition::InMem,
            Some(FileSegmentData::InStore) => FileSegmentPosition::InStore,
            None => FileSegmentPosition::NotExist,
        }
    }

    pub async fn prefetch_segment(
        &self,
        ident: FileSegmentIdent,
        ftype: FileType,
        keyspace_id: u32,
        deadline: Deadline,
        dfs: Option<&dyn Dfs>,
        charge_ia_remote_read: bool,
    ) -> Result<()> {
        let mut buf: [u8; 0] = [];
        let mut read_at = ReadAt::new(buf.as_mut_slice(), ident.start_off, false);
        self.read_segment(
            ident,
            ftype,
            Some(keyspace_id),
            Some(deadline),
            &mut read_at,
            dfs,
            charge_ia_remote_read,
        )
        .await
    }

    pub async fn get_segment_handle(
        &self,
        ident: FileSegmentIdent,
        ftype: FileType,
        dfs: Option<&dyn Dfs>,
        charge_ia_remote_read: bool,
    ) -> Result<SegmentHandle> {
        let mut buf: [u8; 0] = [];
        let mut read_at = ReadAt::new(buf.as_mut_slice(), ident.start_off, true);
        self.read_segment(
            ident,
            ftype,
            None,
            None,
            &mut read_at,
            dfs,
            charge_ia_remote_read,
        )
        .await?;
        let handle = read_at.segment_handle.unwrap();
        debug!("get segment handle"; "ident" => ?ident, "ftype" => ?ftype, "handle" => ?handle);
        Ok(handle)
    }

    pub async fn acquire_concurrency_permit(
        &self,
        keyspace_id: Option<u32>,
    ) -> (Option<OwnedSemaphorePermit>, OwnedSemaphorePermit) {
        let keyspace_permit = match keyspace_id {
            Some(keyspace_id) => {
                let sem = self
                    .dfs_keyspace_concurrency
                    .entry(keyspace_id)
                    .or_insert_with(|| Arc::new(Semaphore::new(self.opts.dfs_keyspace_concurrency)))
                    .clone();
                Some(sem.acquire_owned().await.unwrap())
            }
            None => None,
        };
        let global_permit = self.dfs_concurrency.clone().acquire_owned().await.unwrap();
        (keyspace_permit, global_permit)
    }

    pub fn cache_hit_rate(&self) -> f64 {
        let hit = ENGINE_IA_REMOTE_READ_SEGMENT_CACHE_HIT.get();
        let miss = ENGINE_IA_REMOTE_READ_SEGMENT_CACHE_MISS.get();
        let total = hit + miss;
        if total > 0 {
            hit as f64 / total as f64
        } else {
            0.0
        }
    }

    pub fn contains_segment(&self, ident: FileSegmentIdent) -> bool {
        self.segments.contains(ident)
    }

    pub fn main_store_paths(&self) -> &[PathBuf] {
        self.main_store.main_paths()
    }

    pub fn access_table_meta(&self, file_id: u64) -> bool /* should_set_mtime */ {
        match self.table_metas.entry(file_id) {
            dashmap::mapref::entry::Entry::Vacant(e) => {
                e.insert(TableMetaInfo::default());
                // Return false because:
                // The cop worker on spot instances will restart frequently for a short while.
                // If return true here, we will set mtime more than expected.
                // On the other hand, as the lifetime of table meta are much longer than
                // interval (24h vs. 1h by default), miss one or two times of
                // set mtime will not impact too much.
                false
            }
            dashmap::mapref::entry::Entry::Occupied(mut e) => {
                let meta = e.get_mut();
                let now = Instant::now_coarse();
                let last_set_mtime =
                    Instant::from_timespec_second_coarse(meta.last_set_mtime_instant_sec);
                if now.saturating_duration_since(last_set_mtime)
                    >= self.opts.table_meta_mtime_interval
                {
                    meta.last_set_mtime_instant_sec = now.second();
                    true
                } else {
                    false
                }
            }
        }
    }

    pub fn remove_table_meta(&self, file_id: u64) {
        self.table_metas.remove(&file_id);
        if let Some(cache) = self.meta_fd_cache.as_ref() {
            cache.remove(file_id)
        }
    }

    /// Reconcile live queue capacities from a periodic IA data-size report.
    ///
    /// Fixed-capacity managers ignore the data size and configured ratio and
    /// target the configured memory and disk capacity limits.
    pub fn notify_total_data_size(&self, total_data_size: u64) {
        let config = self.cache_capacity_config.load();
        let desired_main_queue_cap = Self::desired_main_queue_cap(
            total_data_size,
            config.cache_cap_to_total_data_size_ratio,
        );
        if let Err(err) = self.apply_cache_capacity(&config, desired_main_queue_cap) {
            error!(
                "reconcile IA cache capacity failed";
                "err" => ?err,
                "total_data_size" => total_data_size,
                "config" => ?*config,
            );
            return;
        }
        self.persist_manifest();
    }

    /// Publish IA cache queue capacity limits for the next data-size report.
    pub fn update_cache_capacity(
        &self,
        mem_cap: AbsoluteOrPercentSize,
        disk_cap: AbsoluteOrPercentSize,
    ) -> Result<()> {
        self.update_cache_capacity_config(Some(mem_cap), Some(disk_cap), None, None)
    }

    fn update_cache_capacity_config(
        &self,
        mem_cap: Option<AbsoluteOrPercentSize>,
        disk_cap: Option<AbsoluteOrPercentSize>,
        cache_cap_to_total_data_size_ratio: Option<f64>,
        shrink_step: Option<ReadableSize>,
    ) -> Result<()> {
        let small_queue_max_cap = mem_cap
            .map(|cap| Self::cache_cap_to_i64("ia.mem-cap", cap.as_memory_size()))
            .transpose()?;
        let main_queue_max_cap = disk_cap
            .map(|cap| {
                let cap = cap
                    .as_disks_size(self.main_store_paths())
                    .map_err(|err| Error::Io(format!("get disks capacity failed: {err:?}")))?;
                Self::cache_cap_to_i64("ia.disk-cap", cap)
            })
            .transpose()?;
        self.publish_cache_capacity_config(
            small_queue_max_cap,
            main_queue_max_cap,
            cache_cap_to_total_data_size_ratio,
            shrink_step,
        )
    }

    fn publish_cache_capacity_config(
        &self,
        small_queue_max_cap: Option<i64>,
        main_queue_max_cap: Option<i64>,
        cache_cap_to_total_data_size_ratio: Option<f64>,
        shrink_step: Option<ReadableSize>,
    ) -> Result<()> {
        if let Some(ratio) = cache_cap_to_total_data_size_ratio {
            validate_cache_capacity_ratio(ratio)?;
        }
        if let Some(step) = shrink_step {
            validate_cache_capacity_shrink_step(step)?;
        }
        if small_queue_max_cap.is_some_and(|cap| cap <= 0)
            || main_queue_max_cap.is_some_and(|cap| cap <= 0)
        {
            return Err(Error::IaMgr(format!(
                "IA cache capacity must be positive, small: {small_queue_max_cap:?}, main: {main_queue_max_cap:?}"
            )));
        }
        if let Some(cap) = small_queue_max_cap {
            Self::calc_local_segment_memory_cap_for_small_queue(&self.opts, cap)?;
        }

        let old_config = self
            .cache_capacity_config
            .rcu(|current| CacheCapacityConfig {
                small_queue_max_cap: small_queue_max_cap.unwrap_or(current.small_queue_max_cap),
                main_queue_max_cap: main_queue_max_cap.unwrap_or(current.main_queue_max_cap),
                cache_cap_to_total_data_size_ratio: cache_cap_to_total_data_size_ratio
                    .unwrap_or(current.cache_cap_to_total_data_size_ratio),
                shrink_step: shrink_step.unwrap_or(current.shrink_step),
            });
        let new_config = CacheCapacityConfig {
            small_queue_max_cap: small_queue_max_cap.unwrap_or(old_config.small_queue_max_cap),
            main_queue_max_cap: main_queue_max_cap.unwrap_or(old_config.main_queue_max_cap),
            cache_cap_to_total_data_size_ratio: cache_cap_to_total_data_size_ratio
                .unwrap_or(old_config.cache_cap_to_total_data_size_ratio),
            shrink_step: shrink_step.unwrap_or(old_config.shrink_step),
        };
        info!(
            "publish desired IA cache capacity config";
            "old" => ?old_config,
            "new" => ?new_config,
        );
        Ok(())
    }

    #[inline]
    fn desired_main_queue_cap(total_data_size: u64, ratio: f64) -> i64 {
        let ratio_units = (ratio * CACHE_CAPACITY_RATIO_SCALE as f64).floor() as u128;
        let desired_gib = u128::from(total_data_size).saturating_mul(ratio_units)
            / (GIB as u128 * CACHE_CAPACITY_RATIO_SCALE);
        let desired_gib = desired_gib.min(i64::MAX as u128 / CACHE_CAPACITY_GRANULARITY as u128);
        (desired_gib * CACHE_CAPACITY_GRANULARITY as u128) as i64
    }

    fn cache_cap_to_i64(name: &str, cap: u64) -> Result<i64> {
        let cap = i64::try_from(cap)
            .map_err(|_| Error::IaMgr(format!("{name} exceeds i64::MAX: {cap}")))?;
        if cap <= 0 {
            return Err(Error::IaMgr(format!("{name} must be positive: {cap}")));
        }
        Ok(cap)
    }

    fn apply_cache_capacity(
        &self,
        config: &CacheCapacityConfig,
        desired_main_queue_cap: i64,
    ) -> Result<()> {
        // Grow admission before exposing a larger live small-queue cap. The
        // admission high-water mark is intentionally not reduced on shrink.
        self.grow_local_segment_memory_cap(config.small_queue_max_cap)?;
        let (target_small_cap, target_main_cap) = self.calc_queue_caps(
            desired_main_queue_cap,
            config.small_queue_max_cap,
            config.main_queue_max_cap,
        );
        let current_small_cap = self.small_queue_cap.load(Ordering::Relaxed);
        let current_main_cap = self.main_queue_cap.load(Ordering::Relaxed);
        let main_shrink_step = Self::shrink_step_as_i64(config.shrink_step);
        let small_shrink_step = self.small_queue_shrink_step(config);
        let next_small_cap =
            Self::next_queue_cap(current_small_cap, target_small_cap, small_shrink_step);
        let next_main_cap =
            Self::next_queue_cap(current_main_cap, target_main_cap, main_shrink_step);
        self.set_queue_caps(next_small_cap, next_main_cap);
        Ok(())
    }

    #[inline]
    fn shrink_step_as_i64(step: ReadableSize) -> i64 {
        i64::try_from(step.0).unwrap_or(i64::MAX)
    }

    fn small_queue_shrink_step(&self, config: &CacheCapacityConfig) -> i64 {
        let main_shrink_step = Self::shrink_step_as_i64(config.shrink_step);
        if config.small_queue_max_cap <= 0 || config.main_queue_max_cap <= 0 {
            return main_shrink_step;
        }

        let numerator = u128::from(config.shrink_step.0) * config.small_queue_max_cap as u128;
        let denominator = config.main_queue_max_cap as u128;
        let proportional_step = numerator.saturating_add(denominator - 1) / denominator;
        let proportional_step = proportional_step.min(i64::MAX as u128) as i64;
        proportional_step.max(self.opts.segment_size)
    }

    #[inline]
    fn next_queue_cap(current_cap: i64, target_cap: i64, shrink_step: i64) -> i64 {
        if target_cap >= current_cap {
            target_cap
        } else {
            current_cap.saturating_sub(shrink_step).max(target_cap)
        }
    }

    fn adjust_queue_cap(&self, desired_main_queue_cap: i64) {
        let config = self.cache_capacity_config.load();
        self.adjust_queue_cap_with_config(desired_main_queue_cap, &config);
    }

    fn adjust_queue_cap_with_config(
        &self,
        desired_main_queue_cap: i64,
        config: &CacheCapacityConfig,
    ) {
        let (small_cap, main_cap) = self.calc_queue_caps(
            desired_main_queue_cap,
            config.small_queue_max_cap,
            config.main_queue_max_cap,
        );
        self.set_queue_caps(small_cap, main_cap);
    }

    fn set_queue_caps(&self, small_cap: i64, main_cap: i64) {
        let old_main_cap = self.main_queue_cap.swap(main_cap, Ordering::Relaxed);
        let old_small_cap = self.small_queue_cap.swap(small_cap, Ordering::Relaxed);

        if old_main_cap != main_cap || old_small_cap != small_cap {
            if main_cap < old_main_cap || small_cap < old_small_cap {
                if let Err(err) = self.fifo.trim_to_capacity() {
                    error!("schedule queue capacity trim failed"; "err" => ?err);
                }
            }
            ENGINE_IA_MAIN_QUEUE_CAPACITY.set(main_cap);
            ENGINE_IA_SMALL_QUEUE_CAPACITY.set(small_cap);
            info!(
                "adjust queue cap: small: {}->{}, main: {}->{}",
                old_small_cap, small_cap, old_main_cap, main_cap
            );
        }
    }

    fn calc_queue_caps(
        &self,
        desired_main_queue_cap: i64,
        small_queue_max_cap: i64,
        main_queue_max_cap: i64,
    ) -> (i64, i64) {
        if !self.opts.dynamic_capacity {
            return (small_queue_max_cap, main_queue_max_cap);
        }

        let main_cap = desired_main_queue_cap.clamp(
            self.main_queue_min_cap(main_queue_max_cap),
            main_queue_max_cap,
        );

        // Should use f64. Otherwise, overflow may happen.
        let small_cap =
            (main_cap as f64 / main_queue_max_cap as f64 * small_queue_max_cap as f64) as i64;
        let small_cap = small_cap.clamp(
            self.small_queue_min_cap(small_queue_max_cap),
            small_queue_max_cap,
        );
        (small_cap, main_cap)
    }

    #[cfg(test)]
    fn apply_queue_cap_limits_for_test(
        &self,
        small_queue_max_cap: i64,
        main_queue_max_cap: i64,
    ) -> Result<()> {
        self.publish_cache_capacity_config(
            Some(small_queue_max_cap),
            Some(main_queue_max_cap),
            None,
            None,
        )?;
        let config = self.cache_capacity_config.load();
        self.apply_cache_capacity(&config, main_queue_max_cap)
    }

    #[inline]
    fn small_queue_min_cap(&self, max_cap: i64) -> i64 {
        small_queue_capacity_floor(self.opts.segment_size).min(max_cap)
    }

    #[inline]
    fn main_queue_min_cap(&self, max_cap: i64) -> i64 {
        (self.opts.segment_size * MAIN_QUEUE_MIN_SEGMENT_COUNT).min(max_cap)
    }

    fn persist_manifest(&self) {
        let Some(path) = self.main_store.main_paths().first() else {
            return;
        };

        let last_time_secs = self.last_persist_manifest_time.load(Ordering::Relaxed);
        if Instant::from_timespec_second_coarse(last_time_secs).saturating_elapsed()
            < Self::manifest_persist_interval()
        {
            return;
        }
        if self
            .last_persist_manifest_time
            .compare_exchange_weak(
                last_time_secs,
                Instant::now_coarse().second(),
                Ordering::Relaxed,
                Ordering::Relaxed,
            )
            .is_err()
        {
            return;
        }

        let main_queue_cap = self.main_queue_cap.load(Ordering::Relaxed);
        let manifest = Manifest { main_queue_cap };
        if let Err(err) = manifest.persist_to_path(path) {
            warn!("persist manifest failed"; "err" => ?err);
        }
    }

    pub(crate) fn acquire_sync_read(&self) -> Result<(Arc<()>, Duration)> {
        if self.opts.disable_sync_read {
            return Err(Error::IaMgr("sync read disabled".to_string()));
        }
        let guard = self.sync_read_counter.clone();
        if Arc::strong_count(&guard) > self.opts.sync_read_concurrency + 1 {
            Err(Error::IaMgr("sync read limit exceeded".to_string()))
        } else {
            Ok((guard, self.opts.sync_read_timeout))
        }
    }

    pub fn update_io_rate_limits(
        &self,
        remote_read_max_throughput: ReadableSize,
        local_write_max_throughput: ReadableSize,
    ) {
        self.io_limiter
            .update(remote_read_max_throughput, local_write_max_throughput);
    }

    pub fn update_remote_read_rate_limit(&self, remote_read_max_throughput: ReadableSize) {
        self.io_limiter
            .update_remote_read(remote_read_max_throughput);
    }

    pub fn update_local_write_rate_limit(&self, local_write_max_throughput: ReadableSize) {
        self.io_limiter
            .update_local_write(local_write_max_throughput);
    }

    /// Returns the per-store limiter used by storage-class transition
    /// materialization.
    pub fn transition_limiter(&self) -> TransitionLimiter {
        self.transition_limiter.clone()
    }

    #[inline]
    fn manifest_persist_interval() -> Duration {
        if cfg!(debug_assertion) {
            return Duration::from_secs(3);
        }
        MANIFEST_PERSIST_INTERVAL
    }
}

#[cfg(any(test, feature = "testexport"))]
impl IaManager {
    // Flush all tasks and wait for completion.
    pub async fn flush_tasks(&self, timeout: Duration) -> Result<()> {
        let start_time = Instant::now_coarse();
        while start_time.saturating_elapsed() < timeout {
            let pending_tasks = self.fifo.flush_tasks().await?;
            if pending_tasks == 0 {
                return Ok(());
            }
            tokio::time::sleep(Duration::from_millis(100)).await;
        }
        Err(Error::IaMgr("wait for pending tasks timeout".to_string()))
    }

    pub async fn get_local_segments(
        &self,
    ) -> Vec<(
        FileSegmentIdent,
        crate::ia::types::FileSegmentData,
        Option<crate::ia::queue::QueueItem>,
    )> {
        let local_segments = self.segments.get_all();
        let mut segments = Vec::with_capacity(local_segments.len());
        // TODO: find segments in queue but not in local store.
        for (ident, segment) in local_segments {
            let queue_item = self.fifo.get_item(ident).await.unwrap();
            segments.push((ident, segment, queue_item));
        }
        segments
    }

    /// Returns bytes committed to the IA remote-read limiter.
    pub fn remote_read_total_bytes_consumed(&self) -> usize {
        self.io_limiter.remote_read_total_bytes_consumed()
    }
}

#[cfg(test)]
impl SegmentDataContext {
    pub(crate) fn set_segment_data_for_test(
        &self,
        ident: FileSegmentIdent,
        segment_data: FileSegmentData,
    ) -> Option<FileSegmentData> {
        self.segments.set_segment_data(ident, segment_data)
    }
}

#[derive(Clone)]
pub(crate) struct SegmentDataContext {
    segments: Arc<LocalSegmentMap>,
    main_store: Arc<dyn LocalStore>,
}

impl SegmentDataContext {
    #[inline]
    pub(crate) fn set_segment_data_from_mem_to_store(
        &self,
        ident: FileSegmentIdent,
    ) -> std::result::Result<(), Option<FileSegmentData>> {
        self.segments.set_segment_data_from_mem_to_store(ident)
    }

    #[inline]
    pub(crate) fn get_segment_data(&self, ident: FileSegmentIdent) -> Option<FileSegmentData> {
        self.segments.get_segment(ident)
    }

    #[inline]
    pub(crate) fn remove_segment_data(&self, ident: FileSegmentIdent) -> Option<FileSegmentData> {
        self.segments.remove(ident)
    }

    #[inline]
    pub(crate) fn remove_from_main_store(&self, ident: FileSegmentIdent) -> Result<Option<()>> {
        self.main_store
            .remove(ident.file_id, &ident.local_filename())
    }

    #[inline]
    pub(crate) fn save_to_main_store(&self, ident: FileSegmentIdent, bytes: Bytes) -> Result<()> {
        self.main_store
            .save(ident.file_id, &ident.local_filename(), bytes)
    }

    #[cfg(test)]
    pub(crate) fn new_for_test() -> Self {
        Self {
            main_store: Arc::new(crate::ia::util::LocalMemoryStore::default()),
            segments: Arc::new(LocalSegmentMap::default()),
        }
    }
}

const MANIFEST_FORMAT_VER: u8 = 1;
const MANIFEST_FILE_NAME: &str = "manifest";

#[derive(Debug, PartialEq)]
struct Manifest {
    main_queue_cap: i64,
    // TODO: main queue snapshot
}

impl Manifest {
    fn marshal(&self) -> Vec<u8> {
        let mut buf = Vec::with_capacity(U8_SIZE /* ver */ + I64_SIZE /* main_queue_cap */);
        buf.put_u8(MANIFEST_FORMAT_VER);
        buf.put_i64_le(self.main_queue_cap);
        buf
    }

    fn unmarshal(mut data: &[u8]) -> Result<Self> {
        if data.len() < U8_SIZE + I64_SIZE {
            return Err(Error::IaMgr("manifest data too small".to_string()));
        }

        let ver = data.get_u8();
        if ver != MANIFEST_FORMAT_VER {
            return Err(Error::IaMgr(format!("manifest version not match: {ver}")));
        }

        let main_queue_cap = data.get_i64_le();
        Ok(Self { main_queue_cap })
    }

    fn persist_to_path(&self, path: &Path) -> std::io::Result<()> {
        let mut f = tempfile::Builder::new()
            .prefix(MANIFEST_FILE_NAME)
            .tempfile_in(path)?;
        f.write_all(&self.marshal())?;
        f.flush()?;
        f.persist(path.join(MANIFEST_FILE_NAME))?;
        Ok(())
    }

    fn read_from_path(path: &Path) -> Result<Option<Self>> {
        let mut f = match std::fs::File::open(path.join(MANIFEST_FILE_NAME)) {
            Ok(f) => f,
            Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
                return Ok(None);
            }
            Err(err) => return Err(Error::IaMgr(format!("open manifest error {err:?}"))),
        };
        let mut data = vec![];
        f.read_to_end(&mut data)
            .map_err(|e| Error::IaMgr(format!("read manifest error {e:?}")))?;
        Ok(Some(Self::unmarshal(&data)?))
    }
}

#[cfg(any(test, feature = "testexport"))]
#[cfg_attr(feature = "testexport", allow(unused))]
pub mod tests {
    use futures::FutureExt;
    use tikv_util::config::ReadableSize;

    use super::*;
    use crate::{
        dfs::{InMemFs, Options},
        ia::{
            queue::QueueItemPos,
            util::{IaCapacity, IaManagerOptionsBuilder},
        },
    };

    fn insert_cached_segment(mgr: &IaManager, ident: FileSegmentIdent, insert_main: bool) {
        let memory_permit = Arc::new(Semaphore::new(1)).try_acquire_owned().unwrap();
        mgr.core.segments.set_segment_data(
            ident,
            FileSegmentData::in_memory(Bytes::from_static(b"x"), memory_permit),
        );
        mgr.core.fifo.read(ident, insert_main).unwrap();
    }

    fn queue_item_counts(
        mgr: &IaManager,
        idents: &[FileSegmentIdent],
    ) -> (usize /* small */, usize /* main */) {
        let mut small_count = 0;
        let mut main_count = 0;
        for &ident in idents {
            let runtime_handle = mgr.runtime_handle();
            let mgr_for_get = mgr.clone();
            let queue_item = runtime_handle
                .block_on(async move { mgr_for_get.core.fifo.get_item(ident).await })
                .unwrap();
            let segment_data = mgr.core.segments.get_segment(ident);
            match (queue_item.map(|item| item.pos), segment_data) {
                (Some(QueueItemPos::Small), Some(FileSegmentData::InMem(_))) => small_count += 1,
                (Some(QueueItemPos::Main), Some(FileSegmentData::InStore)) => main_count += 1,
                (None, None) => {}
                state => panic!(
                    "queue and segment state differ for {:?}: {:?}",
                    ident, state
                ),
            }
        }
        (small_count, main_count)
    }

    #[test]
    fn test_manifest() {
        let dir = tempfile::tempdir().unwrap();

        assert!(Manifest::read_from_path(dir.path()).unwrap().is_none());

        let manifest = Manifest {
            main_queue_cap: 1024,
        };
        manifest
            .persist_to_path(dir.path())
            .expect("persist manifest");

        let manifest1 = Manifest::read_from_path(dir.path()).unwrap().unwrap();
        assert_eq!(manifest1, manifest);
    }

    #[test]
    fn test_dynamic_capacity_controls_manifest_restore() {
        let runtime = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(1)
            .enable_all()
            .build()
            .unwrap();
        let fs = Arc::new(InMemFs::new());
        let dir = tempfile::tempdir().unwrap();
        Manifest {
            main_queue_cap: 256,
        }
        .persist_to_path(dir.path())
        .unwrap();
        let options = IaManagerOptionsBuilder::default()
            .capacity(IaCapacity::Manual {
                small_queue: QueueOptions {
                    paths: vec![],
                    cap: 100,
                },
                main_queue: QueueOptions {
                    paths: vec![dir.path().to_path_buf()],
                    cap: 1000,
                },
            })
            .segment_size(1)
            .dfs_concurrency(1, 1)
            .dynamic_capacity(false)
            .build()
            .unwrap();

        let mgr = IaManager::new(options, fs, None, runtime.into()).unwrap();
        assert_eq!(mgr.cache_queue_capacities(), (100, 1000));

        let runtime = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(1)
            .enable_all()
            .build()
            .unwrap();
        let fs = Arc::new(InMemFs::new());
        let dir = tempfile::tempdir().unwrap();
        Manifest {
            main_queue_cap: 500,
        }
        .persist_to_path(dir.path())
        .unwrap();
        let options = IaManagerOptionsBuilder::default()
            .capacity(IaCapacity::Manual {
                small_queue: QueueOptions {
                    paths: vec![],
                    cap: 100,
                },
                main_queue: QueueOptions {
                    paths: vec![dir.path().to_path_buf()],
                    cap: 1000,
                },
            })
            .segment_size(1)
            .dfs_concurrency(1, 1)
            .dynamic_capacity(true)
            .build()
            .unwrap();

        let mgr = IaManager::new(options, fs, None, runtime.into()).unwrap();
        assert_eq!(mgr.cache_queue_capacities(), (64, 500));
    }

    #[test]
    fn test_remote_read_waits_for_local_segment_map_memory_budget() {
        let runtime = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(1)
            .enable_all()
            .build()
            .unwrap();
        let fs = Arc::new(InMemFs::new());
        runtime
            .block_on(fs.create(
                1,
                Bytes::from(vec![1; 10]),
                Options::default().with_type(FileType::Sst),
            ))
            .unwrap();

        let options = IaManagerOptionsBuilder::default()
            .capacity(IaCapacity::Manual {
                small_queue: QueueOptions {
                    paths: vec![],
                    cap: 10,
                },
                main_queue: QueueOptions {
                    paths: vec![],
                    cap: 100,
                },
            })
            .segment_size(10)
            .dfs_concurrency(1, 1)
            .build()
            .unwrap();
        let mgr = IaManager::new(options, fs, None, runtime.into()).unwrap();

        // The admission cap is small_queue_cap + segment_size * dfs_concurrency * 2 =
        // 30. Filling it forces the next remote read to wait until memory is
        // released.
        let cached_ident = FileSegmentIdent::new(2, 0, 30);
        let cached_permit = mgr
            .core
            .local_segment_memory_limiter
            .clone()
            .try_acquire_many_owned(30)
            .unwrap();
        mgr.core.segments.set_segment_data(
            cached_ident,
            FileSegmentData::in_memory(Bytes::from(vec![2; 30]), cached_permit),
        );

        let remote_ident = FileSegmentIdent::new(1, 0, 10);
        let mgr_for_read = mgr.clone();
        let (read_tx, read_rx) = std::sync::mpsc::channel();
        let read_handle = mgr.runtime_handle().spawn(async move {
            let mut buf = [0; 10];
            let mut read_at = ReadAt::new(&mut buf, 0, false);
            mgr_for_read
                .read_segment(
                    remote_ident,
                    FileType::Sst,
                    None,
                    None,
                    &mut read_at,
                    None,
                    true,
                )
                .await
                .unwrap();
            read_tx.send(buf).unwrap();
        });
        std::thread::sleep(Duration::from_millis(20));
        assert!(matches!(
            read_rx.try_recv(),
            Err(std::sync::mpsc::TryRecvError::Empty)
        ));

        mgr.core.segments.remove(cached_ident);
        let buf = read_rx.recv_timeout(Duration::from_secs(1)).unwrap();
        mgr.runtime_handle().block_on(read_handle).unwrap();
        assert_eq!(buf, [1; 10]);
    }

    #[test]
    fn test_remote_read_deadline_after_throttle_unconsumes() {
        let runtime = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(1)
            .enable_all()
            .build()
            .unwrap();
        let fs = Arc::new(InMemFs::new());
        runtime
            .block_on(fs.create(
                1,
                Bytes::from(vec![1; 110]),
                Options::default().with_type(FileType::Sst),
            ))
            .unwrap();

        let options = IaManagerOptionsBuilder::default()
            .capacity(IaCapacity::Manual {
                small_queue: QueueOptions {
                    paths: vec![],
                    cap: 110,
                },
                main_queue: QueueOptions {
                    paths: vec![],
                    cap: 100,
                },
            })
            .segment_size(110)
            .dfs_concurrency(1, 1)
            .remote_read_max_throughput(ReadableSize::kb(1))
            .build()
            .unwrap();
        let mgr = IaManager::new(options, fs, None, runtime.into()).unwrap();

        let ident = FileSegmentIdent::new(1, 0, 110);
        let mgr_for_read = mgr.clone();
        let err = mgr.runtime_handle().block_on(async move {
            let mut buf = [0; 110];
            let mut read_at = ReadAt::new(&mut buf, 0, false);
            mgr_for_read
                .read_segment(
                    ident,
                    FileType::Sst,
                    None,
                    Some(Deadline::from_now(Duration::from_millis(50))),
                    &mut read_at,
                    None,
                    true,
                )
                .await
                .unwrap_err()
        });

        assert!(matches!(err, Error::DeadlineExceeded(_)));
        assert_eq!(mgr.core.io_limiter.remote_read_total_bytes_consumed(), 0);
    }

    #[test]
    fn test_remote_read_can_skip_ia_throttle() {
        let runtime = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(1)
            .enable_all()
            .build()
            .unwrap();
        let fs = Arc::new(InMemFs::new());
        runtime
            .block_on(fs.create(
                1,
                Bytes::from(vec![1; 64]),
                Options::default().with_type(FileType::Sst),
            ))
            .unwrap();

        let options = IaManagerOptionsBuilder::default()
            .capacity(IaCapacity::Manual {
                small_queue: QueueOptions {
                    paths: vec![],
                    cap: 1024,
                },
                main_queue: QueueOptions {
                    paths: vec![],
                    cap: 1024,
                },
            })
            .segment_size(64)
            .remote_read_max_throughput(ReadableSize::kb(1))
            .build()
            .unwrap();
        let mgr = IaManager::new(options, fs, None, runtime.into()).unwrap();

        let ident = FileSegmentIdent::new(1, 0, 64);
        let mgr_for_read = mgr.clone();
        mgr.runtime_handle()
            .block_on(async move {
                mgr_for_read
                    .get_segment_handle(ident, FileType::Sst, None, false)
                    .await
            })
            .unwrap();

        assert_eq!(mgr.core.io_limiter.remote_read_total_bytes_consumed(), 0);
    }

    #[test]
    fn test_local_segment_memory_permits_rejects_u32_overflow() {
        let runtime = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(1)
            .enable_all()
            .build()
            .unwrap();
        let fs = Arc::new(InMemFs::new());
        let options = IaManagerOptionsBuilder::default()
            .capacity(IaCapacity::Manual {
                small_queue: QueueOptions {
                    paths: vec![],
                    cap: u32::MAX as i64 + 2,
                },
                main_queue: QueueOptions {
                    paths: vec![],
                    cap: 100,
                },
            })
            .segment_size(1)
            .dfs_concurrency(1, 1)
            .build()
            .unwrap();
        let mgr = IaManager::new(options, fs, None, runtime.into()).unwrap();

        let err = mgr
            .core
            .local_segment_memory_permits_for_bytes(u32::MAX as u64 + 1)
            .unwrap_err();
        assert!(format!("{err:?}").contains("permit limit"));
    }

    #[test]
    fn test_local_segment_map_memory_budget_uses_configured_small_cap() {
        let runtime = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(1)
            .enable_all()
            .build()
            .unwrap();
        let fs = Arc::new(InMemFs::new());
        let options = IaManagerOptionsBuilder::default()
            .capacity(IaCapacity::Manual {
                small_queue: QueueOptions {
                    paths: vec![],
                    cap: 100,
                },
                main_queue: QueueOptions {
                    paths: vec![],
                    cap: 1000,
                },
            })
            .segment_size(1)
            .dfs_concurrency(1, 1)
            .dynamic_capacity(true)
            .build()
            .unwrap();
        let mgr = IaManager::new(options, fs, None, runtime.into()).unwrap();

        // Memory admission uses the configured small cap, not the dynamic runtime cap,
        // and reserves room for one largest legal item.
        assert_eq!(mgr.core.local_segment_memory_cap(), 200);
        assert_eq!(
            mgr.core.local_segment_memory_limiter.available_permits(),
            200
        );

        mgr.core.adjust_queue_cap(1000);
        assert_eq!(mgr.core.small_queue_cap.load(Ordering::Relaxed), 100);
        assert_eq!(mgr.core.local_segment_memory_cap(), 200);
        assert_eq!(
            mgr.core.local_segment_memory_limiter.available_permits(),
            200
        );

        mgr.core.adjust_queue_cap(256);
        assert_eq!(mgr.core.small_queue_cap.load(Ordering::Relaxed), 64);
        assert_eq!(mgr.core.local_segment_memory_cap(), 200);
        assert_eq!(
            mgr.core.local_segment_memory_limiter.available_permits(),
            200
        );
    }

    #[test]
    fn test_online_mem_cap_growth_grows_memory_admission_high_water() {
        let runtime = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(1)
            .enable_all()
            .build()
            .unwrap();
        let fs = Arc::new(InMemFs::new());
        let options = IaManagerOptionsBuilder::default()
            .capacity(IaCapacity::Manual {
                small_queue: QueueOptions {
                    paths: vec![],
                    cap: 100,
                },
                main_queue: QueueOptions {
                    paths: vec![],
                    cap: 1000,
                },
            })
            .segment_size(1)
            .dfs_concurrency(1, 1)
            .dynamic_capacity(true)
            .build()
            .unwrap();
        let mgr = IaManager::new(options, fs, None, runtime.into()).unwrap();

        let held = mgr
            .core
            .local_segment_memory_limiter
            .clone()
            .try_acquire_many_owned(200)
            .unwrap();
        assert_eq!(mgr.core.local_segment_memory_cap(), 200);
        assert_eq!(mgr.core.local_segment_memory_limiter.available_permits(), 0);

        let mgr_for_acquire = mgr.clone();
        let (pending_tx, pending_rx) = std::sync::mpsc::channel();
        let (permit_tx, permit_rx) = std::sync::mpsc::channel();
        let acquire_handle = mgr.runtime_handle().spawn(async move {
            let ident = FileSegmentIdent::new(42, 0, 100);
            let mut acquire = Box::pin(mgr_for_acquire.core.acquire_remote_read_memory(ident));
            assert!(acquire.as_mut().now_or_never().is_none());
            pending_tx.send(()).unwrap();
            let permit = acquire.await.unwrap();
            permit_tx.send(permit).unwrap();
        });
        pending_rx.recv_timeout(Duration::from_secs(1)).unwrap();

        mgr.core.apply_queue_cap_limits_for_test(200, 1000).unwrap();
        let growth_permit = permit_rx.recv_timeout(Duration::from_secs(1)).unwrap();
        mgr.runtime_handle().block_on(acquire_handle).unwrap();
        assert_eq!(mgr.core.local_segment_memory_cap(), 400);
        assert_eq!(
            mgr.core.local_segment_memory_limiter.available_permits(),
            100
        );

        // Shrinking the queue leaves admission at its high-water mark because
        // permits may still be in use. The active queue trim enforces the live cap.
        mgr.core.apply_queue_cap_limits_for_test(80, 1000).unwrap();
        assert_eq!(mgr.core.local_segment_memory_cap(), 400);
        assert_eq!(
            mgr.core.local_segment_memory_limiter.available_permits(),
            100
        );

        drop(held);
        assert_eq!(
            mgr.core.local_segment_memory_limiter.available_permits(),
            300
        );
        drop(growth_permit);
        assert_eq!(
            mgr.core.local_segment_memory_limiter.available_permits(),
            400
        );

        mgr.core.apply_queue_cap_limits_for_test(250, 1000).unwrap();
        assert_eq!(mgr.core.local_segment_memory_cap(), 500);
        assert_eq!(
            mgr.core.local_segment_memory_limiter.available_permits(),
            500
        );
    }

    #[test]
    fn test_cap_sized_remote_reads_progress_with_full_small_queue() {
        let runtime = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(2)
            .enable_all()
            .build()
            .unwrap();
        let fs = Arc::new(InMemFs::new());
        for file_id in 1..=2 {
            runtime
                .block_on(fs.create(
                    file_id,
                    Bytes::from(vec![file_id as u8; 200]),
                    Options::default().with_type(FileType::Sst),
                ))
                .unwrap();
        }
        let options = IaManagerOptionsBuilder::default()
            .capacity(IaCapacity::Manual {
                small_queue: QueueOptions {
                    paths: vec![],
                    cap: 200,
                },
                main_queue: QueueOptions {
                    paths: vec![],
                    cap: 1000,
                },
            })
            .segment_size(1)
            .dfs_concurrency(2, 2)
            .dynamic_capacity(false)
            .build()
            .unwrap();
        let mgr = IaManager::new(options, fs, None, runtime.into()).unwrap();

        let cached_ident = FileSegmentIdent::new(3, 0, 200);
        let cached_permit = mgr
            .core
            .local_segment_memory_limiter
            .clone()
            .try_acquire_many_owned(200)
            .unwrap();
        mgr.core.segments.set_segment_data(
            cached_ident,
            FileSegmentData::in_memory(Bytes::from(vec![3; 200]), cached_permit),
        );
        mgr.core.fifo.read(cached_ident, false).unwrap();
        let mgr_for_flush = mgr.clone();
        mgr.runtime_handle()
            .block_on(async move { mgr_for_flush.flush_tasks(Duration::from_secs(5)).await })
            .unwrap();
        assert_eq!(mgr.core.local_segment_memory_cap(), 400);
        assert_eq!(
            mgr.core.local_segment_memory_limiter.available_permits(),
            200
        );

        let mgr_for_first = mgr.clone();
        let mgr_for_second = mgr.clone();
        let reads = async move {
            let first = async move {
                let ident = FileSegmentIdent::new(1, 0, 200);
                let mut buf = [0; 200];
                let mut read_at = ReadAt::new(&mut buf, 0, false);
                mgr_for_first
                    .read_segment(ident, FileType::Sst, None, None, &mut read_at, None, true)
                    .await?;
                Ok::<_, Error>(buf)
            };
            let second = async move {
                let ident = FileSegmentIdent::new(2, 0, 200);
                let mut buf = [0; 200];
                let mut read_at = ReadAt::new(&mut buf, 0, false);
                mgr_for_second
                    .read_segment(ident, FileType::Sst, None, None, &mut read_at, None, true)
                    .await?;
                Ok::<_, Error>(buf)
            };
            tokio::try_join!(first, second)
        };
        let (first, second) = mgr
            .runtime_handle()
            .block_on(async move { tokio::time::timeout(Duration::from_secs(2), reads).await })
            .expect("cap-sized reads should not stall behind a full small queue")
            .unwrap();
        assert_eq!(first, [1; 200]);
        assert_eq!(second, [2; 200]);

        let mgr_for_flush = mgr.clone();
        mgr.runtime_handle()
            .block_on(async move { mgr_for_flush.flush_tasks(Duration::from_secs(5)).await })
            .unwrap();
        assert_eq!(
            mgr.core.local_segment_memory_limiter.available_permits(),
            200
        );
    }

    #[test]
    fn test_mem_cap_growth_rejects_admission_overflow_atomically() {
        let runtime = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(1)
            .enable_all()
            .build()
            .unwrap();
        let fs = Arc::new(InMemFs::new());
        let options = IaManagerOptionsBuilder::default()
            .capacity(IaCapacity::Manual {
                small_queue: QueueOptions {
                    paths: vec![],
                    cap: 100,
                },
                main_queue: QueueOptions {
                    paths: vec![],
                    cap: 1000,
                },
            })
            .segment_size(1)
            .dfs_concurrency(1, 1)
            .dynamic_capacity(false)
            .build()
            .unwrap();
        let mgr = IaManager::new(options, fs, None, runtime.into()).unwrap();

        let err = mgr
            .core
            .publish_cache_capacity_config(
                Some(Semaphore::MAX_PERMITS as i64),
                Some(2000),
                None,
                None,
            )
            .unwrap_err();
        assert!(format!("{err:?}").contains("admission capacity is out of range"));
        let config = mgr.core.cache_capacity_config.load();
        assert_eq!(config.small_queue_max_cap, 100);
        assert_eq!(config.main_queue_max_cap, 1000);
        assert_eq!(mgr.core.small_queue_cap.load(Ordering::Relaxed), 100);
        assert_eq!(mgr.core.main_queue_cap.load(Ordering::Relaxed), 1000);
        assert_eq!(mgr.core.local_segment_memory_cap(), 200);
        assert_eq!(
            mgr.core.local_segment_memory_limiter.available_permits(),
            200
        );
    }

    #[test]
    fn test_remote_read_rejects_new_item_larger_than_live_small_cap() {
        let runtime = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(1)
            .enable_all()
            .build()
            .unwrap();
        let fs = Arc::new(InMemFs::new());
        runtime
            .block_on(fs.create(
                1,
                Bytes::from(vec![1; 65]),
                Options::default().with_type(FileType::Sst),
            ))
            .unwrap();
        let options = IaManagerOptionsBuilder::default()
            .capacity(IaCapacity::Manual {
                small_queue: QueueOptions {
                    paths: vec![],
                    cap: 128,
                },
                main_queue: QueueOptions {
                    paths: vec![],
                    cap: 1000,
                },
            })
            .segment_size(1)
            .dfs_concurrency(1, 1)
            .dynamic_capacity(true)
            .build()
            .unwrap();
        let mgr = IaManager::new(options, fs, None, runtime.into()).unwrap();
        mgr.core.adjust_queue_cap(1000);
        assert_eq!(mgr.core.small_queue_cap.load(Ordering::Relaxed), 128);
        mgr.core.apply_queue_cap_limits_for_test(64, 1000).unwrap();

        let available_permits = mgr.core.local_segment_memory_limiter.available_permits();
        let ident = FileSegmentIdent::new(1, 0, 65);
        let mgr_for_read = mgr.clone();
        let err = mgr.runtime_handle().block_on(async move {
            let mut buf = [0; 65];
            let mut read_at = ReadAt::new(&mut buf, 0, false);
            mgr_for_read
                .read_segment(ident, FileType::Sst, None, None, &mut read_at, None, true)
                .await
                .unwrap_err()
        });

        let err = format!("{err:?}");
        assert!(err.contains("size: 65"), "{}", err);
        assert!(err.contains("capacity: 64"), "{}", err);
        assert!(mgr.core.segments.get_segment(ident).is_none());
        let mgr_for_get = mgr.clone();
        assert!(
            mgr.runtime_handle()
                .block_on(async move { mgr_for_get.core.fifo.get_item(ident).await })
                .unwrap()
                .is_none()
        );
        assert_eq!(
            mgr.core.local_segment_memory_limiter.available_permits(),
            available_permits
        );
        assert_eq!(mgr.core.io_limiter.remote_read_total_bytes_consumed(), 0);
    }

    #[test]
    fn test_remote_read_admitted_before_shrink_finishes_and_is_trimmed() {
        let runtime = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(1)
            .enable_all()
            .build()
            .unwrap();
        let fs = Arc::new(InMemFs::new());
        runtime
            .block_on(fs.create(
                1,
                Bytes::from(vec![1; 65]),
                Options::default().with_type(FileType::Sst),
            ))
            .unwrap();
        let options = IaManagerOptionsBuilder::default()
            .capacity(IaCapacity::Manual {
                small_queue: QueueOptions {
                    paths: vec![],
                    cap: 128,
                },
                main_queue: QueueOptions {
                    paths: vec![],
                    cap: 1000,
                },
            })
            .segment_size(1)
            .dfs_concurrency(1, 1)
            .dynamic_capacity(false)
            .build()
            .unwrap();
        let mgr = IaManager::new(options, fs, None, runtime.into()).unwrap();
        let ident = FileSegmentIdent::new(1, 0, 65);

        let mgr_for_lock = mgr.clone();
        let loading_guard = mgr
            .runtime_handle()
            .block_on(async move { mgr_for_lock.core.loading_segments.get_locked(ident).await });
        let mgr_for_read = mgr.clone();
        let read_handle = mgr.runtime_handle().spawn(async move {
            let mut buf = [0; 65];
            let mut read_at = ReadAt::new(&mut buf, 0, false);
            mgr_for_read
                .read_segment(ident, FileType::Sst, None, None, &mut read_at, None, true)
                .await
                .unwrap();
            buf
        });

        let wait_deadline = std::time::Instant::now() + Duration::from_secs(1);
        loop {
            let strong_count = mgr
                .core
                .loading_segments
                .get(&ident)
                .map(|mutex| Arc::strong_count(mutex.value()))
                .unwrap_or(0);
            if strong_count >= 3 {
                break;
            }
            assert!(
                std::time::Instant::now() < wait_deadline,
                "remote read did not wait on the per-segment loading lock"
            );
            std::thread::sleep(Duration::from_millis(1));
        }

        mgr.core.apply_queue_cap_limits_for_test(64, 1000).unwrap();
        drop(loading_guard);
        let buf = mgr.runtime_handle().block_on(read_handle).unwrap();
        assert_eq!(buf, [1; 65]);

        let mgr_for_flush = mgr.clone();
        mgr.runtime_handle()
            .block_on(async move { mgr_for_flush.flush_tasks(Duration::from_secs(5)).await })
            .unwrap();

        assert!(mgr.core.segments.get_segment(ident).is_none());
        let mgr_for_get = mgr.clone();
        assert!(
            mgr.runtime_handle()
                .block_on(async move { mgr_for_get.core.fifo.get_item(ident).await })
                .unwrap()
                .is_none()
        );
        assert_eq!(
            mgr.core.local_segment_memory_limiter.available_permits(),
            256
        );
        assert_eq!(mgr.core.io_limiter.remote_read_total_bytes_consumed(), 65);
    }

    #[test]
    fn test_remote_item_admission_rolls_back_when_fifo_stopped() {
        let runtime = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(1)
            .enable_all()
            .build()
            .unwrap();
        let fs = Arc::new(InMemFs::new());
        let options = IaManagerOptionsBuilder::default()
            .capacity(IaCapacity::Manual {
                small_queue: QueueOptions {
                    paths: vec![],
                    cap: 100,
                },
                main_queue: QueueOptions {
                    paths: vec![],
                    cap: 100,
                },
            })
            .segment_size(1)
            .dfs_concurrency(1, 1)
            .build()
            .unwrap();
        let mut mgr = IaManager::new(options, fs, None, runtime.into()).unwrap();
        let ident = FileSegmentIdent::new(1, 0, 10);
        let memory_permit = mgr
            .core
            .local_segment_memory_limiter
            .clone()
            .try_acquire_many_owned(10)
            .unwrap();
        Arc::get_mut(&mut mgr.core).unwrap().fifo.stop();

        let err = mgr
            .core
            .admit_remote_segment(ident, Bytes::from(vec![1; 10]), memory_permit)
            .unwrap_err();
        assert!(format!("{err:?}").contains("send failed"));
        assert!(mgr.core.segments.get_segment(ident).is_none());
        assert_eq!(
            mgr.core.local_segment_memory_limiter.available_permits(),
            200
        );
    }

    #[test]
    fn test_init_tracks_main_item_larger_than_small_cap() {
        let runtime = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(1)
            .enable_all()
            .build()
            .unwrap();
        let fs = Arc::new(InMemFs::new());
        let options = IaManagerOptionsBuilder::default()
            .capacity(IaCapacity::Manual {
                small_queue: QueueOptions {
                    paths: vec![],
                    cap: 64,
                },
                main_queue: QueueOptions {
                    paths: vec![],
                    cap: 100,
                },
            })
            .segment_size(1)
            .dfs_concurrency(1, 1)
            .dynamic_capacity(false)
            .build()
            .unwrap();
        let mgr = IaManager::new(options, fs, None, runtime.into()).unwrap();
        let ident = FileSegmentIdent::new(1, 0, 65);
        let filename = ident.local_filename();
        mgr.core
            .main_store
            .save(ident.file_id, &filename, Bytes::from(vec![7; 65]))
            .unwrap();
        mgr.core.init_segments(vec![filename]).unwrap();
        let mgr_for_flush = mgr.clone();
        mgr.runtime_handle()
            .block_on(async move { mgr_for_flush.flush_tasks(Duration::from_secs(5)).await })
            .unwrap();

        assert!(matches!(
            mgr.core.segments.get_segment(ident),
            Some(FileSegmentData::InStore)
        ));
        let mgr_for_get = mgr.clone();
        assert_eq!(
            mgr.runtime_handle()
                .block_on(async move { mgr_for_get.core.fifo.get_item(ident).await })
                .unwrap()
                .unwrap()
                .pos,
            QueueItemPos::Main
        );

        let mgr_for_read = mgr.clone();
        let buf = mgr.runtime_handle().block_on(async move {
            let mut buf = [0; 65];
            let mut read_at = ReadAt::new(&mut buf, 0, false);
            mgr_for_read
                .read_segment(ident, FileType::Sst, None, None, &mut read_at, None, true)
                .await
                .unwrap();
            buf
        });
        assert_eq!(buf, [7; 65]);
    }

    #[test]
    fn test_cache_capacity_updates_apply_only_on_data_size_report() {
        let runtime = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(1)
            .enable_all()
            .build()
            .unwrap();
        let fs = Arc::new(InMemFs::new());
        let options = IaManagerOptionsBuilder::default()
            .capacity(IaCapacity::Manual {
                small_queue: QueueOptions {
                    paths: vec![],
                    cap: 100,
                },
                main_queue: QueueOptions {
                    paths: vec![],
                    cap: 1000,
                },
            })
            .segment_size(1)
            .dfs_concurrency(1, 1)
            .dynamic_capacity(true)
            .cache_cap_to_total_data_size_ratio(0.1)
            .build()
            .unwrap();
        let mgr = IaManager::new(options, fs, None, runtime.into()).unwrap();

        assert_eq!(mgr.core.small_queue_cap.load(Ordering::Relaxed), 64);
        assert_eq!(mgr.core.main_queue_cap.load(Ordering::Relaxed), 256);

        mgr.core.notify_total_data_size(100 * GIB);
        assert_eq!(mgr.core.small_queue_cap.load(Ordering::Relaxed), 100);
        assert_eq!(mgr.core.main_queue_cap.load(Ordering::Relaxed), 1000);

        mgr.update_cache_capacity(120_u64.into(), 400_u64.into())
            .unwrap();
        assert_eq!(mgr.core.small_queue_cap.load(Ordering::Relaxed), 100);
        assert_eq!(mgr.core.main_queue_cap.load(Ordering::Relaxed), 1000);
        assert_eq!(mgr.core.local_segment_memory_cap(), 200);

        // A later desired config replaces the unapplied update. Neither the
        // live caps nor memory admission change before the next report.
        mgr.update_cache_capacity(80_u64.into(), 400_u64.into())
            .unwrap();
        assert_eq!(mgr.core.small_queue_cap.load(Ordering::Relaxed), 100);
        assert_eq!(mgr.core.main_queue_cap.load(Ordering::Relaxed), 1000);
        assert_eq!(mgr.core.local_segment_memory_cap(), 200);

        mgr.core.notify_total_data_size(100 * GIB);
        assert_eq!(mgr.core.small_queue_cap.load(Ordering::Relaxed), 80);
        assert_eq!(mgr.core.main_queue_cap.load(Ordering::Relaxed), 400);

        mgr.update_cache_capacity(20_u64.into(), 100_u64.into())
            .unwrap();
        assert_eq!(mgr.core.small_queue_cap.load(Ordering::Relaxed), 80);
        assert_eq!(mgr.core.main_queue_cap.load(Ordering::Relaxed), 400);

        mgr.core.notify_total_data_size(100);
        assert_eq!(mgr.core.small_queue_cap.load(Ordering::Relaxed), 20);
        assert_eq!(mgr.core.main_queue_cap.load(Ordering::Relaxed), 100);
    }

    #[test]
    fn test_fixed_capacity_manager_changes_only_after_data_size_report() {
        let runtime = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(1)
            .enable_all()
            .build()
            .unwrap();
        let fs = Arc::new(InMemFs::new());
        let options = IaManagerOptionsBuilder::default()
            .capacity(IaCapacity::Manual {
                small_queue: QueueOptions {
                    paths: vec![],
                    cap: 100,
                },
                main_queue: QueueOptions {
                    paths: vec![],
                    cap: 1000,
                },
            })
            .segment_size(1)
            .dfs_concurrency(1, 1)
            .dynamic_capacity(false)
            .build()
            .unwrap();
        let mgr = IaManager::new(options, fs, None, runtime.into()).unwrap();

        assert_eq!(mgr.core.small_queue_cap.load(Ordering::Relaxed), 100);
        assert_eq!(mgr.core.main_queue_cap.load(Ordering::Relaxed), 1000);

        mgr.update_cache_capacity(30_u64.into(), 300_u64.into())
            .unwrap();
        assert_eq!(mgr.core.small_queue_cap.load(Ordering::Relaxed), 100);
        assert_eq!(mgr.core.main_queue_cap.load(Ordering::Relaxed), 1000);

        mgr.update_cache_capacity(60_u64.into(), 600_u64.into())
            .unwrap();
        assert_eq!(mgr.core.small_queue_cap.load(Ordering::Relaxed), 100);
        assert_eq!(mgr.core.main_queue_cap.load(Ordering::Relaxed), 1000);

        // Fixed-capacity mode ignores the reported IA data size and applies
        // the latest configured memory and disk capacities.
        mgr.core.notify_total_data_size(10 * GIB);
        assert_eq!(mgr.core.small_queue_cap.load(Ordering::Relaxed), 60);
        assert_eq!(mgr.core.main_queue_cap.load(Ordering::Relaxed), 600);

        mgr.update_cache_capacity_config(None, None, Some(0.0), None)
            .unwrap();
        mgr.core.notify_total_data_size(0);
        assert_eq!(mgr.core.small_queue_cap.load(Ordering::Relaxed), 60);
        assert_eq!(mgr.core.main_queue_cap.load(Ordering::Relaxed), 600);
    }

    fn new_cache_capacity_test_manager() -> IaManager {
        const MIB: u64 = tikv_util::config::MIB;

        let runtime = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(1)
            .enable_all()
            .build()
            .unwrap();
        let fs = Arc::new(InMemFs::new());
        let options = IaManagerOptionsBuilder::default()
            .capacity(IaCapacity::Manual {
                small_queue: QueueOptions {
                    paths: vec![],
                    cap: 100 * MIB as i64,
                },
                main_queue: QueueOptions {
                    paths: vec![],
                    cap: 10 * GIB as i64,
                },
            })
            .segment_size(MIB as i64)
            .dfs_concurrency(1, 1)
            .dynamic_capacity(true)
            .cache_cap_to_total_data_size_ratio(1.0)
            .cache_capacity_shrink_step(ReadableSize::gb(10))
            .build()
            .unwrap();
        IaManager::new(options, fs, None, runtime.into()).unwrap()
    }

    #[test]
    fn test_new_rejects_invalid_cache_capacity_ratio() {
        for invalid_ratio in [-0.1, f64::NAN, f64::INFINITY, f64::NEG_INFINITY] {
            let mut options = IaManagerOptionsBuilder::default().build().unwrap();
            options.cache_cap_to_total_data_size_ratio = invalid_ratio;
            let runtime = tokio::runtime::Builder::new_multi_thread()
                .worker_threads(1)
                .enable_all()
                .build()
                .unwrap();
            let result = IaManager::new(options, Arc::new(InMemFs::new()), None, runtime.into());
            let err = match result {
                Ok(_) => panic!("invalid ratio {} should be rejected", invalid_ratio),
                Err(err) => err,
            };
            assert!(
                err.to_string()
                    .contains("cache-cap-to-total-data-size-ratio"),
                "unexpected error for {}: {}",
                invalid_ratio,
                err
            );
        }
    }

    #[test]
    fn test_cache_capacity_ratio_applies_on_data_size_notification() {
        const MIB: u64 = tikv_util::config::MIB;

        let mgr = new_cache_capacity_test_manager();
        mgr.notify_total_data_size(100 * GIB);

        // Ratio updates publish desired state without changing live caps.
        mgr.update_cache_capacity_config(None, None, Some(0.5), None)
            .unwrap();
        assert_eq!(
            mgr.core.small_queue_cap.load(Ordering::Relaxed),
            100 * MIB as i64
        );
        assert_eq!(
            mgr.core.main_queue_cap.load(Ordering::Relaxed),
            10 * GIB as i64
        );

        mgr.notify_total_data_size(4 * GIB);
        assert_eq!(
            mgr.core.small_queue_cap.load(Ordering::Relaxed),
            64 * MIB as i64
        );
        assert_eq!(
            mgr.core.main_queue_cap.load(Ordering::Relaxed),
            2 * GIB as i64
        );

        // Ratio changes wait for the next size notification.
        mgr.update_cache_capacity_config(None, None, Some(0.1), None)
            .unwrap();
        assert_eq!(
            mgr.core.small_queue_cap.load(Ordering::Relaxed),
            64 * MIB as i64
        );
        assert_eq!(
            mgr.core.main_queue_cap.load(Ordering::Relaxed),
            2 * GIB as i64
        );
        mgr.notify_total_data_size(4 * GIB);
        assert_eq!(
            mgr.core.small_queue_cap.load(Ordering::Relaxed),
            64 * MIB as i64
        );
        assert_eq!(
            mgr.core.main_queue_cap.load(Ordering::Relaxed),
            256 * MIB as i64
        );

        mgr.update_cache_capacity_config(None, None, Some(0.75), None)
            .unwrap();
        assert_eq!(
            mgr.core.small_queue_cap.load(Ordering::Relaxed),
            64 * MIB as i64
        );
        assert_eq!(
            mgr.core.main_queue_cap.load(Ordering::Relaxed),
            256 * MIB as i64
        );
        mgr.notify_total_data_size(4 * GIB);
        assert_eq!(
            mgr.core.small_queue_cap.load(Ordering::Relaxed),
            64 * MIB as i64
        );
        assert_eq!(
            mgr.core.main_queue_cap.load(Ordering::Relaxed),
            3 * GIB as i64
        );

        assert!(
            mgr.update_cache_capacity_config(None, None, Some(f64::NAN), None)
                .is_err()
        );
        mgr.notify_total_data_size(2 * GIB);
        assert_eq!(
            mgr.core.small_queue_cap.load(Ordering::Relaxed),
            64 * MIB as i64
        );
        assert_eq!(mgr.core.main_queue_cap.load(Ordering::Relaxed), GIB as i64);

        // A non-GiB max cap remains the hard ceiling after quantization.
        let non_gib_max_cap = GIB as i64 - 1;
        mgr.core
            .apply_queue_cap_limits_for_test(80 * MIB as i64, non_gib_max_cap)
            .unwrap();
        assert_eq!(
            mgr.core.small_queue_cap.load(Ordering::Relaxed),
            80 * MIB as i64
        );
        assert_eq!(
            mgr.core.main_queue_cap.load(Ordering::Relaxed),
            non_gib_max_cap
        );
    }

    fn new_cache_capacity_shrink_test_manager() -> IaManager {
        const MIB: u64 = tikv_util::config::MIB;

        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: 5 * GIB as i64,
                },
                main_queue: QueueOptions {
                    paths: vec![],
                    cap: 20 * GIB as i64,
                },
            })
            .segment_size(MIB as i64)
            .dfs_concurrency(1, 1)
            .dynamic_capacity(true)
            .cache_cap_to_total_data_size_ratio(1.0)
            .cache_capacity_shrink_step(ReadableSize::gb(4))
            .build()
            .unwrap();
        let mgr = IaManager::new(options, Arc::new(InMemFs::new()), None, runtime.into()).unwrap();
        mgr.notify_total_data_size(100 * GIB);
        mgr.update_cache_capacity_config(None, None, Some(0.0), None)
            .unwrap();
        mgr
    }

    #[test]
    fn test_cache_capacity_shrink_converges_to_latest_target_and_grows_immediately() {
        const MIB: u64 = tikv_util::config::MIB;

        let mgr = new_cache_capacity_shrink_test_manager();

        mgr.notify_total_data_size(100 * GIB);
        assert_eq!(
            mgr.cache_queue_capacities(),
            (4 * GIB as i64, 16 * GIB as i64)
        );

        // Replace the target while a shrink is in progress. The next report
        // moves toward the latest target, not the superseded one.
        mgr.update_cache_capacity_config(None, None, Some(0.5), None)
            .unwrap();
        mgr.notify_total_data_size(20 * GIB);
        assert_eq!(
            mgr.cache_queue_capacities(),
            (3 * GIB as i64, 12 * GIB as i64)
        );
        mgr.notify_total_data_size(20 * GIB);
        assert_eq!(
            mgr.cache_queue_capacities(),
            (5 * GIB as i64 / 2, 10 * GIB as i64)
        );

        // Growth is not rate limited, even when the previous shrink has only
        // just converged.
        mgr.update_cache_capacity_config(None, None, Some(1.0), None)
            .unwrap();
        mgr.notify_total_data_size(20 * GIB);
        assert_eq!(
            mgr.cache_queue_capacities(),
            (5 * GIB as i64, 20 * GIB as i64)
        );

        mgr.update_cache_capacity_config(None, None, Some(0.0), None)
            .unwrap();
        for expected_main_gib in [16, 12, 8, 4] {
            mgr.notify_total_data_size(100 * GIB);
            assert_eq!(
                mgr.cache_queue_capacities(),
                (
                    expected_main_gib * GIB as i64 / 4,
                    expected_main_gib * GIB as i64,
                )
            );
        }
        mgr.notify_total_data_size(100 * GIB);
        assert_eq!(
            mgr.cache_queue_capacities(),
            (64 * MIB as i64, 256 * MIB as i64)
        );
        mgr.notify_total_data_size(100 * GIB);
        assert_eq!(
            mgr.cache_queue_capacities(),
            (64 * MIB as i64, 256 * MIB as i64)
        );
    }

    #[test]
    fn test_periodic_reconciliation_bounds_trim_to_shrink_step() {
        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: 100,
                },
                main_queue: QueueOptions {
                    paths: vec![],
                    cap: 300,
                },
            })
            .segment_size(1)
            .dfs_concurrency(1, 1)
            .dynamic_capacity(true)
            .cache_cap_to_total_data_size_ratio(1.0)
            .cache_capacity_shrink_step(ReadableSize(4))
            .build()
            .unwrap();
        let mgr = IaManager::new(options, Arc::new(InMemFs::new()), None, runtime.into()).unwrap();
        mgr.notify_total_data_size(100 * GIB);
        mgr.update_cache_capacity_config(None, None, Some(0.0), None)
            .unwrap();
        let idents: Vec<_> = (0..300)
            .map(|offset| FileSegmentIdent::new(1, offset, offset + 1))
            .collect();
        for &ident in &idents {
            insert_cached_segment(&mgr, ident, true);
        }
        let mgr_for_flush = mgr.clone();
        mgr.runtime_handle()
            .block_on(async move { mgr_for_flush.flush_tasks(Duration::from_secs(5)).await })
            .unwrap();
        assert_eq!(queue_item_counts(&mgr, &idents), (0, 300));

        mgr.notify_total_data_size(0);
        let mgr_for_flush = mgr.clone();
        mgr.runtime_handle()
            .block_on(async move { mgr_for_flush.flush_tasks(Duration::from_secs(5)).await })
            .unwrap();
        assert_eq!(mgr.cache_queue_capacities(), (98, 296));
        assert_eq!(queue_item_counts(&mgr, &idents), (0, 296));

        mgr.notify_total_data_size(0);
        let mgr_for_flush = mgr.clone();
        mgr.runtime_handle()
            .block_on(async move { mgr_for_flush.flush_tasks(Duration::from_secs(5)).await })
            .unwrap();
        assert_eq!(mgr.cache_queue_capacities(), (96, 292));
        assert_eq!(queue_item_counts(&mgr, &idents), (0, 292));
    }

    #[test]
    fn test_desired_main_queue_cap_uses_gib_granularity() {
        assert_eq!(
            IaManagerCore::desired_main_queue_cap(3 * GIB, 0.5),
            GIB as i64
        );
        assert_eq!(
            IaManagerCore::desired_main_queue_cap(4 * GIB, 0.5),
            2 * GIB as i64
        );
        assert_eq!(IaManagerCore::desired_main_queue_cap(GIB - 1, 1.0), 0);
        assert_eq!(IaManagerCore::desired_main_queue_cap(u64::MAX, 0.0), 0);
        assert_eq!(
            IaManagerCore::desired_main_queue_cap(GIB, 0.999_999_999_999_999_6),
            0
        );
        assert_eq!(
            IaManagerCore::desired_main_queue_cap(GIB, 1.000_000_000_000_000_4),
            GIB as i64
        );
        assert_eq!(IaManagerCore::desired_main_queue_cap(u64::MAX, -0.0), 0);
        assert_eq!(
            IaManagerCore::desired_main_queue_cap(u64::MAX, f64::MIN_POSITIVE),
            0
        );
        assert_eq!(
            IaManagerCore::desired_main_queue_cap(90 * GIB, 0.7),
            63 * GIB as i64
        );
        assert_eq!(
            IaManagerCore::desired_main_queue_cap(u64::MAX, f64::MAX),
            i64::MAX / CACHE_CAPACITY_GRANULARITY * CACHE_CAPACITY_GRANULARITY
        );
    }

    #[test]
    fn test_lowering_queue_caps_trims_existing_items_without_new_reads() {
        let runtime = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(1)
            .enable_all()
            .build()
            .unwrap();
        let fs = Arc::new(InMemFs::new());
        let options = IaManagerOptionsBuilder::default()
            .capacity(IaCapacity::Manual {
                small_queue: QueueOptions {
                    paths: vec![],
                    cap: 4,
                },
                main_queue: QueueOptions {
                    paths: vec![],
                    cap: 4,
                },
            })
            .segment_size(1)
            .dfs_concurrency(1, 1)
            .dynamic_capacity(true)
            .build()
            .unwrap();
        let mgr = IaManager::new(options, fs, None, runtime.into()).unwrap();

        let small_idents: Vec<_> = (0..4)
            .map(|offset| FileSegmentIdent::new(1, offset, offset + 1))
            .collect();
        let main_idents: Vec<_> = (4..8)
            .map(|offset| FileSegmentIdent::new(1, offset, offset + 1))
            .collect();
        let all_idents: Vec<_> = small_idents.iter().chain(&main_idents).copied().collect();

        for &ident in &small_idents {
            insert_cached_segment(&mgr, ident, false);
        }
        for &ident in &main_idents {
            insert_cached_segment(&mgr, ident, true);
        }
        let runtime_handle = mgr.runtime_handle();
        let mgr_for_flush = mgr.clone();
        runtime_handle
            .block_on(async move { mgr_for_flush.flush_tasks(Duration::from_secs(5)).await })
            .unwrap();
        assert_eq!(queue_item_counts(&mgr, &all_idents), (4, 4));

        mgr.core.apply_queue_cap_limits_for_test(2, 2).unwrap();
        let runtime_handle = mgr.runtime_handle();
        let mgr_for_flush = mgr.clone();
        runtime_handle
            .block_on(async move { mgr_for_flush.flush_tasks(Duration::from_secs(5)).await })
            .unwrap();
        assert_eq!(queue_item_counts(&mgr, &all_idents), (2, 2));
        for &ident in &main_idents {
            let file_exists = mgr
                .core
                .main_store
                .handle(ident.file_id, &ident.local_filename())
                .unwrap()
                .is_some();
            assert_eq!(
                file_exists,
                mgr.core.segments.get_segment(ident).is_some(),
                "main-store and segment state differ for {ident:?}"
            );
        }
    }

    #[test]
    fn test_new_allows_mem_cap_below_dynamic_floor() {
        let runtime = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(1)
            .enable_all()
            .build()
            .unwrap();
        let fs = Arc::new(InMemFs::new());
        let options = IaManagerOptionsBuilder::default()
            .capacity(IaCapacity::Manual {
                small_queue: QueueOptions {
                    paths: vec![],
                    cap: 63,
                },
                main_queue: QueueOptions {
                    paths: vec![],
                    cap: 1000,
                },
            })
            .segment_size(1)
            .dfs_concurrency(1, 1)
            .dynamic_capacity(true)
            .build()
            .unwrap();
        let mgr = IaManager::new(options, fs, None, runtime.into()).unwrap();

        let config = mgr.core.cache_capacity_config.load();
        assert_eq!(config.small_queue_max_cap, 63);
        assert_eq!(config.main_queue_max_cap, 1000);
        assert_eq!(mgr.core.small_queue_cap.load(Ordering::Relaxed), 63);
        assert_eq!(mgr.core.main_queue_cap.load(Ordering::Relaxed), 256);
    }

    #[test]
    fn test_io_rate_limits_can_be_updated() {
        let runtime = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(1)
            .enable_all()
            .build()
            .unwrap();
        let fs = Arc::new(InMemFs::new());
        let options = IaManagerOptionsBuilder::default()
            .capacity(IaCapacity::Manual {
                small_queue: QueueOptions {
                    paths: vec![],
                    cap: 100,
                },
                main_queue: QueueOptions {
                    paths: vec![],
                    cap: 100,
                },
            })
            .remote_read_max_throughput(ReadableSize::mb(128))
            .local_write_max_throughput(ReadableSize::mb(64))
            .build()
            .unwrap();
        let mgr = IaManager::new(options, fs, None, runtime.into()).unwrap();

        assert_eq!(
            mgr.core.io_limiter.remote_read_speed_limit(),
            ReadableSize::mb(128).0 as f64
        );
        assert_eq!(
            mgr.core.io_limiter.local_write_speed_limit(),
            ReadableSize::mb(64).0 as f64
        );

        mgr.update_io_rate_limits(ReadableSize::mb(32), ReadableSize(0));
        assert_eq!(
            mgr.core.io_limiter.remote_read_speed_limit(),
            ReadableSize::mb(32).0 as f64
        );
        assert_eq!(mgr.core.io_limiter.local_write_speed_limit(), f64::INFINITY);
    }
}