// Copyright 2018 TiKV Project Authors. Licensed under Apache-2.0.
// Modifications Copyright 2022 PingCAP, Inc.

mod encryption_meta;
mod encryption_meta_sync;
use std::{
    borrow::Cow,
    convert::{Infallible, TryInto},
    env::args,
    error::Error as StdError,
    net::SocketAddr,
    ops,
    str::{self, FromStr},
    sync::{
        Arc,
        atomic::{AtomicU64, AtomicUsize, Ordering},
    },
    time::Duration,
};

use api_version::{ApiV2, api_v2::TXN_KEY_PREFIX};
use builtin_dfs::BuiltinDfsFileMeta;
use bytes::{Buf, BufMut, Bytes, BytesMut};
use collections::{HashMap, HashMapExt, HashSet};
use concurrency_manager::ConcurrencyManager;
use encryption_conf_change::{
    CancelRolloutJobRequest, EncryptionConfChangeError, EncryptionConfChangeManager,
    OwnerCheckMode, StartRolloutJobRequest,
};
use file_system::IoType;
use flate2::{Compression, write::GzEncoder};
use futures::{compat::Compat01As03, future::ok, prelude::*};
use hyper::{
    self, Body, Method, Request, Response, Server, StatusCode, header,
    header::{ACCEPT_ENCODING, CONTENT_ENCODING, CONTENT_TYPE, HeaderValue},
    server::{
        Builder as HyperBuilder,
        accept::Accept,
        conn::{AddrIncoming, AddrStream},
    },
    service::{make_service_fn, service_fn},
};
use keys::next_key;
use kvengine::{
    ENCRYPTION_KEY, ENCRYPTION_META_KEY, GLOBAL_SHARD_END_KEY, IdVer, Shard, ShardStats, ShardTag,
    SnapAccess, WRITE_CF, WRITE_CF_BOTTOM_LEVEL,
    dfs::FileType,
    table::{BoundedDataSet, DataBound, InnerKey, SnapVersion},
    util::new_table_create_pb,
};
use kvenginepb::ChangeSet;
use kvproto::{
    coprocessor::DelegateResponse,
    errorpb, metapb,
    metapb::BucketStats,
    pdpb,
    pdpb::{Peers, SyncRegionResponse},
    raft_serverpb::{PeerState, StoreIdent},
};
use online_config::OnlineConfig;
use pd_client::PdClient;
use prometheus::TEXT_FORMAT;
use protobuf::Message;
use raftstore::{RegionInfo, RegionInfoAccessor, coprocessor::RegionInfoProvider};
use rfengine::{
    Error, RAFT_TRUNCATED_STATE_KEY, RfEngine, WriteBatch, load_store_ident, raft_state_key,
};
use rfstore::{
    RaftRouter, RaftStoreRouter,
    store::{
        Callback, CasualMessage, RAFT_INIT_LOG_INDEX, RAFT_INIT_LOG_TERM, RegionSnapshot, StoreMsg,
        TERM_KEY,
        peer_storage::{collect_prefix_regions, load_raft_engine_meta, load_region_state},
        state::RaftState,
        write_engine_meta_bytes, write_peer_state,
    },
};
use schema::schema::{StorageClass, StorageClassSpec};
use security::{self, SecurityConfig, TlsStream};
use serde_json::{Value, json};
use tidb_query_datatype::codec::table::encode_row_key_prefix;
use tikv::{
    config::{ConfigController, LogLevel},
    server::{config::Config as ServerConfig, status_server::profile::start_one_cpu_profile},
    storage::CloudStore,
};
use tikv_util::{
    Either, GLOBAL_SERVER_READINESS,
    codec::bytes::{decode_bytes, encode_bytes},
    config::{AbsoluteOrPercentSize, ReadableSize, VersionTrack},
    future::paired_future_callback,
    http::{CONTENT_TYPE_PROTOBUF, HeaderExt},
    init_task_local_sync,
    logger::set_log_level,
    memory::MemoryLimiterGuard,
    metrics::{dump, dump_to},
    shutdown_runtimes,
    store::find_peer,
    sys::{
        SysQuota,
        thread::{StdThreadBuildWrapper, ThreadBuildWrapper},
    },
    time::{Instant, UnixSecs},
    timer::GLOBAL_TIMER_HANDLE,
};
use tokio::{
    io::{AsyncRead, AsyncWrite},
    runtime::{Builder, Runtime},
    sync::{
        OwnedSemaphorePermit, Semaphore,
        oneshot::{self, Receiver, Sender},
    },
    task::JoinSet,
};
use txn_types::{TimeStamp, TsSet};

use crate::{
    server::Result,
    status_server::metrics::{
        DFS_FILE_READ_ACTIVE, DFS_FILE_READ_IN_FLIGHT_BYTES, DFS_FILE_READ_REQUESTED_BYTES,
        DFS_FILE_READ_RESPONSE_BODY_DURATION, DFS_FILE_READ_UNSUCCESSFUL_TOTAL,
        SNAPSHOT_BUILD_DURATION, SNAPSHOT_CACHE_VALIDATION_TOTAL, SNAPSHOT_LOCK_CHECK_DURATION,
        SNAPSHOT_PAYLOAD_BYTES, STATUS_REQUEST_DURATION,
    },
    tikv_util::codec::number::NumberEncoder,
};

pub(crate) mod metrics;

static TIMER_CANCELED: &str = "tokio timer canceled";

#[cfg(feature = "failpoints")]
static MISSING_NAME: &[u8] = b"Missing param name";
#[cfg(feature = "failpoints")]
static MISSING_ACTIONS: &[u8] = b"Missing param actions";
#[cfg(feature = "failpoints")]
static FAIL_POINTS_REQUEST_PATH: &str = "/fail";

const SERVER_READ_TIMEOUT: Duration = Duration::from_secs(600);
const SERVER_TCP_KEEPALIVE: Duration = Duration::from_secs(120);

const BACKUP_TS_WAIT_RETRY_INTERVAL: Duration = Duration::from_secs(1);

const STORES_INFO_CACHE_TTL: Duration = Duration::from_secs(5);
// Maximum TTL when fail to get stores info from PD.
const STORES_INFO_CACHE_MAX_TTL: Duration = Duration::from_secs(60);

#[derive(Debug)]
struct DfsFileReadRequestGuard(Arc<AtomicUsize>);

impl DfsFileReadRequestGuard {
    fn try_acquire(in_flight: &Arc<AtomicUsize>, limit: usize) -> std::result::Result<Self, usize> {
        in_flight
            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| {
                (current < limit).then_some(current + 1)
            })
            .map(|_| Self(in_flight.clone()))
    }
}

impl Drop for DfsFileReadRequestGuard {
    fn drop(&mut self) {
        let previous = self.0.fetch_sub(1, Ordering::Release);
        debug_assert!(previous > 0);
    }
}

struct DfsFileReadAdmissionGuard {
    // Field order is intentional: release the byte reservation before making
    // the request slot visible to another caller.
    _bytes: MemoryLimiterGuard,
    _request: DfsFileReadRequestGuard,
    charged_bytes: u64,
    response_started_at: Option<Instant>,
}

impl DfsFileReadAdmissionGuard {
    fn new(
        request: DfsFileReadRequestGuard,
        bytes: MemoryLimiterGuard,
        charged_bytes: u64,
    ) -> Self {
        DFS_FILE_READ_ACTIVE.inc();
        DFS_FILE_READ_IN_FLIGHT_BYTES.add(charged_bytes as i64);
        Self {
            _bytes: bytes,
            _request: request,
            charged_bytes,
            response_started_at: None,
        }
    }

    fn start_response(&mut self) {
        self.response_started_at = Some(Instant::now());
    }
}

impl Drop for DfsFileReadAdmissionGuard {
    fn drop(&mut self) {
        DFS_FILE_READ_ACTIVE.dec();
        DFS_FILE_READ_IN_FLIGHT_BYTES.sub(self.charged_bytes as i64);
        if let Some(started_at) = self.response_started_at {
            DFS_FILE_READ_RESPONSE_BODY_DURATION.observe(started_at.saturating_elapsed_secs());
        }
    }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum DfsFileReadRangeError {
    StartAfterEnd,
    EndAfterFile,
}

enum DfsFileReadTaskResult {
    Success {
        data: Bytes,
        admission: DfsFileReadAdmissionGuard,
    },
    Rejected(kvengine::limiter::DfsCacheMemoryRejection),
    InvalidRange(DfsFileReadRangeError),
    SizeError(kvengine::Error),
    ReadError(kvengine::Error),
    SizeChanged {
        expected: u64,
        actual: u64,
    },
}
const SNAPSHOT_CACHE_CAPABILITY_HEADER: &str = "x-cse-snapshot-cache-version";
const SNAPSHOT_CACHE_VALIDATION_HIT: &str = "hit";
const SNAPSHOT_CACHE_VALIDATION_LOCKED: &str = "locked";
const SNAPSHOT_CACHE_VALIDATION_MISS: &str = "miss";
const SNAPSHOT_LOCK_CHECK_OK: &str = "ok";
const SNAPSHOT_LOCK_CHECK_LOCKED: &str = "locked";
const SNAPSHOT_LOCK_CHECK_ERROR: &str = "error";
const SNAPSHOT_PAYLOAD_MEMTABLE: &str = "memtable";
const SNAPSHOT_PAYLOAD_SNAPSHOT: &str = "snapshot";
const SNAPSHOT_PAYLOAD_RESPONSE: &str = "response";
/// Current metadata-only file-reference response version.
pub const ONLINE_SNAPSHOT_CAPABILITY_VERSION: u32 = 1;
const ONLINE_SNAPSHOT_MAX_RESPONSE_BYTES: usize = 512 * 1024 * 1024;
const ONLINE_SNAPSHOT_MEMORY_AMPLIFICATION: usize = 2;
const ONLINE_SNAPSHOT_MEMORY_UNIT_BYTES: usize = 16 * 1024 * 1024;
const ONLINE_SNAPSHOT_MEMORY_BUDGET_BYTES: usize =
    ONLINE_SNAPSHOT_MAX_RESPONSE_BYTES * ONLINE_SNAPSHOT_MEMORY_AMPLIFICATION;
const ONLINE_SNAPSHOT_MEMORY_PERMITS: usize =
    ONLINE_SNAPSHOT_MEMORY_BUDGET_BYTES / ONLINE_SNAPSHOT_MEMORY_UNIT_BYTES;
const ONLINE_SNAPSHOT_PREPARE_WAIT_TIMEOUT: Duration = Duration::from_secs(10 * 60);
const ONLINE_SNAPSHOT_BACKUP_TS_TTL: Duration = Duration::from_secs(60 * 60);
/// Upper bounds for caller-controlled online-snapshot waits and renewable
/// leases. Besides limiting resource retention, these keep durations within
/// the range supported by platform timers.
const ONLINE_SNAPSHOT_MAX_WAIT_TIMEOUT: Duration = Duration::from_secs(10 * 60);
const ONLINE_SNAPSHOT_MAX_TTL: Duration = Duration::from_secs(24 * 60 * 60);

// When doing unsafe recover, add this delta to term and last_index.
const UNSAFE_RECOVER_DELTA: u64 = 64;

fn observe_snapshot_cache_validation(result: &'static str) {
    SNAPSHOT_CACHE_VALIDATION_TOTAL
        .with_label_values(&[result])
        .inc();
}

fn observe_snapshot_lock_check(result: &'static str, lock_check_start: Instant) {
    SNAPSHOT_LOCK_CHECK_DURATION
        .with_label_values(&[result])
        .observe(lock_check_start.saturating_elapsed().as_secs_f64());
}

fn observe_snapshot_build(build_start: Instant) {
    SNAPSHOT_BUILD_DURATION.observe(build_start.saturating_elapsed().as_secs_f64());
}

fn observe_snapshot_payload_bytes(part: &'static str, bytes: usize) {
    SNAPSHOT_PAYLOAD_BYTES
        .with_label_values(&[part])
        .observe(bytes as f64);
}

fn online_snapshot_dump_concurrency(cpu_cores: f64) -> usize {
    ((cpu_cores / 2.0) as usize).max(1)
}

fn online_snapshot_memory_permits(mem_table_kv_size: usize) -> u32 {
    mem_table_kv_size
        .saturating_mul(ONLINE_SNAPSHOT_MEMORY_AMPLIFICATION)
        .div_ceil(ONLINE_SNAPSHOT_MEMORY_UNIT_BYTES)
        .clamp(1, ONLINE_SNAPSHOT_MEMORY_PERMITS) as u32
}

#[derive(Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
struct LogLevelRequest {
    pub log_level: LogLevel,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct SyncRegionRequest {
    pub start: String,
    pub end: String,
    pub limit: usize,
    pub reverse: bool,
    pub debug: bool,
}

#[derive(Deserialize)]
#[serde(rename_all = "kebab-case")]
struct SyncRegionByIdRequest {
    pub region_id: u64,
    #[serde(default)]
    pub debug: bool,
}

use encryption_meta::{DeriveResult, EncryptionMetaGenMode, parse_existing_meta};
pub use encryption_meta::{EncryptionMetaGenRequest, EncryptionMetaGenResponse};

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct EncryptionMetaSyncLatestRequest {
    pub keyspace_id: u32,
    pub region_id: u64,
    pub current_file_id: u64,
    pub next_file_id: u64,
    pub expected_region_version: u64,
    pub expected_region_conf_ver: u64,
}

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct EncryptionMetaSyncLatestResponse {
    pub file_id: u64,
    pub data_key_id: u32,
    pub synced: bool,
}

/// One typed immutable file reference in an online snapshot metadata response.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
pub struct OnlineSnapshotFileRef {
    pub file_type: u8,
    pub file_id: u64,
}

/// Metadata-only referenced-file response for one coherent shard snapshot.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
pub struct OnlineSnapshotFileRefsResponse {
    pub version: u32,
    pub store_id: u64,
    pub shard_id: u64,
    pub shard_ver: u64,
    pub outer_start: Vec<u8>,
    pub outer_end: Vec<u8>,
    pub meta_sequence: u64,
    pub write_sequence: u64,
    pub files: Vec<OnlineSnapshotFileRef>,
}

enum ShardLeaderCheck {
    Leader,
    RegionError(Vec<u8>),
}

struct OnlineSnapshotShardRequest {
    shard_id: u64,
    shard_ver: u64,
    branch_ts: Option<u64>,
}

struct OnlineSnapshotAccess {
    store_id: u64,
    snap_access: SnapAccess,
    term: u64,
    _dump_permit: OwnedSemaphorePermit,
}

pub struct StatusServer {
    thread_pool: Arc<Runtime>,
    close_tx: Sender<()>,
    close_rx: Option<Receiver<()>>,
    close_handle: Option<std::thread::JoinHandle<()>>,
    addr: Option<SocketAddr>,
    server_config: Arc<VersionTrack<ServerConfig>>,
    cfg_controller: ConfigController,
    router: RaftRouter,
    security_config: Arc<SecurityConfig>,
    kvengine: kvengine::Engine,
    rfengine: rfengine::RfEngine,
    concurrency_manager: ConcurrencyManager,
    pd_client: Arc<dyn PdClient>,
    region_info_accessor: RegionInfoAccessor,
}

#[derive(Default, Debug, Serialize, Deserialize, Clone, PartialEq)]
#[serde(default)]
#[serde(rename_all = "kebab-case")]
pub struct SstMeta {
    pub id: u64,
    pub smallest: Vec<u8>,
    pub biggest: Vec<u8>,
    pub meta_offset: u32,
    pub commit_ts: u64,
    pub size: usize,
    pub uncompressed_size: usize,
    pub keys: usize,
}

impl StatusServer {
    pub fn new(
        status_thread_pool_size: usize,
        server_config: Arc<VersionTrack<ServerConfig>>,
        cfg_controller: ConfigController,
        security_config: Arc<SecurityConfig>,
        router: RaftRouter,
        kvengine: kvengine::Engine,
        rfengine: rfengine::RfEngine,
        concurrency_manager: ConcurrencyManager,
        pd_client: Arc<dyn PdClient>,
        region_info_accessor: RegionInfoAccessor,
    ) -> Result<Self> {
        let thread_pool = Builder::new_multi_thread()
            .enable_all()
            .worker_threads(status_thread_pool_size)
            .thread_name("status-server")
            .after_start_wrapper(|| debug!("Status server started"))
            .before_stop_wrapper(|| debug!("stopping status server"))
            .build()?;

        let (close_tx, close_rx) = oneshot::channel::<()>();
        Ok(StatusServer {
            thread_pool: Arc::new(thread_pool),
            close_tx,
            close_rx: Some(close_rx),
            close_handle: None,
            addr: None,
            server_config,
            cfg_controller,
            router,
            security_config,
            kvengine,
            rfengine,
            concurrency_manager,
            pd_client,
            region_info_accessor,
        })
    }

    fn load_raft_state(rf: &RfEngine, peer_id: u64, ver: u64) -> Option<RaftState> {
        let key = raft_state_key(ver);
        let raft_state_val = match rf.get_state(peer_id, key.chunk()) {
            Some(val) => val,
            None => {
                error!("failed to load raft state");
                return None;
            }
        };
        let mut raft_state = RaftState::default();
        raft_state.unmarshal(&raft_state_val);
        Some(raft_state)
    }

    fn write_empty_engine_meta(
        rf: &RfEngine,
        wb: &mut WriteBatch,
        peer_id: u64,
        region_id: u64,
        ver: u64,
        keyspace_id: u32,
        index: u64,
        term: u64,
        inner_key_off: u32,
        new_cs: Option<ChangeSet>,
        encryption_key: Option<Bytes>,
        encryption_meta: Option<Bytes>,
    ) -> Result<kvenginepb::ChangeSet> {
        let region_local_state = match load_region_state(rf, peer_id, ver) {
            Some(val) => val,
            None => {
                return Err(crate::server::Error::Other(box_err!(
                    "region state not found"
                )));
            }
        };
        write_peer_state(
            wb,
            peer_id,
            region_local_state.get_region(),
            PeerState::Normal,
            None,
        );
        let mut enc_start_key = region_local_state.get_region().get_start_key();
        let mut enc_end_key = region_local_state.get_region().get_end_key();
        let start_key = decode_bytes(&mut enc_start_key, false).unwrap();
        let end_key = decode_bytes(&mut enc_end_key, false).unwrap();

        let mut cs = new_cs.unwrap_or_default();
        cs.set_shard_id(region_id);
        cs.set_shard_ver(ver);
        cs.set_sequence(index);
        let snap = cs.mut_snapshot();
        snap.set_outer_start(start_key.to_vec());
        snap.set_outer_end(end_key.to_vec());
        snap.set_inner_key_off(inner_key_off);
        snap.set_data_sequence(index);
        let props = snap.mut_properties();
        props.set_shard_id(region_id);
        props.mut_keys().push(TERM_KEY.to_string());
        props.mut_values().push(term.to_le_bytes().to_vec());
        if encryption_meta.is_none() {
            if let Some(encryption_key) = encryption_key {
                props.mut_keys().push(ENCRYPTION_KEY.to_string());
                props.mut_values().push(encryption_key.to_vec());
            }
        }
        if let Some(encryption_meta) = encryption_meta {
            props.mut_keys().push(ENCRYPTION_META_KEY.to_string());
            props.mut_values().push(encryption_meta.to_vec());
        }

        let cs_val = cs.write_to_bytes().unwrap();
        write_engine_meta_bytes(wb, peer_id, region_id, keyspace_id, &cs_val);
        Ok(cs)
    }

    fn write_truncated_state(
        wb: &mut WriteBatch,
        peer_id: u64,
        region_id: u64,
        keyspace_id: u32,
        index: u64,
        term: u64,
    ) {
        let mut ts_val = BytesMut::with_capacity(16);
        ts_val.put_u64_le(term);
        ts_val.put_u64_le(index);
        let ts = ts_val.freeze();
        wb.set_state(
            peer_id,
            region_id,
            keyspace_id,
            RAFT_TRUNCATED_STATE_KEY,
            ts.chunk(),
        );
    }

    fn write_raft_state(
        wb: &mut WriteBatch,
        peer_id: u64,
        region_id: u64,
        ver: u64,
        keyspace_id: u32,
        index: u64,
        term: u64,
    ) -> RaftState {
        let mut rs_val = BytesMut::with_capacity(40);
        rs_val.put_u64_le(term);
        rs_val.put_u64_le(0);
        rs_val.put_u64_le(index);
        rs_val.put_u64_le(index);
        rs_val.put_u64_le(index);
        let rs = rs_val.freeze();
        let key = raft_state_key(ver);
        wb.set_state(peer_id, region_id, keyspace_id, key.chunk(), rs.chunk());
        let mut raft_state = RaftState::default();
        raft_state.unmarshal(rs.chunk());
        raft_state
    }

    #[cfg(not(test))]
    fn dump_heap_profile_pprof_bytes() -> std::result::Result<Vec<u8>, String> {
        memory_profile::dump_heap_profile_pprof()
    }

    #[cfg(test)]
    fn dump_heap_profile_pprof_bytes() -> std::result::Result<Vec<u8>, String> {
        Ok(Vec::new())
    }

    #[cfg(not(test))]
    fn dump_heap_profile_svg_bytes() -> std::result::Result<Vec<u8>, String> {
        memory_profile::dump_heap_profile_svg()
    }

    #[cfg(test)]
    fn dump_heap_profile_svg_bytes() -> std::result::Result<Vec<u8>, String> {
        Ok(Vec::new())
    }

    pub fn dump_heap_prof_to_resp(req: Request<Body>) -> hyper::Result<Response<Body>> {
        let query = req.uri().query().unwrap_or("");
        let query_pairs: HashMap<_, _> = url::form_urlencoded::parse(query.as_bytes()).collect();

        let use_svg = query_pairs.get("svg").map(|x| x.as_ref()) == Some("true");
        let result = if use_svg {
            Self::dump_heap_profile_svg_bytes()
        } else {
            Self::dump_heap_profile_pprof_bytes()
        };

        match result {
            Ok(body) => {
                info!("dump or get heap profile successfully");
                let file_name = if use_svg {
                    "profile.svg"
                } else {
                    "profile.pprof"
                };
                let mut response = Response::builder()
                    .header("X-Content-Type-Options", "nosniff")
                    .header(
                        "Content-Disposition",
                        format!("attachment; filename=\"{}\"", file_name),
                    )
                    .header("Content-Length", body.len());
                response = if use_svg {
                    response.header("Content-Type", mime::IMAGE_SVG.to_string())
                } else {
                    response.header("Content-Type", mime::APPLICATION_OCTET_STREAM.to_string())
                };
                Ok(response.body(body.into()).unwrap())
            }
            Err(e) => {
                info!("dump or get heap profile fail: {}", e);
                Ok(make_response(StatusCode::INTERNAL_SERVER_ERROR, e))
            }
        }
    }

    async fn get_config(
        req: Request<Body>,
        cfg_controller: &ConfigController,
    ) -> hyper::Result<Response<Body>> {
        let mut full = false;
        if let Some(query) = req.uri().query() {
            let query_pairs: HashMap<_, _> =
                url::form_urlencoded::parse(query.as_bytes()).collect();
            full = match query_pairs.get("full") {
                Some(val) => match val.parse() {
                    Ok(val) => val,
                    Err(err) => return Ok(make_response(StatusCode::BAD_REQUEST, err.to_string())),
                },
                None => false,
            };
        }
        let encode_res = if full {
            // Get all config
            serde_json::to_string(&cfg_controller.get_current())
        } else {
            // Filter hidden config
            serde_json::to_string(&cfg_controller.get_current().get_encoder())
        };
        Ok(match encode_res {
            Ok(json) => Response::builder()
                .header(header::CONTENT_TYPE, "application/json")
                .body(Body::from(json))
                .unwrap(),
            Err(_) => make_response(StatusCode::INTERNAL_SERVER_ERROR, "Internal Server Error"),
        })
    }

    pub fn get_cmdline(_req: Request<Body>) -> hyper::Result<Response<Body>> {
        let args = args().fold(String::new(), |mut a, b| {
            a.push_str(&b);
            a.push('\x00');
            a
        });
        let response = Response::builder()
            .header("Content-Type", mime::TEXT_PLAIN.to_string())
            .header("X-Content-Type-Options", "nosniff")
            .body(args.into())
            .unwrap();
        Ok(response)
    }

    pub fn get_symbol_count(req: Request<Body>) -> hyper::Result<Response<Body>> {
        assert_eq!(req.method(), Method::GET);
        // We don't know how many symbols we have, but we
        // do have symbol information. pprof only cares whether
        // this number is 0 (no symbols available) or > 0.
        let text = "num_symbols: 1\n";
        let response = Response::builder()
            .header("Content-Type", mime::TEXT_PLAIN.to_string())
            .header("X-Content-Type-Options", "nosniff")
            .header("Content-Length", text.len())
            .body(text.into())
            .unwrap();
        Ok(response)
    }

    // The request and response format follows pprof remote server
    // https://gperftools.github.io/gperftools/pprof_remote_servers.html
    // Here is the go pprof implementation:
    // https://github.com/golang/go/blob/3857a89e7eb872fa22d569e70b7e076bec74ebbb/src/net/http/pprof/pprof.go#L191
    pub async fn get_symbol(req: Request<Body>) -> hyper::Result<Response<Body>> {
        assert_eq!(req.method(), Method::POST);
        let mut text = String::new();
        let body_bytes = hyper::body::to_bytes(req.into_body()).await?;
        let body = String::from_utf8(body_bytes.to_vec());
        if body.is_err() {
            return Ok(make_response(
                StatusCode::INTERNAL_SERVER_ERROR,
                "Body is not a valid UTF8",
            ));
        }
        let body = body.unwrap();

        // The request body is a list of addr to be resolved joined by '+'.
        // Resolve addrs with addr2line and write the symbols each per line in
        // response.
        for pc in body.split('+') {
            let addr = usize::from_str_radix(pc.trim_start_matches("0x"), 16).unwrap_or(0);
            if addr == 0 {
                info!("invalid addr: {}", addr);
                continue;
            }

            // Would be multiple symbols if inlined.
            let mut syms = vec![];
            backtrace::resolve(addr as *mut std::ffi::c_void, |sym| {
                let name = sym
                    .name()
                    .unwrap_or_else(|| backtrace::SymbolName::new(b"<unknown>"));
                syms.push(name.to_string());
            });

            if !syms.is_empty() {
                // join inline functions with '--'
                let f = syms.join("--");
                // should be <hex address> <function name>
                text.push_str(format!("{:#x} {}\n", addr, f).as_str());
            } else {
                info!("can't resolve mapped addr: {:#x}", addr);
                text.push_str(format!("{:#x} ??\n", addr).as_str());
            }
        }
        let response = Response::builder()
            .header("Content-Type", mime::TEXT_PLAIN.to_string())
            .header("X-Content-Type-Options", "nosniff")
            .header("Content-Length", text.len())
            .body(text.into())
            .unwrap();
        Ok(response)
    }

    async fn update_config(
        cfg_controller: ConfigController,
        req: Request<Body>,
    ) -> hyper::Result<Response<Body>> {
        let mut body = Vec::new();
        req.into_body()
            .try_for_each(|bytes| {
                body.extend(bytes);
                ok(())
            })
            .await?;
        Ok(match decode_json(&body) {
            Ok(change) => match cfg_controller.update(change) {
                Err(e) => make_response(
                    StatusCode::INTERNAL_SERVER_ERROR,
                    format!("failed to update, error: {:?}", e),
                ),
                Ok(_) => {
                    let mut resp = Response::default();
                    *resp.status_mut() = StatusCode::OK;
                    resp
                }
            },
            Err(e) => make_response(
                StatusCode::INTERNAL_SERVER_ERROR,
                format!("failed to decode, error: {:?}", e),
            ),
        })
    }

    pub async fn dump_cpu_prof_to_resp(req: Request<Body>) -> hyper::Result<Response<Body>> {
        let query = req.uri().query().unwrap_or("");
        let query_pairs: HashMap<_, _> = url::form_urlencoded::parse(query.as_bytes()).collect();

        let seconds: u64 = match query_pairs.get("seconds") {
            Some(val) => match val.parse() {
                Ok(val) => val,
                Err(err) => return Ok(make_response(StatusCode::BAD_REQUEST, err.to_string())),
            },
            None => 10,
        };

        let frequency: i32 = match query_pairs.get("frequency") {
            Some(val) => match val.parse() {
                Ok(val) => val,
                Err(err) => return Ok(make_response(StatusCode::BAD_REQUEST, err.to_string())),
            },
            None => 99, /* Default frequency of sampling. 99Hz to avoid coincide with special
                         * periods */
        };

        let prototype_content_type: hyper::http::HeaderValue =
            hyper::http::HeaderValue::from_str("application/protobuf").unwrap();
        let output_protobuf = req.headers().get("Content-Type") == Some(&prototype_content_type);

        let timer =
            GLOBAL_TIMER_HANDLE.delay(std::time::Instant::now() + Duration::from_secs(seconds));
        let end = async move {
            Compat01As03::new(timer)
                .await
                .map_err(|_| TIMER_CANCELED.to_owned())
        };
        match start_one_cpu_profile(end, frequency, output_protobuf).await {
            Ok(body) => {
                info!("dump cpu profile successfully");
                Ok(make_response(StatusCode::OK, body))
            }
            Err(e) => {
                info!("dump cpu profile fail: {}", e);
                Ok(make_response(StatusCode::INTERNAL_SERVER_ERROR, e))
            }
        }
    }

    async fn change_log_level(req: Request<Body>) -> hyper::Result<Response<Body>> {
        let mut body = Vec::new();
        req.into_body()
            .try_for_each(|bytes| {
                body.extend(bytes);
                ok(())
            })
            .await?;

        let log_level_request: std::result::Result<LogLevelRequest, serde_json::error::Error> =
            serde_json::from_slice(&body);

        match log_level_request {
            Ok(req) => {
                set_log_level(req.log_level.into());
                Ok(Response::new(Body::empty()))
            }
            Err(err) => Ok(make_response(StatusCode::BAD_REQUEST, err.to_string())),
        }
    }

    // URI: /kvengine/columnar_status?keyspace_id=xxx&table_id=xxx[&index_id=xxx]
    // Collect the columnar replica status of the table
    fn collect_columnar_status(
        req: Request<Body>,
        ctx: &StatusContext,
    ) -> hyper::Result<Response<Body>> {
        let query = req.uri().query().unwrap_or("");
        let query_pairs: HashMap<_, _> = url::form_urlencoded::parse(query.as_bytes()).collect();

        if !query_pairs.contains_key("keyspace_id") || !query_pairs.contains_key("table_id") {
            return Ok(make_response(
                StatusCode::BAD_REQUEST,
                "keyspace_id and table_id are required".to_string(),
            ));
        }
        let keyspace_id = match u32::from_str(query_pairs.get("keyspace_id").unwrap()) {
            Ok(id) => id,
            Err(err) => return Ok(make_response(StatusCode::BAD_REQUEST, err.to_string())),
        };
        let table_id = match i64::from_str(query_pairs.get("table_id").unwrap()) {
            Ok(id) => id,
            Err(err) => return Ok(make_response(StatusCode::BAD_REQUEST, err.to_string())),
        };
        let index_id = if query_pairs.contains_key("index_id") {
            match i64::from_str(query_pairs.get("index_id").unwrap()) {
                Ok(id) => Some(id),
                Err(err) => return Ok(make_response(StatusCode::BAD_REQUEST, err.to_string())),
            }
        } else {
            None
        };
        let mut table_prefix_key = api_version::ApiV2::get_txn_keyspace_prefix(keyspace_id);
        table_prefix_key.extend_from_slice(&encode_row_key_prefix(table_id));
        let table_prefix_end_key = next_key(&table_prefix_key);
        let encoded_table_prefix_key = encode_bytes(&table_prefix_key);
        let encoded_table_prefix_end_key = encode_bytes(&table_prefix_end_key);
        let ids = ctx
            .region_info_accessor
            .get_region_ids_in_range(encoded_table_prefix_key, encoded_table_prefix_end_key);
        let resp = ctx
            .kvengine
            .collect_columnar_status(keyspace_id, table_id, index_id, ids);
        let resp_json = serde_json::to_string_pretty(&resp).unwrap();
        Ok(Response::builder()
            .header(header::CONTENT_TYPE, "application/json")
            .body(Body::from(resp_json))
            .unwrap())
    }

    // URI: /kvengine/storage_class_status
    // Query: keyspace_id=xxx&table_id=xxx&target=IA|STANDARD
    // Collect the applied storage class status of the local table replicas.
    fn collect_storage_class_status(
        req: Request<Body>,
        ctx: &StatusContext,
    ) -> hyper::Result<Response<Body>> {
        let query = req.uri().query().unwrap_or("");
        let query_pairs: HashMap<_, _> = url::form_urlencoded::parse(query.as_bytes()).collect();
        if !query_pairs.contains_key("keyspace_id")
            || !query_pairs.contains_key("table_id")
            || !query_pairs.contains_key("target")
        {
            return Ok(make_response(
                StatusCode::BAD_REQUEST,
                "keyspace_id, table_id and target are required".to_string(),
            ));
        }
        let keyspace_id = match u32::from_str(query_pairs.get("keyspace_id").unwrap()) {
            Ok(id) => id,
            Err(err) => return Ok(make_response(StatusCode::BAD_REQUEST, err.to_string())),
        };
        let table_id = match i64::from_str(query_pairs.get("table_id").unwrap()) {
            Ok(id) => id,
            Err(err) => return Ok(make_response(StatusCode::BAD_REQUEST, err.to_string())),
        };
        let target = match query_pairs.get("target").unwrap().as_ref() {
            "IA" => StorageClassSpec::from(StorageClass::Ia),
            // CSE represents STANDARD by omitting the storage class property.
            "STANDARD" => StorageClassSpec::default(),
            target => {
                return Ok(make_response(
                    StatusCode::BAD_REQUEST,
                    format!("unsupported target storage class: {target}"),
                ));
            }
        };

        let mut table_prefix_key = api_version::ApiV2::get_txn_keyspace_prefix(keyspace_id);
        table_prefix_key.extend(kvengine::table_id::encode_table_prefix_key(table_id).to_vec());
        let table_prefix_end_key = next_key(&table_prefix_key);
        let encoded_table_prefix_key = encode_bytes(&table_prefix_key);
        let encoded_table_prefix_end_key = encode_bytes(&table_prefix_end_key);
        let ids = ctx
            .region_info_accessor
            .get_region_ids_in_range(encoded_table_prefix_key, encoded_table_prefix_end_key);
        let resp = ctx
            .kvengine
            .collect_storage_class_status(keyspace_id, table_id, target, ids);
        let resp_json = serde_json::to_string_pretty(&resp).unwrap();
        Ok(Response::builder()
            .header(header::CONTENT_TYPE, "application/json")
            .body(Body::from(resp_json))
            .unwrap())
    }

    /// URI: /kvengine/columnar_index_stats?keyspace_id=xxx
    /// Returns the columnar index coverage statistics for all tables and
    /// indexes in the keyspace
    async fn collect_columnar_index_stats(
        req: Request<Body>,
        ctx: StatusContext,
    ) -> hyper::Result<Response<Body>> {
        let query = req.uri().query().unwrap_or("");
        let query_pairs: HashMap<_, _> = url::form_urlencoded::parse(query.as_bytes()).collect();
        if !query_pairs.contains_key("keyspace_id") {
            return Ok(make_response(
                StatusCode::BAD_REQUEST,
                "keyspace_id is required".to_string(),
            ));
        }
        let keyspace_id = match u32::from_str(query_pairs.get("keyspace_id").unwrap()) {
            Ok(id) => id,
            Err(err) => return Ok(make_response(StatusCode::BAD_REQUEST, err.to_string())),
        };
        // collect_columnar_index_stats may have blocking IO, so run it in a blocking
        // thread pool
        let engine = ctx.kvengine.clone();
        let join_handle = ctx.thread_pool.spawn_blocking(move || {
            init_task_local_sync(|| {
                engine
                    .collect_columnar_index_stats(keyspace_id)
                    .map(|stats| serde_json::to_string_pretty(&stats).unwrap())
            })
        });
        let stats_json = match join_handle.await {
            Ok(Ok(json)) => json,
            Ok(Err(err)) => {
                return Ok(make_response(
                    StatusCode::INTERNAL_SERVER_ERROR,
                    format!("failed to collect columnar index stats: {}", err),
                ));
            }
            Err(err) => {
                return Ok(make_response(
                    StatusCode::INTERNAL_SERVER_ERROR,
                    format!("failed to collect columnar index stats: {}", err),
                ));
            }
        };
        Ok(Response::builder()
            .header(header::CONTENT_TYPE, "application/json")
            .body(Body::from(stats_json))
            .unwrap())
    }

    // POST /kvengine/online-snapshot/prepare?branch_ts=xxx
    //
    // Installs the existing Store-wide backup-ts boundary and waits until all
    // older tracked transactions have finished applying their authoritative
    // commit record. A successful response is the completeness barrier for
    // subsequent online snapshot captures on this Store.
    async fn prepare_online_snapshot(
        req: Request<Body>,
        ctx: StatusContext,
    ) -> hyper::Result<Response<Body>> {
        let query = req.uri().query().unwrap_or("");
        let query_pairs: HashMap<_, _> = url::form_urlencoded::parse(query.as_bytes()).collect();
        let branch_ts = match get_online_snapshot_branch_ts(&query_pairs) {
            Ok(Some(branch_ts)) => branch_ts,
            Ok(None) => {
                return Ok(make_response(
                    StatusCode::BAD_REQUEST,
                    "branch_ts is required",
                ));
            }
            Err(err) => {
                return Ok(make_response(StatusCode::BAD_REQUEST, err));
            }
        };
        if ctx.concurrency_manager.get_latest_backup_ts().is_none() {
            return Ok(make_response(
                StatusCode::PRECONDITION_FAILED,
                "storage.check_backup_ts is disabled",
            ));
        }
        let wait_timeout = match get_bounded_duration_param(
            &query_pairs,
            "wait_timeout_secs",
            ONLINE_SNAPSHOT_PREPARE_WAIT_TIMEOUT,
            ONLINE_SNAPSHOT_MAX_WAIT_TIMEOUT,
        ) {
            Ok(wait_timeout) => wait_timeout,
            Err(err) => return Ok(make_response(StatusCode::BAD_REQUEST, err)),
        };
        let ttl = match get_bounded_duration_param(
            &query_pairs,
            "backup_ts_ttl_secs",
            ONLINE_SNAPSHOT_BACKUP_TS_TTL,
            ONLINE_SNAPSHOT_MAX_TTL,
        ) {
            Ok(ttl) => ttl,
            Err(err) => return Ok(make_response(StatusCode::BAD_REQUEST, err)),
        };

        if ctx.online_snapshot_prepared_ts.load(Ordering::Acquire) < branch_ts {
            if let Err(err) = ctx
                .concurrency_manager
                .update_max_ts(branch_ts.into(), "online_snapshot_prepare")
            {
                error!("online snapshot failed to update max_ts"; "branch_ts" => branch_ts, "err" => ?err);
                return Ok(make_response(
                    StatusCode::INTERNAL_SERVER_ERROR,
                    format!("failed to update max_ts: {err:?}"),
                ));
            }
            ctx.concurrency_manager.replace_backup_ts(branch_ts.into());
            let wait_start = Instant::now_coarse();
            let complete = ctx
                .concurrency_manager
                .wait_old_backup_ts_released(
                    branch_ts.into(),
                    wait_timeout,
                    ttl,
                    BACKUP_TS_WAIT_RETRY_INTERVAL,
                )
                .await;
            if !complete {
                warn!(
                    "online snapshot backup-ts barrier failed";
                    "store_id" => ctx.kvengine.get_engine_id(),
                    "branch_ts" => branch_ts,
                    "elapsed" => ?wait_start.saturating_elapsed(),
                );
                return Ok(make_response(
                    StatusCode::REQUEST_TIMEOUT,
                    "backup-ts barrier timed out or an old marker remains referenced",
                ));
            }
            ctx.online_snapshot_prepared_ts
                .fetch_max(branch_ts, Ordering::AcqRel);
        }

        Ok(Response::new(Body::empty()))
    }

    async fn check_kvengine_shard_leader(
        ctx: StatusContext,
        shard_id: u64,
        shard_ver: u64,
    ) -> std::result::Result<ShardLeaderCheck, String> {
        let (cb, fut) = paired_future_callback();
        let callback = Callback::read(Box::new(move |res| {
            cb(res);
        }));
        ctx.router.send_casual_msg(
            shard_id,
            CasualMessage::CheckLeader {
                shard_ver,
                callback,
            },
        );
        let mut res = fut.await.map_err(|err| {
            format!(
                "{}:{}:{} check leader channel error: {:?}",
                ctx.kvengine.get_engine_id(),
                shard_id,
                shard_ver,
                err
            )
        })?;
        let mut region_error = res.response.take_header().take_error();
        let mut response = DelegateResponse::default();
        if region_error.has_not_leader() {
            response
                .mut_region_error()
                .set_not_leader(region_error.take_not_leader());
        } else if region_error.has_epoch_not_match() {
            response
                .mut_region_error()
                .set_epoch_not_match(region_error.take_epoch_not_match());
        } else {
            return Ok(ShardLeaderCheck::Leader);
        }
        Ok(ShardLeaderCheck::RegionError(
            response.write_to_bytes().unwrap(),
        ))
    }

    fn validate_online_snapshot_shard(shard: &Shard) -> std::result::Result<(), String> {
        let Some(keyspace_id) = ApiV2::get_u32_keyspace_id_by_key(shard.outer_start.as_ref())
        else {
            return Err("online snapshot requires an API-v2 keyspace".to_string());
        };
        if ApiV2::is_default_keyspace(keyspace_id) {
            return Err("online snapshot does not support the default keyspace".to_string());
        }
        Ok(())
    }

    fn parse_online_snapshot_shard_request(
        req: &Request<Body>,
    ) -> std::result::Result<OnlineSnapshotShardRequest, Response<Body>> {
        let shard_id = get_last_path_segment(req.uri().path())
            .parse::<u64>()
            .map_err(|err| make_response(StatusCode::BAD_REQUEST, err.to_string()))?;
        let query = req.uri().query().unwrap_or("");
        let query_pairs: HashMap<_, _> = url::form_urlencoded::parse(query.as_bytes()).collect();
        let Some(shard_ver) = query_pairs.get("shard_ver") else {
            return Err(make_response(
                StatusCode::BAD_REQUEST,
                "shard_ver is required",
            ));
        };
        let shard_ver = shard_ver
            .parse::<u64>()
            .map_err(|err| make_response(StatusCode::BAD_REQUEST, err.to_string()))?;
        Ok(OnlineSnapshotShardRequest {
            shard_id,
            shard_ver,
            branch_ts: get_online_snapshot_branch_ts(&query_pairs)
                .map_err(|err| make_response(StatusCode::BAD_REQUEST, err))?,
        })
    }

    async fn open_online_snapshot_access(
        ctx: StatusContext,
        request: &OnlineSnapshotShardRequest,
        required_prepared_ts: Option<u64>,
    ) -> std::result::Result<OnlineSnapshotAccess, Response<Body>> {
        let dump_permit = ctx
            .online_snapshot_dump_semaphore
            .clone()
            .acquire_owned()
            .await
            .map_err(|_| {
                make_response(
                    StatusCode::SERVICE_UNAVAILABLE,
                    "online snapshot dump is shutting down",
                )
            })?;
        match Self::check_kvengine_shard_leader(ctx.clone(), request.shard_id, request.shard_ver)
            .await
        {
            Ok(ShardLeaderCheck::Leader) => {}
            Ok(ShardLeaderCheck::RegionError(body)) => {
                return Err(Response::builder()
                    .header(header::CONTENT_TYPE, CONTENT_TYPE_PROTOBUF)
                    .body(Body::from(body))
                    .unwrap());
            }
            Err(err) => {
                error!("{}", err);
                return Err(make_response(StatusCode::INTERNAL_SERVER_ERROR, err));
            }
        }

        let shard = ctx
            .kvengine
            .get_shard_with_ver(request.shard_id, request.shard_ver)
            .map_err(|err| make_response(StatusCode::NOT_FOUND, err.to_string()))?;
        Self::validate_online_snapshot_shard(&shard)
            .map_err(|err| make_response(StatusCode::PRECONDITION_FAILED, err))?;

        let store_id = ctx.kvengine.get_engine_id();
        if let Some(branch_ts) = required_prepared_ts {
            let prepared_ts = ctx.online_snapshot_prepared_ts.load(Ordering::Acquire);
            if prepared_ts < branch_ts {
                return Err(make_response(
                    StatusCode::PRECONDITION_FAILED,
                    format!(
                        "Store {} is prepared only through {}, requested {}",
                        store_id, prepared_ts, branch_ts
                    ),
                ));
            }
        }
        let snap_access = shard
            .try_new_snap_access()
            .map_err(|err| make_response(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))?;
        let term = shard
            .get_property(TERM_KEY)
            .and_then(|term| term.as_ref().try_into().ok())
            .map(u64::from_le_bytes)
            .ok_or_else(|| {
                make_response(
                    StatusCode::INTERNAL_SERVER_ERROR,
                    format!("{} has missing or malformed {}", shard.tag(), TERM_KEY),
                )
            })?;
        Ok(OnlineSnapshotAccess {
            store_id,
            snap_access,
            term,
            _dump_permit: dump_permit,
        })
    }

    // GET /kvengine/online-snapshot/capture/<shard_id>?shard_ver=xxx&branch_ts=xxx
    //
    // Captures a coherent full-CF shard snapshot after the Store has completed
    // the backup-ts preparation barrier for `branch_ts`.
    async fn capture_online_snapshot(
        req: Request<Body>,
        ctx: StatusContext,
    ) -> hyper::Result<Response<Body>> {
        let request = match Self::parse_online_snapshot_shard_request(&req) {
            Ok(request) => request,
            Err(response) => return Ok(response),
        };
        let Some(branch_ts) = request.branch_ts else {
            return Ok(make_response(
                StatusCode::BAD_REQUEST,
                "branch_ts is required and must be greater than zero",
            ));
        };
        let access =
            match Self::open_online_snapshot_access(ctx.clone(), &request, Some(branch_ts)).await {
                Ok(access) => access,
                Err(response) => return Ok(response),
            };
        let OnlineSnapshotAccess {
            store_id,
            snap_access,
            term,
            _dump_permit,
        } = access;
        let mem_table_kv_size = snap_access.full_mem_table_kv_size();
        let memory_permits = online_snapshot_memory_permits(mem_table_kv_size);
        let memory_permit = match ctx
            .online_snapshot_memory_semaphore
            .clone()
            .acquire_many_owned(memory_permits)
            .await
        {
            Ok(memory_permit) => memory_permit,
            Err(_) => {
                return Ok(make_response(
                    StatusCode::SERVICE_UNAVAILABLE,
                    "online snapshot memory admission is shutting down",
                ));
            }
        };
        let shard_id = request.shard_id;
        let shard_ver = request.shard_ver;
        let start = Instant::now_coarse();
        let build_start = Instant::now_coarse();
        let build_task = ctx.thread_pool.spawn_blocking(move || {
            let _dump_permit = _dump_permit;
            init_task_local_sync(move || {
                let mem_data = snap_access.build_full_mem_data();
                let snap_data = snap_access.marshal_online_snapshot(term);
                let mem_data_len = mem_data.len();
                let snap_data_len = snap_data.len();
                let mut response = DelegateResponse::default();
                response.set_mem_table_data(mem_data);
                response.set_snapshot(snap_data);
                response.set_mem_table_sequence(snap_access.get_write_sequence());
                (
                    response.write_to_bytes().unwrap(),
                    mem_data_len,
                    snap_data_len,
                    memory_permit,
                )
            })
        });
        let (body, mem_data_len, snap_data_len, memory_permit) = match build_task.await {
            Ok(result) => result,
            Err(err) => {
                error!(
                    "online snapshot capture build task failed";
                    "store_id" => store_id,
                    "shard_id" => shard_id,
                    "shard_ver" => shard_ver,
                    "err" => ?err,
                );
                return Ok(make_response(
                    StatusCode::INTERNAL_SERVER_ERROR,
                    format!("online snapshot capture build task failed: {err}"),
                ));
            }
        };
        if body.len() > ONLINE_SNAPSHOT_MAX_RESPONSE_BYTES {
            return Ok(make_response(
                StatusCode::PAYLOAD_TOO_LARGE,
                format!(
                    "online snapshot capture response exceeds {} bytes",
                    ONLINE_SNAPSHOT_MAX_RESPONSE_BYTES
                ),
            ));
        }
        observe_snapshot_build(build_start);
        observe_snapshot_payload_bytes(SNAPSHOT_PAYLOAD_MEMTABLE, mem_data_len);
        observe_snapshot_payload_bytes(SNAPSHOT_PAYLOAD_SNAPSHOT, snap_data_len);
        observe_snapshot_payload_bytes(SNAPSHOT_PAYLOAD_RESPONSE, body.len());
        debug!(
            "capture_online_snapshot time: {:?}, mem_data_size: {:?}, snap_data_size: {:?}, total_size: {:?}",
            start.saturating_elapsed(),
            mem_data_len,
            snap_data_len,
            body.len()
        );
        Ok(Self::make_permit_guarded_response(
            Bytes::from(body),
            memory_permit,
            CONTENT_TYPE_PROTOBUF,
            Self::DFS_FILE_READ_CHUNK_SIZE,
        ))
    }

    // GET /kvengine/online-snapshot/file-refs/<shard_id>?shard_ver=xxx
    //
    // Returns the typed immutable-file references from one coherent shard
    // snapshot. This metadata-only endpoint intentionally does not perform the
    // timestamp, lock, cache, or snapshot payload work of
    // `dump_kvengine_snapshot`.
    async fn get_online_snapshot_file_refs(
        req: Request<Body>,
        ctx: StatusContext,
    ) -> hyper::Result<Response<Body>> {
        let request = match Self::parse_online_snapshot_shard_request(&req) {
            Ok(request) => request,
            Err(response) => return Ok(response),
        };
        let access = match Self::open_online_snapshot_access(ctx.clone(), &request, None).await {
            Ok(access) => access,
            Err(response) => return Ok(response),
        };
        let OnlineSnapshotAccess {
            store_id,
            snap_access,
            term: _,
            _dump_permit,
        } = access;
        let shard_id = request.shard_id;
        let shard_ver = request.shard_ver;
        let build_task = ctx.thread_pool.spawn_blocking(move || {
            let _dump_permit = _dump_permit;
            init_task_local_sync(|| {
                let response = OnlineSnapshotFileRefsResponse {
                    version: ONLINE_SNAPSHOT_CAPABILITY_VERSION,
                    store_id,
                    shard_id,
                    shard_ver,
                    outer_start: snap_access.clone_start_key().to_vec(),
                    outer_end: snap_access.clone_end_key().to_vec(),
                    meta_sequence: snap_access.get_meta_seq(),
                    write_sequence: snap_access.get_write_sequence(),
                    files: snap_access
                        .get_all_files_with_type()
                        .into_iter()
                        .map(|(file_type, file_id)| OnlineSnapshotFileRef {
                            file_type: file_type as u8,
                            file_id,
                        })
                        .collect(),
                };
                serde_json::to_vec(&response).unwrap()
            })
        });
        let body = match build_task.await {
            Ok(body) => body,
            Err(err) => {
                error!(
                    "online snapshot file-refs build task failed";
                    "store_id" => store_id,
                    "shard_id" => shard_id,
                    "shard_ver" => shard_ver,
                    "err" => ?err,
                );
                return Ok(make_response(
                    StatusCode::INTERNAL_SERVER_ERROR,
                    format!("online snapshot file-refs build task failed: {err}"),
                ));
            }
        };
        if body.len() > ONLINE_SNAPSHOT_MAX_RESPONSE_BYTES {
            return Ok(make_response(
                StatusCode::PAYLOAD_TOO_LARGE,
                format!(
                    "online snapshot file-refs response exceeds {} bytes",
                    ONLINE_SNAPSHOT_MAX_RESPONSE_BYTES
                ),
            ));
        }
        Ok(Response::builder()
            .header(header::CONTENT_TYPE, "application/json")
            .body(Body::from(body))
            .unwrap())
    }

    // URI: /kvengine/snapshot/<shard_id>?start_ts=xxx&shard_ver=xxx[&start_key=xxx&
    // end_key=xxx][&meta_seq=xxx&write_seq=xxx]
    //
    // dump kvengine shard snapshot with start_ts
    // `start_key` and `end_key` are in the upper hex string format, e.g.
    // `7800000174800000000000007F5F`
    // For requests with real `start_ts` (`start_ts != u64::MAX`), the response
    // always contains all MVCC versions in memtable data and carries a capability
    // header so the client can safely reuse it across `start_ts`. In this mode,
    // the server still checks locks with the real `start_ts` before returning a
    // cache hit. Legacy requests with `start_ts == u64::MAX` keep the old behavior
    // for rolling upgrade compatibility.
    async fn dump_kvengine_snapshot(
        req: Request<Body>,
        ctx: StatusContext,
    ) -> hyper::Result<Response<Body>> {
        let path = req.uri().path();
        let last = get_last_path_segment(path);
        let query = req.uri().query().unwrap_or("");
        let query_pairs: HashMap<_, _> = url::form_urlencoded::parse(query.as_bytes()).collect();
        if !query_pairs.contains_key("start_ts") {
            return Ok(make_response(
                StatusCode::BAD_REQUEST,
                "start_ts is required".to_string(),
            ));
        }
        if !query_pairs.contains_key("shard_ver") {
            return Ok(make_response(
                StatusCode::BAD_REQUEST,
                "shard_ver is required".to_string(),
            ));
        }
        let shard_ver = match u64::from_str(query_pairs.get("shard_ver").unwrap()) {
            Ok(ver) => ver,
            Err(err) => return Ok(make_response(StatusCode::BAD_REQUEST, err.to_string())),
        };
        let start_ts = match u64::from_str(query_pairs.get("start_ts").unwrap()) {
            Ok(ts) => ts,
            Err(err) => return Ok(make_response(StatusCode::BAD_REQUEST, err.to_string())),
        };
        // TODO: this can be removed after columnar snapshot upgrade is done.
        let for_new_snapshot_cache = start_ts != u64::MAX;
        let shard_id = match u64::from_str(last) {
            Ok(id) => id,
            Err(err) => return Ok(make_response(StatusCode::BAD_REQUEST, err.to_string())),
        };

        // Parse optional meta_seq and write_seq parameters for cache validation
        let client_cache_validation = query_pairs
            .get("meta_seq")
            .and_then(|s| u64::from_str(s).ok())
            .zip(
                query_pairs
                    .get("write_seq")
                    .and_then(|s| u64::from_str(s).ok()),
            );

        let (cb, fut) = paired_future_callback();
        let callback = Callback::read(Box::new(move |res| {
            cb(res);
        }));
        ctx.router.send_casual_msg(
            shard_id,
            CasualMessage::CheckLeader {
                shard_ver,
                callback,
            },
        );
        let mut res = match fut.await {
            Ok(res) => res,
            Err(e) => {
                let err_msg = format!(
                    "{}:{}:{} check leader channel error: {:?}",
                    ctx.kvengine.get_engine_id(),
                    shard_id,
                    shard_ver,
                    e
                );
                error!("{}", err_msg);
                return Ok(make_response(StatusCode::INTERNAL_SERVER_ERROR, err_msg));
            }
        };
        let make_snapshot_response =
            |builder: hyper::http::response::Builder, body: Body| -> Response<Body> {
                let builder = if for_new_snapshot_cache {
                    builder.header(SNAPSHOT_CACHE_CAPABILITY_HEADER, "1")
                } else {
                    builder
                };
                builder.body(body).unwrap()
            };
        let make_ok_response = |body: Vec<u8>| -> Response<Body> {
            make_snapshot_response(Response::builder(), Body::from(body))
        };
        let mut delegate_resp = DelegateResponse::default();
        let mut err = res.response.take_header().take_error();
        if err.has_not_leader() {
            delegate_resp
                .mut_region_error()
                .set_not_leader(err.take_not_leader());
            let body = delegate_resp.write_to_bytes().unwrap();
            return Ok(make_ok_response(body));
        } else if err.has_epoch_not_match() {
            delegate_resp
                .mut_region_error()
                .set_epoch_not_match(err.take_epoch_not_match());
            let body = delegate_resp.write_to_bytes().unwrap();
            return Ok(make_ok_response(body));
        }
        Ok(match ctx.kvengine.get_shard_with_ver(shard_id, shard_ver) {
            Ok(shard) => {
                if !for_new_snapshot_cache {
                    if let Some((client_meta_seq, client_write_seq)) = client_cache_validation {
                        let server_meta_seq = shard.get_meta_sequence();
                        let server_write_seq = shard.get_write_sequence();

                        if client_meta_seq == server_meta_seq
                            && client_write_seq == server_write_seq
                        {
                            debug!(
                                "legacy cache hit for shard {} with meta_seq={}, write_seq={}",
                                shard_id, server_meta_seq, server_write_seq
                            );
                            observe_snapshot_cache_validation(SNAPSHOT_CACHE_VALIDATION_HIT);
                            return Ok(Response::builder()
                                .status(StatusCode::NOT_MODIFIED)
                                .body(Body::empty())
                                .unwrap());
                        }
                    }
                }
                let start = Instant::now_coarse();
                let snap_access = match shard.try_new_snap_access() {
                    Ok(snap_access) => snap_access,
                    Err(err) => {
                        return Ok(make_response(
                            StatusCode::INTERNAL_SERVER_ERROR,
                            err.to_string(),
                        ));
                    }
                };
                let outer_start = if query_pairs.contains_key("start_key") {
                    match hex::decode(query_pairs.get("start_key").unwrap().to_string()) {
                        Ok(v) => std::cmp::max(v.into(), shard.outer_start.clone()),
                        Err(_) => {
                            return Ok(make_response(
                                StatusCode::BAD_REQUEST,
                                "start_key is not a valid hex string",
                            ));
                        }
                    }
                } else {
                    shard.outer_start.clone()
                };
                let outer_end = if query_pairs.contains_key("end_key") {
                    match hex::decode(query_pairs.get("end_key").unwrap().to_string()) {
                        Ok(v) => std::cmp::min(v.into(), shard.outer_end.clone()),
                        Err(_) => {
                            return Ok(make_response(
                                StatusCode::BAD_REQUEST,
                                "end_key is not a valid hex string",
                            ));
                        }
                    }
                } else {
                    shard.outer_end.clone()
                };
                let region_snapshot =
                    RegionSnapshot::from_snapshot(snap_access.clone(), None, None);
                let cloud_store =
                    CloudStore::new(region_snapshot, start_ts, TsSet::default(), true, false);
                // Check if the shard contains locks belongs to the ranges. If there is any
                // lock, return the first key as LockInfo. The client should retry in this case.
                let lock_check_start = Instant::now_coarse();
                let lock_check_res = cloud_store
                    .check_locks_in_range(&outer_start, &outer_end, false)
                    .map_err(tikv::coprocessor::Error::from);
                let lock_check_result = match &lock_check_res {
                    Ok(_) => SNAPSHOT_LOCK_CHECK_OK,
                    Err(tikv::coprocessor::Error::Locked(_)) => SNAPSHOT_LOCK_CHECK_LOCKED,
                    Err(_) => SNAPSHOT_LOCK_CHECK_ERROR,
                };
                observe_snapshot_lock_check(lock_check_result, lock_check_start);
                if let Err(tikv::coprocessor::Error::Locked(lock_info)) = lock_check_res {
                    observe_snapshot_cache_validation(SNAPSHOT_CACHE_VALIDATION_LOCKED);
                    delegate_resp.set_locked(lock_info);
                    let body = delegate_resp.write_to_bytes().unwrap();
                    return Ok(make_ok_response(body));
                }
                if for_new_snapshot_cache {
                    if let Some((client_meta_seq, client_write_seq)) = client_cache_validation {
                        let server_meta_seq = snap_access.get_meta_seq();
                        let server_write_seq = snap_access.get_write_sequence();

                        if client_meta_seq == server_meta_seq
                            && client_write_seq == server_write_seq
                        {
                            debug!(
                                "cache hit for shard {} with meta_seq={}, write_seq={}",
                                shard_id, server_meta_seq, server_write_seq
                            );
                            observe_snapshot_cache_validation(SNAPSHOT_CACHE_VALIDATION_HIT);
                            return Ok(make_snapshot_response(
                                Response::builder().status(StatusCode::NOT_MODIFIED),
                                Body::empty(),
                            ));
                        }
                    }
                }
                observe_snapshot_cache_validation(SNAPSHOT_CACHE_VALIDATION_MISS);
                let outer_ranges = vec![(outer_start, outer_end)];
                // Always build memtable data for all versions.
                let build_start = Instant::now_coarse();
                let (mem_data, _) = snap_access.build_mem_data(&outer_ranges, u64::MAX);
                let (_, snap_data) = snap_access.marshal(&outer_ranges, false, false);
                let mem_data_len = mem_data.len();
                let snap_data_len = snap_data.len();
                delegate_resp.set_mem_table_data(mem_data);
                delegate_resp.set_snapshot(snap_data);
                delegate_resp.set_mem_table_sequence(snap_access.get_write_sequence());
                let body = delegate_resp.write_to_bytes().unwrap();
                observe_snapshot_build(build_start);
                observe_snapshot_payload_bytes(SNAPSHOT_PAYLOAD_MEMTABLE, mem_data_len);
                observe_snapshot_payload_bytes(SNAPSHOT_PAYLOAD_SNAPSHOT, snap_data_len);
                observe_snapshot_payload_bytes(SNAPSHOT_PAYLOAD_RESPONSE, body.len());
                debug!(
                    "dump_kvengine_snapshot time: {:?}, mem_data_size: {:?}, snap_data_size: {:?}, total_size: {:?}",
                    start.saturating_elapsed(),
                    mem_data_len,
                    snap_data_len,
                    body.len()
                );
                make_ok_response(body)
            }
            Err(e) => make_response(StatusCode::NOT_FOUND, e.to_string()),
        })
    }

    async fn dump_kvengine_meta(
        req: Request<Body>,
        engine: &kvengine::Engine,
    ) -> hyper::Result<Response<Body>> {
        let path = req.uri().path();
        let last = get_last_path_segment(path);
        let meta_bin = match u64::from_str(last) {
            Ok(id) => match engine.get_shard(id) {
                Some(shard) => {
                    let snap_access = match shard.try_new_snap_access() {
                        Ok(snap_access) => snap_access,
                        Err(err) => {
                            return Ok(make_response(
                                StatusCode::INTERNAL_SERVER_ERROR,
                                err.to_string(),
                            ));
                        }
                    };
                    let (_, meta_bin) = snap_access.marshal(
                        &[(shard.outer_start.clone(), shard.outer_end.clone())],
                        false,
                        false,
                    );
                    meta_bin
                }
                None => Vec::new(),
            },
            Err(_) => Vec::new(),
        };
        Ok(Response::builder().body(Body::from(meta_bin)).unwrap())
    }

    async fn dump_kvengine_stats(
        req: Request<Body>,
        engine: &kvengine::Engine,
    ) -> hyper::Result<Response<Body>> {
        let path = req.uri().path();
        let last = get_last_path_segment(path);
        let res;
        if let Ok(id) = u64::from_str(last) {
            // get all shard state in given keyspace id.
            if path.starts_with("/kvengine/keyspace") {
                let range = ApiV2::get_txn_keyspace_range(id as u32);
                match Self::get_covered_shards_by_range(Some(range), engine) {
                    Ok(shards) => {
                        let states: Vec<ShardStats> =
                            shards.iter().map(|s| s.get_stats()).collect();
                        res = serde_json::to_string_pretty(&states);
                    }
                    Err(e) => {
                        return Ok(make_response(
                            StatusCode::INTERNAL_SERVER_ERROR,
                            e.to_string(),
                        ));
                    }
                }
            } else {
                let shard_stats = engine.get_shard_stat(id);
                res = serde_json::to_string_pretty(&shard_stats);
            }
        } else if path.starts_with("/kvengine/all") {
            let all_shard_stats = engine.get_all_shard_stats();
            res = serde_json::to_string_pretty(&all_shard_stats);
        } else if path.starts_with("/kvengine/active_lite") {
            // get all active shard stats lite.
            let all_active_lite = engine.get_all_active_shard_stats_lite();
            res = serde_json::to_string(&all_active_lite);
        } else if path.starts_with("/kvengine/files_with_type") {
            let files_with_type = Self::dump_kvengine_files_with_type(req, engine);
            res = serde_json::to_string(&files_with_type);
        } else if path.starts_with("/kvengine/files") {
            // NOTE: Must be the last for /kvengine/files... paths.
            let mut all_shard_files: Vec<(u64, Vec<u64>)> = engine
                .get_all_shard_id_vers()
                .iter()
                .filter_map(|id_ver| {
                    engine
                        .get_shard(id_ver.id)
                        .map(|shard| (shard.id, shard.get_all_files()))
                })
                .collect();
            // add blacklist files to the collection of files in use,
            // for simplicity, zero shard id represents the blacklist files
            let blacklist_file_ids = engine
                .get_files_in_blacklist()
                .iter()
                .copied()
                .collect::<Vec<_>>();
            all_shard_files.push((0, blacklist_file_ids));
            res = serde_json::to_string(&all_shard_files);
        } else if path.starts_with("/kvengine/compactor") {
            let remote_urls = engine.comp_client.get_remote_compactors();
            res = serde_json::to_string_pretty(&remote_urls);
        } else {
            let all_shard_stats = engine.get_all_shard_stats();
            let engine_stats = engine.get_engine_stats(all_shard_stats);
            res = serde_json::to_string_pretty(&engine_stats);
        }
        Ok(match res {
            Ok(json) => Response::builder()
                .header(header::CONTENT_TYPE, "application/json")
                .body(Body::from(json))
                .unwrap(),
            Err(_) => make_response(StatusCode::INTERNAL_SERVER_ERROR, "Internal Server Error"),
        })
    }

    /// Request: /kvengine/files_with_type?filter=sst,sst_ia,txn,col
    /// Response: { "sst": [file_id1, file_id2], "blob": [file_id3], ... }
    /// Note: currently support "sst" and "sst_ia" only.
    fn dump_kvengine_files_with_type(
        req: Request<Body>,
        engine: &kvengine::Engine,
    ) -> HashMap<String /* file_suffix */, HashSet<u64 /* file_id */>> {
        let query = req.uri().query().unwrap_or("");
        let query_pairs: HashMap<_, _> = url::form_urlencoded::parse(query.as_bytes()).collect();
        let Some(filter_str) = query_pairs.get("filter").map(|s| s.as_ref()) else {
            return HashMap::default();
        };
        let filter_types: HashSet<&str> = filter_str.split(',').collect();
        let mut files_with_type: HashMap<String, HashSet<u64>> =
            HashMap::with_capacity(filter_types.len());
        let need_sst = filter_types.contains("sst");
        let need_sst_ia = filter_types.contains("sst_ia");
        if !need_sst && !need_sst_ia {
            return files_with_type;
        }

        engine.get_all_shard_id_vers().iter().for_each(|id_ver| {
            let Some(shard) = engine.get_shard(id_ver.id) else {
                return;
            };
            let (local_files, ia_files) = shard.get_local_sst_files();
            if need_sst {
                files_with_type
                    .entry("sst".to_string())
                    .or_default()
                    .extend(local_files);
            }
            if need_sst_ia {
                files_with_type
                    .entry("sst_ia".to_string())
                    .or_default()
                    .extend(ia_files);
            }
        });
        files_with_type
    }

    async fn add_remote_compactor(
        req: Request<Body>,
        comp_client: &kvengine::CompactionClient,
    ) -> hyper::Result<Response<Body>> {
        let mut body = Vec::new();
        req.into_body()
            .try_for_each(|bytes| {
                body.extend(bytes);
                ok(())
            })
            .await?;
        Ok(match String::from_utf8(body) {
            Ok(remote_url) => {
                comp_client.add_remote_compactor(remote_url);
                let mut resp = Response::default();
                *resp.status_mut() = StatusCode::OK;
                resp
            }
            Err(e) => make_response(
                StatusCode::INTERNAL_SERVER_ERROR,
                format!("failed to decode, error: {:?}", e),
            ),
        })
    }

    async fn get_change_set_request(
        raw_req: Request<Body>,
    ) -> hyper::Result<kvenginepb::ChangeSet> {
        let mut body = Vec::new();
        raw_req
            .into_body()
            .try_for_each(|bytes| {
                body.extend(bytes);
                ok(())
            })
            .await?;
        let mut cs = kvenginepb::ChangeSet::default();
        cs.merge_from_bytes(&body).unwrap();
        Ok(cs)
    }

    async fn handle_ingest_request(
        cs: kvenginepb::ChangeSet,
        ctx: StatusContext,
    ) -> hyper::Result<Response<Body>> {
        let tag = tag_from_cs(&ctx.kvengine, &cs);
        info!("[{}] receive ingest_files request: {:?}", tag, cs);

        if let Err(errpb) = check_available_space(&ctx.kvengine) {
            warn!("[{}] reject ingest files, low space", tag);
            return Ok(make_response(
                StatusCode::INTERNAL_SERVER_ERROR,
                errpb.write_to_bytes().unwrap(),
            ));
        }
        if let Err(errpb) =
            check_follower_available_space(ctx.clone(), cs.shard_id, cs.shard_ver).await
        {
            warn!("[{}] reject ingest files, follower low space", tag; "err" => ?errpb);
            return Ok(make_response(
                StatusCode::INTERNAL_SERVER_ERROR,
                errpb.write_to_bytes().unwrap(),
            ));
        }

        let (cb, fut) = paired_future_callback();
        let callback = Callback::write(Box::new(move |res| {
            cb(res);
        }));
        ctx.router.send_casual_msg(
            cs.get_shard_id(),
            CasualMessage::IngestFiles { cs, callback },
        );

        let res = match fut.await {
            Ok(res) => res,
            Err(e) => {
                let err_msg = format!("{} ingest_files channel error: {:?}", tag, e);
                error!("{}", err_msg);
                // Return "not leader" to let callers retry.
                let mut errpb = kvproto::errorpb::Error::default();
                errpb.set_not_leader(Default::default());
                errpb.set_message(err_msg);
                return Ok(make_response(
                    StatusCode::INTERNAL_SERVER_ERROR,
                    errpb.write_to_bytes().unwrap(),
                ));
            }
        };
        if res.response.get_header().has_error() {
            error!(
                "{} ingest_files error: {:?}",
                tag,
                res.response.get_header().get_error()
            );
            let err_data = res
                .response
                .get_header()
                .get_error()
                .write_to_bytes()
                .unwrap();
            Ok(make_response(StatusCode::INTERNAL_SERVER_ERROR, err_data))
        } else {
            Ok(make_response(StatusCode::OK, ""))
        }
    }

    async fn ingest_files(req: Request<Body>, ctx: StatusContext) -> hyper::Result<Response<Body>> {
        let cs = Self::get_change_set_request(req).await?;
        Self::handle_ingest_request(cs, ctx).await
    }

    async fn ingest_s3(req: Request<Body>, ctx: StatusContext) -> hyper::Result<Response<Body>> {
        let query = req.uri().query().unwrap_or("");
        let params: HashMap<_, _> = url::form_urlencoded::parse(query.as_bytes())
            .into_owned()
            .collect();
        let cluster_id = params.get("cluster_id").and_then(|v| v.parse::<u64>().ok());

        let region_id = params.get("region_id").and_then(|v| v.parse::<u64>().ok());
        let epoch_version = params
            .get("epoch_version")
            .and_then(|v| v.parse::<u64>().ok());

        if cluster_id.is_none() || region_id.is_none() || epoch_version.is_none() {
            let mut err = kvproto::errorpb::Error::default();
            err.set_message("missing required parameter".to_string());
            return Ok(Response::builder()
                .status(StatusCode::BAD_REQUEST)
                .body(Body::from(err.write_to_bytes().unwrap()))
                .unwrap());
        }

        // TODO: check cluster id
        // if cluster_id.unwrap() != store_ident.cluster_id {
        //     let mut err = kvproto::errorpb::Error::default();
        //     err.set_message(format!(
        //         "cluster id mismatch, expect {}, got {}",
        //         self.store_ident.cluster_id,
        //         cluster_id.unwrap()
        //     ));
        //     let mut buf = Vec::new();
        //     err.encode(&mut buf).unwrap();
        //     return Ok(Response::builder()
        //         .status(StatusCode::BAD_REQUEST)
        //         .body(Body::from(buf))
        //         .unwrap());
        // }

        let body = match hyper::body::to_bytes(req.into_body()).await {
            Ok(body) => body,
            Err(e) => {
                let mut err = kvproto::errorpb::Error::default();
                err.set_message(format!("failed to read request body: {}", e));
                return Ok(Response::builder()
                    .status(StatusCode::INTERNAL_SERVER_ERROR)
                    .body(Body::from(err.write_to_bytes().unwrap()))
                    .unwrap());
            }
        };

        // Assuming SstMeta is a struct that can be deserialized from JSON.
        let sst_meta: SstMeta = match serde_json::from_slice(&body) {
            Ok(meta) => meta,
            Err(e) => {
                let mut err = kvproto::errorpb::Error::default();
                err.set_message(format!("failed to deserialize SstMeta from body: {}", e));
                return Ok(Response::builder()
                    .status(StatusCode::INTERNAL_SERVER_ERROR)
                    .body(Body::from(err.write_to_bytes().unwrap()))
                    .unwrap());
            }
        };
        // Check s3 file exist
        let file_exists = match ctx
            .kvengine
            .core
            .fs
            .exists(sst_meta.id, kvengine::dfs::Options::default())
            .await
        {
            Ok(exists) => exists,
            Err(e) => {
                let mut err = kvproto::errorpb::Error::default();
                err.set_message(format!("failed to check s3 file existence: {}", e));
                return Ok(Response::builder()
                    .status(StatusCode::INTERNAL_SERVER_ERROR)
                    .body(Body::from(err.write_to_bytes().unwrap()))
                    .unwrap());
            }
        };
        if !file_exists {
            let mut err = kvproto::errorpb::Error::default();
            err.set_message(format!("file id {} not found in s3", sst_meta.id));
            return Ok(Response::builder()
                .status(StatusCode::NOT_FOUND)
                .body(Body::from(err.write_to_bytes().unwrap()))
                .unwrap());
        }

        // Construct ChangeSet
        let mut cs = kvenginepb::ChangeSet::default();
        cs.set_shard_id(region_id.unwrap());
        cs.set_shard_ver(epoch_version.unwrap());
        let ingest_files = cs.mut_ingest_files();
        ingest_files.set_max_ts(sst_meta.commit_ts);

        // Don't set INGEST_ID_KEY property to indicate that it's from load data.
        let table_create = new_table_create_pb(
            sst_meta.id,
            WRITE_CF_BOTTOM_LEVEL, // TODO: get ingest level instead of
            WRITE_CF as i32,
            sst_meta.smallest.clone(),
            sst_meta.biggest.clone(),
            sst_meta.meta_offset,
        );
        ingest_files.mut_table_creates().push(table_create);

        Self::handle_ingest_request(cs, ctx).await
    }

    async fn dump_rfengine_stats(
        req: Request<Body>,
        engine: &RfEngine,
    ) -> hyper::Result<Response<Body>> {
        let path = req.uri().path();
        let last = get_last_path_segment(path);
        let res = if let Ok(peer_id) = u64::from_str(last) {
            let peer_stats = engine.get_peer_stats(peer_id);
            serde_json::to_string_pretty(&peer_stats)
        } else {
            let engine_stats = engine.get_engine_stats();
            serde_json::to_string_pretty(&engine_stats)
        };
        Ok(match res {
            Ok(json) => Response::builder()
                .header(header::CONTENT_TYPE, "application/json")
                .body(Body::from(json))
                .unwrap(),
            Err(_) => make_response(StatusCode::INTERNAL_SERVER_ERROR, "Internal Server Error"),
        })
    }

    // URI: /rfengine/wal_chunk?epoch_id=xxx&start_off=xxx&end_off=xxx
    // If end_off is zero, it means to dump to the latest, then the response status
    // code maybe PARTIAL_CONTENT if the request epoch_id is not the latest.
    async fn rfengine_wal_chunk(
        req: Request<Body>,
        engine: &RfEngine,
    ) -> hyper::Result<Response<Body>> {
        let bad_request_resp = |msg: &str| make_response(StatusCode::BAD_REQUEST, msg.to_owned());

        let query = req.uri().query().unwrap_or("");
        let query_pairs: HashMap<_, _> = url::form_urlencoded::parse(query.as_bytes()).collect();
        info!("/rfengine/wal_chunk {:?}", query_pairs);
        if !query_pairs.contains_key("epoch_id")
            || !query_pairs.contains_key("start_off")
            || !query_pairs.contains_key("end_off")
        {
            return Ok(bad_request_resp("query parameters invalid"));
        }
        let (callback, future) = paired_future_callback();
        let epoch_id = match u32::from_str(query_pairs.get("epoch_id").unwrap()) {
            Ok(epoch_id) => epoch_id,
            Err(err) => return Ok(bad_request_resp(err.to_string().as_str())),
        };
        let start_off = match u64::from_str(query_pairs.get("start_off").unwrap()) {
            Ok(start_off) => start_off,
            Err(err) => return Ok(bad_request_resp(err.to_string().as_str())),
        };
        let end_off = match u64::from_str(query_pairs.get("end_off").unwrap()) {
            Ok(end_off) => end_off,
            Err(err) => return Ok(bad_request_resp(err.to_string().as_str())),
        };
        // Async WAL remains available when the DFS worker is disabled. Regular
        // backups may legitimately need this local tail when it has not reached
        // object storage, so WAL reads must not depend on the worker's runtime
        // configuration. `dump_wal_chunk` reports unavailable or overwritten WAL.
        engine.dump_wal_chunk(epoch_id, start_off, end_off, callback);

        Ok(match future.await {
            Ok(resp) => match resp {
                Ok((chunk, partial_content)) => {
                    let status = if partial_content {
                        StatusCode::PARTIAL_CONTENT
                    } else {
                        StatusCode::OK
                    };
                    Response::builder()
                        .status(status)
                        .body(Body::from(chunk))
                        .unwrap()
                }
                Err(Error::WalEpochOverwritten { epoch_id }) => make_response(
                    StatusCode::GONE,
                    format!("WAL epoch {epoch_id} is overwritten"),
                ),
                Err(err) => {
                    error!("rfengine_wal_chunk error: {}", err);
                    make_response(StatusCode::BAD_REQUEST, format!("bad request {}", err))
                }
            },
            Err(e) => make_response(
                StatusCode::INTERNAL_SERVER_ERROR,
                format!("Internal Server Error {}", e),
            ),
        })
    }

    // URI: /unsafe_recover/clear?cluster_id=xxx&[keyspace_id=xxx[&table_id=xxx]][&
    // region_id=xxx]
    // The request body can be empty or a DelegateResponse protobuf message.
    // If the request body is provided, it only support single region clear, and the
    // ChangeSet in the body will be used instead of empty ChangeSet.
    async fn unsafe_recover(
        req: Request<Body>,
        rf: &RfEngine,
        engine: &kvengine::Engine,
    ) -> hyper::Result<Response<Body>> {
        let bad_request_resp =
            |msg: &str| make_response(StatusCode::BAD_REQUEST, msg.to_owned() + "\n");
        let path = req.uri().path();
        if path != "/unsafe_recover/clear" {
            return Ok(bad_request_resp("bad request URI"));
        }
        let query = req.uri().query().unwrap_or("");
        let query_pairs: HashMap<_, _> = url::form_urlencoded::parse(query.as_bytes()).collect();
        let cluster_id = query_pairs.get("cluster_id");
        if cluster_id.is_none() {
            return Ok(bad_request_resp("cluster_id not found"));
        }
        let cluster_id = match u64::from_str(cluster_id.unwrap()) {
            Ok(cluster_id) => cluster_id,
            Err(e) => return Ok(bad_request_resp(e.to_string().as_str())),
        };
        let store_ident = load_store_ident(rf).unwrap_or_default();
        if cluster_id != store_ident.cluster_id {
            return Ok(bad_request_resp("cluster_id not match"));
        }
        let region_id = query_pairs.get("region_id");
        let keyspace_id = query_pairs.get("keyspace_id");
        let table_id = query_pairs.get("table_id");

        let target_regions = if let Some(region_id) = region_id {
            let region_id = match u64::from_str(region_id) {
                Ok(region_id) => region_id,
                Err(err) => return Ok(bad_request_resp(err.to_string().as_str())),
            };
            match rf.get_region_peer_map().get(&region_id) {
                Some(&peer_id) => {
                    let cs = load_raft_engine_meta(rf, peer_id);
                    if cs.is_none() {
                        return Ok(bad_request_resp(
                            format!("region {} peer {} not exists", region_id, peer_id).as_str(),
                        ));
                    }
                    let cs = cs.unwrap();
                    let keyspace_id =
                        ApiV2::get_u32_keyspace_id_by_key(&cs.get_snapshot().outer_start)
                            .unwrap_or_default();
                    vec![(peer_id, region_id, cs.shard_ver, keyspace_id)]
                }
                None => {
                    return Ok(bad_request_resp(
                        format!("region {} not exists", region_id).as_str(),
                    ));
                }
            }
        } else if let Some(keyspace_id) = keyspace_id {
            let keyspace_id = match u32::from_str(keyspace_id) {
                Ok(keyspace_id) => keyspace_id,
                Err(e) => return Ok(bad_request_resp(e.to_string().as_str())),
            };
            let mut prefix = ApiV2::get_txn_keyspace_prefix(keyspace_id);
            if let Some(table_id) = table_id {
                let table_id = match u64::from_str(table_id) {
                    Ok(table_id) => table_id,
                    Err(e) => return Ok(bad_request_resp(e.to_string().as_str())),
                };
                prefix.put_u8(b't');
                prefix.encode_i64(table_id as i64).unwrap();
            }
            if let Some(regions) = collect_prefix_regions(rf, &prefix) {
                regions
            } else {
                return Ok(bad_request_resp("collect none region with prefix"));
            }
        } else {
            return Ok(bad_request_resp("query parameters invalid"));
        };

        let target_regions_len = target_regions.len();
        let mut wb = WriteBatch::new();
        let body = hyper::body::to_bytes(req.into_body()).await?;
        let cs_snap = if !body.is_empty() {
            if target_regions_len != 1 {
                return Ok(bad_request_resp(
                    "body not empty only support single region clear",
                ));
            }
            let (_, target_shard_id, target_shard_ver, _) = target_regions[0];
            let mut delegate_resp = DelegateResponse::default();
            delegate_resp.merge_from_bytes(body.chunk()).unwrap();
            let mut change_set = kvenginepb::ChangeSet::default();
            change_set
                .merge_from_bytes(delegate_resp.get_snapshot())
                .unwrap();
            if change_set.get_shard_id() != target_shard_id {
                return Ok(bad_request_resp(&format!(
                    "body shard_id {} not match target {}",
                    change_set.get_shard_id(),
                    target_shard_id
                )));
            }
            if change_set.get_shard_ver() != target_shard_ver {
                return Ok(bad_request_resp(&format!(
                    "body shard_ver {} not match target {}",
                    change_set.get_shard_ver(),
                    target_shard_ver
                )));
            }
            Some(change_set)
        } else {
            None
        };
        for (peer_id, region_id, region_version, keyspace_id) in target_regions {
            let shard_stats = engine.get_shard_stat(region_id);
            info!(
                "unsafe recover region {} peer {} shard_stats {:?}",
                region_id, peer_id, shard_stats
            );
            if shard_stats.id != 0 {
                return Ok(bad_request_resp(
                    format!("region {} not in blacklist", region_id).as_str(),
                ));
            }

            let old_raft_state = match Self::load_raft_state(rf, peer_id, region_version) {
                Some(state) => state,
                None => {
                    return Ok(bad_request_resp(
                        format!(
                            "region {} peer {} raft state not exists",
                            region_id, peer_id
                        )
                        .as_str(),
                    ));
                }
            };
            let raft_state_last_index = old_raft_state.get_last_index();
            let raft_log_last_index = rf.get_last_index(peer_id).unwrap_or(RAFT_INIT_LOG_INDEX);
            let origin_last_index = std::cmp::max(raft_state_last_index, raft_log_last_index);
            // Add a delta to last_index to replicate empty snapshot to followers.
            let last_index = origin_last_index + UNSAFE_RECOVER_DELTA;
            info!(
                "unsafe_recover region: {} peer: {} truncate raft log to {}",
                region_id, peer_id, last_index
            );
            wb.truncate_raft_log(peer_id, region_id, keyspace_id, last_index);

            let raft_log_term = rf
                .get_term(peer_id, origin_last_index)
                .unwrap_or(RAFT_INIT_LOG_TERM);
            let Some(old_cs) = load_raft_engine_meta(rf, peer_id) else {
                return Ok(bad_request_resp(
                    format!(
                        "region {} peer {} raft engine meta not exists",
                        region_id, peer_id
                    )
                    .as_str(),
                ));
            };
            let snap = old_cs.get_snapshot();
            let inner_key_off = snap.get_inner_key_off();
            let mut properties = kvengine::Properties::new();
            properties = properties.apply_pb(snap.get_properties());
            let meta_term = properties.get(TERM_KEY).unwrap().get_u64_le();
            let encryption_key = properties.get(ENCRYPTION_KEY);
            let encryption_meta = properties.get(ENCRYPTION_META_KEY);

            // Add a delta to term make sure it's greater than term in pd cache.
            let term = std::cmp::max(raft_log_term, meta_term) + UNSAFE_RECOVER_DELTA;

            let _ = match Self::write_empty_engine_meta(
                rf,
                &mut wb,
                peer_id,
                region_id,
                region_version,
                keyspace_id,
                last_index,
                term,
                inner_key_off,
                cs_snap.clone(),
                encryption_key,
                encryption_meta,
            ) {
                Ok(cs) => {
                    info!(
                        "unsafe_recover region: {} peer: {} reset kv engine meta {:?} -> {:?}",
                        region_id, peer_id, old_cs, cs
                    );
                    cs
                }
                Err(e) => return Ok(bad_request_resp(e.to_string().as_str())),
            };

            Self::write_truncated_state(&mut wb, peer_id, region_id, keyspace_id, last_index, term);
            info!(
                "unsafe_recover region: {} peer: {} reset truncated state to truncated_index: {} truncated_term: {}",
                region_id, peer_id, last_index, term
            );

            let raft_state = Self::write_raft_state(
                &mut wb,
                peer_id,
                region_id,
                region_version,
                keyspace_id,
                last_index,
                term,
            );
            info!(
                "unsafe_recover region: {} peer: {} reset raft state {:?} -> {:?}",
                region_id, peer_id, old_raft_state, raft_state
            );
        }

        // Write batch to raft engine only when it's not empty. Or it will cause the wal
        // iteration interrupted.
        if !wb.is_empty() {
            rf.write(wb).unwrap();
        }

        Ok(make_response(
            StatusCode::OK,
            format!("{} region(s) clear success\n", target_regions_len),
        ))
    }

    // URI: /major_compact?major_compact=xxx[&keyspace_id=xxx[&table_id=xxx]][&
    // region_id=xxx][&columnar=true]
    // Note: return 404 when no match region is found.
    async fn major_compact(
        req: Request<Body>,
        ctx: StatusContext,
    ) -> hyper::Result<Response<Body>> {
        info!("major compact request: {:?}", req);
        let bad_request_resp = |msg: &str| make_response(StatusCode::BAD_REQUEST, msg.to_owned());
        let not_found_resp = |msg: &str| make_response(StatusCode::NOT_FOUND, msg.to_owned());

        let path = req.uri().path();
        if path != "/major-compact" {
            return Ok(bad_request_resp("bad request URI"));
        }
        let query = req.uri().query().unwrap_or("");
        let query_pairs: HashMap<_, _> = url::form_urlencoded::parse(query.as_bytes()).collect();
        let major_compact = query_pairs.get("major_compact");
        if major_compact.is_none() {
            return Ok(bad_request_resp("major_compact not found"));
        }
        let major_compact = match bool::from_str(major_compact.unwrap()) {
            Ok(major_compact) => major_compact,
            Err(e) => return Ok(bad_request_resp(e.to_string().as_str())),
        };
        let region_id = query_pairs.get("region_id");
        let keyspace_id = query_pairs.get("keyspace_id");
        let table_id = query_pairs.get("table_id");
        let columnar = query_pairs
            .get("columnar")
            .is_some_and(|s| bool::from_str(s).unwrap_or(false));

        let target_regions = if let Some(region_id) = region_id {
            let region_id = match u64::from_str(region_id) {
                Ok(region_id) => region_id,
                Err(err) => return Ok(bad_request_resp(err.to_string().as_str())),
            };
            if ctx.kvengine.get_shard(region_id).is_none() {
                return Ok(not_found_resp(
                    format!("region {} not exists", region_id).as_str(),
                ));
            }
            vec![region_id]
        } else if let Some(keyspace_id) = keyspace_id {
            let keyspace_id = match u32::from_str(keyspace_id) {
                Ok(keyspace_id) => keyspace_id,
                Err(e) => return Ok(bad_request_resp(e.to_string().as_str())),
            };
            let Some(keyspace_shards) = ctx.kvengine.get_keyspace_shards(keyspace_id) else {
                return Ok(not_found_resp(
                    format!("keyspace {} not exists", keyspace_id).as_str(),
                ));
            };

            if let Some(table_id) = table_id {
                let table_id = match u64::from_str(table_id) {
                    Ok(table_id) => table_id,
                    Err(e) => return Ok(bad_request_resp(e.to_string().as_str())),
                };
                let mut table_key = Vec::new();
                table_key.put_u8(b't');
                table_key.encode_i64(table_id as i64).unwrap();
                let table_end_key = next_key(&table_key);
                let table_data_bound = DataBound::new(
                    InnerKey::from_inner_buf(&table_key),
                    InnerKey::from_inner_buf(&table_end_key),
                    false,
                );

                keyspace_shards
                    .iter()
                    .map(|r| *r)
                    .filter_map(|shard_id| ctx.kvengine.get_shard(shard_id))
                    .filter(|shard| table_data_bound.overlap_bound(shard.data_bound()))
                    .map(|shard| shard.id)
                    .collect::<Vec<_>>()
            } else {
                keyspace_shards
                    .iter()
                    .map(|shard_id| *shard_id)
                    .collect::<Vec<_>>()
            }
        } else {
            return Ok(bad_request_resp("query parameters invalid"));
        };
        if target_regions.is_empty() {
            return Ok(not_found_resp("target regions is empty"));
        } else {
            info!("manual major compact regions {:?}", target_regions);
        }
        let mut region_futures = Vec::with_capacity(target_regions.len());
        for region_id in target_regions.iter() {
            let (cb, fu) = paired_future_callback();
            let callback = Callback::write(Box::new(move |_| {
                cb(());
            }));
            region_futures.push(fu);
            ctx.router.send_casual_msg(
                *region_id,
                CasualMessage::MajorCompact {
                    major_compact,
                    columnar,
                    callback,
                },
            );
        }
        let _ = futures::future::join_all(region_futures).await;
        Ok(make_response(
            StatusCode::OK,
            format!(
                "trigger manual major compaction on region(s) {:?} success",
                target_regions
            ),
        ))
    }

    // curl -X POST "http://127.0.0.1:20180/flush?keyspace_id=xxx&table_id=1,2,3"
    // Flush all regions of the specified table(s) for regions whose leader is the
    // current TiKV, and wait until flush finished. Maximum wait time is 60s.
    // table_id can be a single table ID or comma-separated list of table IDs.
    async fn flush(req: Request<Body>, ctx: StatusContext) -> hyper::Result<Response<Body>> {
        let query = req.uri().query().unwrap_or("");
        let query_pairs: HashMap<_, _> = url::form_urlencoded::parse(query.as_bytes()).collect();
        if !query_pairs.contains_key("keyspace_id") || !query_pairs.contains_key("table_id") {
            return Ok(make_response(
                StatusCode::BAD_REQUEST,
                "keyspace_id and table_id are required".to_string(),
            ));
        }
        let keyspace_id = match u32::from_str(query_pairs.get("keyspace_id").unwrap()) {
            Ok(keyspace_id) => keyspace_id,
            Err(err) => return Ok(make_response(StatusCode::BAD_REQUEST, err.to_string())),
        };

        // Parse table_id parameter which can be a single ID or comma-separated list
        let table_id_str = query_pairs.get("table_id").unwrap();
        let mut table_ids = std::collections::HashSet::new();
        for id_str in table_id_str.split(',') {
            let id_str = id_str.trim();
            if id_str.is_empty() {
                continue;
            }
            match i64::from_str(id_str) {
                Ok(id) => {
                    table_ids.insert(id);
                }
                Err(err) => {
                    return Ok(make_response(
                        StatusCode::BAD_REQUEST,
                        format!("Invalid table_id format: {}", err),
                    ));
                }
            }
        }
        if table_ids.is_empty() {
            return Ok(make_response(
                StatusCode::BAD_REQUEST,
                "table_id cannot be empty".to_string(),
            ));
        }
        info!(
            "ManualFlush: Received flush request for keyspace_id={} table_ids={:?}",
            keyspace_id, table_ids
        );

        // Collect regions from keyspace id and table ids.
        // We will skip regions that are not leader on this TiKV, because this manual
        // flush is supposed to be run locally.
        let mut target_regions = Vec::new();
        let Some(ks_shards) = ctx.kvengine.get_keyspace_shards(keyspace_id) else {
            info!("ManualFlush: Keyspace {} not found", keyspace_id);
            return Ok(make_response(
                StatusCode::NOT_FOUND,
                format!("Keyspace {} not found", keyspace_id),
            ));
        };
        let mut table_key = Vec::new();
        for shard_id in ks_shards.iter().map(|r| *r) {
            let Some(shard) = ctx.kvengine.get_shard(shard_id) else {
                continue;
            };
            if !shard.is_active() {
                continue;
            }
            for table_id in table_ids.iter() {
                table_key.clear();
                table_key.put_u8(b't');
                table_key.encode_i64(*table_id).unwrap();

                let table_end_key = next_key(&table_key);

                // Construct a DataBound for the table range.
                let table_data_bound = DataBound::new(
                    InnerKey::from_inner_buf(&table_key),
                    InnerKey::from_inner_buf(&table_end_key),
                    false,
                );

                // The data in a table may be distributed across multiple shards, so the
                // actual range of table_data_bound is larger than shard.data_bound().
                if table_data_bound.overlap_bound(shard.data_bound()) {
                    target_regions.push(shard_id);
                    break;
                }
            }
        }

        target_regions.sort();
        info!(
            "ManualFlush: will flush these local regions: {:?}",
            target_regions
        );
        if target_regions.is_empty() {
            info!("ManualFlush: skipped, no regions found");
            return Ok(make_response(
                StatusCode::OK,
                "Flush skipped, no regions found",
            ));
        }

        // For each shard, we need to wait for a version that signifies that the
        // memtable has been flushed.
        // Case 1. Writable memtable is not empty -- We don't know which version to wait
        // until we switched it. In this case, the memtable ref will be kept so that we
        // can know a version to wait after switch.
        // Case 2. Writable memtable is empty -- We can simply wait existing memtable
        // flush to complete. In this case we record which version to wait and do not
        // need to switch the memtable.

        // 1. Get the writable memtable for each shard, or fill the version to wait
        // if writable memtable is empty.
        let mut wait_versions_by_regions = HashMap::default();
        let mut writable_memtables = HashMap::default();
        target_regions.retain(|region_id| {
            let Some(shard) = ctx.kvengine.get_shard(*region_id) else {
                // Region changed
                return false;
            };
            let memtable = shard.get_writable_mem_table();
            if memtable.size() == 0 {
                let v = shard.get_mem_table_max_version();
                // Note, it is possible that v == 0, if there is only one writable memtable
                // and no sealed memtables (e.g. already flushed).
                info!(
                    "ManualFlush: region {} writable memtable is empty, will wait for version {}",
                    region_id, v
                );
                wait_versions_by_regions.insert(*region_id, v);
            } else {
                // which version to wait is only known after switch.
                // we keep a reference.
                info!(
                    "ManualFlush: region {} writable memtable is not empty, will switch",
                    region_id
                );
                writable_memtables.insert(*region_id, memtable);
            }
            true
        });

        if !writable_memtables.is_empty() {
            // 2. send FORCE_SWITCH_MEM_TABLE messages
            // Only shards with non-empty writable memtable need to switch.
            info!(
                "ManualFlush: calling FORCE_SWITCH_MEM_TABLE for regions: {:?}",
                writable_memtables.keys()
            );
            let mut region_futures = Vec::with_capacity(writable_memtables.len());
            for (&region_id, memtable) in writable_memtables.iter() {
                let (cb, fu) = paired_future_callback();
                let callback = Callback::write(Box::new(move |_| {
                    cb(());
                }));
                region_futures.push(fu);
                ctx.router.send_casual_msg(
                    region_id,
                    CasualMessage::ForceSwitchMemTable {
                        current_size: memtable.size(),
                        callback,
                    },
                );
            }
            let _ = futures::future::join_all(region_futures).await;

            // 3. At this moment, we should be able to know all versions to wait, including
            // those writable memtables (they have a version after switch).
            for (region_id, memtable) in writable_memtables {
                let v = memtable.get_snap_version();
                info!(
                    "ManualFlush: region {} writable memtable is switched, will wait for version {}",
                    region_id, v
                );
                wait_versions_by_regions.insert(region_id, v);
            }
        }

        let max_version_to_wait = wait_versions_by_regions
            .values()
            .max()
            .copied()
            .unwrap_or(SnapVersion::zero());
        if max_version_to_wait.is_zero() {
            info!(
                "ManualFlush: skipped, no memtable to wait on regions: {:?}",
                target_regions
            );
            return Ok(make_response(
                StatusCode::OK,
                format!(
                    "Flush skipped, no memtable to wait on regions: {:?}",
                    target_regions
                ),
            ));
        }

        // 4. Flush will be doing in background, after flush the changeset will be
        // applied, so here we wait for changeset to take effect.
        info!(
            "ManualFlush: waiting for flush to finish on regions: {:?}",
            wait_versions_by_regions.keys()
        );
        let now = Instant::now();
        loop {
            if now.saturating_elapsed_secs() > 60.0 {
                info!(
                    "ManualFlush: timed out waiting for flush to finish on regions: {:?}, will not wait any more",
                    wait_versions_by_regions.keys()
                );
                return Ok(make_response(
                    StatusCode::REQUEST_TIMEOUT,
                    format!(
                        "Timedout waiting flush to finish on regions: {:?}, all flush regions: {:?}",
                        wait_versions_by_regions.keys(),
                        target_regions
                    ),
                ));
            }
            let timer =
                GLOBAL_TIMER_HANDLE.delay(std::time::Instant::now() + Duration::from_secs(1));
            let end = async move {
                Compat01As03::new(timer)
                    .await
                    .map_err(|_| TIMER_CANCELED.to_owned())
            };
            let _ = end.await;
            wait_versions_by_regions.retain(|region_id, version| {
                let Some(shard) = ctx.kvengine.get_shard(*region_id) else {
                    return false;
                };
                if !shard.is_active() {
                    return false;
                }
                // Keep this region only if it's flushed version has not reached
                // what we want.
                shard.get_persisted_snap_version() < *version
            });
            if wait_versions_by_regions.is_empty() {
                break;
            }
        }

        info!("ManualFlush: all finished");
        Ok(make_response(
            StatusCode::OK,
            format!("Flush memtable succeeded on regions: {:?}", target_regions),
        ))
    }

    fn get_dfs_file_id(req: &Request<Body>) -> Option<u64> {
        let path = req.uri().path();
        let last = get_last_path_segment(path);
        u64::from_str(last).ok()
    }

    /// Handle meta request: return file metadata in JSON format.
    async fn handle_meta(
        id: u64,
        file_type: FileType,
        ctx: StatusContext,
    ) -> hyper::Result<Response<Body>> {
        let (callback, future) = paired_future_callback();
        let kvengine = ctx.kvengine.clone();
        spawn_anonymous_thread_with!(move || {
            let res = match file_type {
                FileType::TxnChunk => kvengine.get_txn_chunk_manager().local_chunk_size(id),
                _ => kvengine.local_file_size(id, file_type),
            };
            callback(res)
        });
        let res = future.await.unwrap();
        Ok(match res {
            Ok(size) => make_meta_response(size),
            // Handle IO NotFound error
            Err(kvengine::Error::Io { err, .. }) if err.kind() == std::io::ErrorKind::NotFound => {
                make_response(StatusCode::NOT_FOUND, "")
            }
            // Handle table NotFound error
            Err(kvengine::Error::TableError(ref e))
                if matches!(e, kvengine::table::Error::NotFound) =>
            {
                make_response(StatusCode::NOT_FOUND, "")
            }
            // Handle table IO error with NotFound message
            Err(kvengine::Error::TableError(kvengine::table::Error::Io(ref msg)))
                if msg.contains("NotFound") || msg.contains("No such file") =>
            {
                make_response(StatusCode::NOT_FOUND, "")
            }
            Err(e) => {
                error!("Error getting file metadata: {:?}", e);
                make_response(
                    StatusCode::INTERNAL_SERVER_ERROR,
                    "Error getting file metadata",
                )
            }
        })
    }

    fn get_dfs_read_args(
        req: &Request<Body>,
    ) -> (
        FileType,
        u64,         // start_off
        Option<u64>, // end_off
        bool,        // check_exists_only
        bool,        // meta
    ) {
        let mut file_type: Option<FileType> = None;
        let mut start_off: Option<u64> = None;
        let mut end_off: Option<u64> = None;
        let mut check_exists_only: Option<bool> = None;
        let mut meta: Option<bool> = None;
        if let Some(query) = req.uri().query() {
            let query_pairs: HashMap<_, _> =
                url::form_urlencoded::parse(query.as_bytes()).collect();
            file_type = query_pairs
                .get("file_type")
                .and_then(|s| s.as_ref().try_into().ok());
            start_off = query_pairs
                .get("start_off")
                .and_then(|s| s.parse::<u64>().ok());
            end_off = query_pairs
                .get("end_off")
                .and_then(|s| s.parse::<u64>().ok());
            check_exists_only = query_pairs
                .get("check_exists_only")
                .and_then(|s| s.parse::<bool>().ok());
            meta = query_pairs.get("meta").and_then(|s| s.parse::<bool>().ok());
        }
        (
            file_type.unwrap_or(FileType::Sst),
            start_off.unwrap_or_default(),
            end_off,
            check_exists_only.unwrap_or_default(),
            meta.unwrap_or(false),
        )
    }

    const DFS_FILE_READ_CHUNK_SIZE: usize = 1024 * 1024;

    fn calculate_dfs_file_read_bytes(
        start_off: u64,
        end_off: u64,
    ) -> std::result::Result<u64, DfsFileReadRangeError> {
        if start_off > end_off {
            return Err(DfsFileReadRangeError::StartAfterEnd);
        }
        Ok(end_off - start_off)
    }

    fn spawn_blocking_with_dfs_read_guard<T, F>(
        handle: &tokio::runtime::Handle,
        guard: DfsFileReadRequestGuard,
        task: F,
    ) -> tokio::task::JoinHandle<T>
    where
        T: Send + 'static,
        F: FnOnce(DfsFileReadRequestGuard) -> T + Send + 'static,
    {
        // Dropping a JoinHandle detaches its blocking task. Keep the guard in
        // the task so request cancellation cannot admit another full-file read
        // while this one is still running.
        handle.spawn_blocking(move || task(guard))
    }

    fn make_dfs_file_read_response(
        data: Bytes,
        mut admission: DfsFileReadAdmissionGuard,
    ) -> Response<Body> {
        let content_length = data.len();
        admission.start_response();
        // A wrapped body bypasses Hyper's HTTP/1 whole-body fast path. The stream
        // owns the admission guard for as long as it owns the complete source
        // buffer.
        let stream = futures::stream::unfold(
            (data, 0, admission),
            |(data, start_off, admission)| async move {
                if start_off >= data.len() {
                    return None;
                }
                fail_point!("before_dfs_file_read_response_chunk");
                let end_off = start_off
                    .saturating_add(Self::DFS_FILE_READ_CHUNK_SIZE)
                    .min(data.len());
                // Hyper can drop the body after queueing Content-Length bytes,
                // before the transport flushes its final chunk. Detach a final
                // slice from a larger source so it cannot retain the complete
                // file after the stream and its permit are dropped.
                let chunk = if end_off == data.len() && start_off > 0 {
                    Bytes::copy_from_slice(&data[start_off..end_off])
                } else {
                    data.slice(start_off..end_off)
                };
                Some((Ok::<_, Infallible>(chunk), (data, end_off, admission)))
            },
        );
        Response::builder()
            .header(header::CONTENT_TYPE, "application/octet-stream")
            .header(header::CONTENT_LENGTH, content_length.to_string())
            .body(Body::wrap_stream(stream))
            .unwrap()
    }

    fn make_permit_guarded_response(
        data: Bytes,
        permit: OwnedSemaphorePermit,
        content_type: &'static str,
        chunk_size: usize,
    ) -> Response<Body> {
        let content_length = data.len();
        // A wrapped body keeps the admission permit with the complete source
        // buffer. Detach the final chunk so releasing the permit cannot leave
        // the full response retained in Hyper's transport queue.
        let stream = futures::stream::unfold(
            (data, 0, permit),
            move |(data, start_off, permit)| async move {
                if start_off >= data.len() {
                    return None;
                }
                let end_off = start_off.saturating_add(chunk_size).min(data.len());
                let chunk = if end_off == data.len() && start_off > 0 {
                    Bytes::copy_from_slice(&data[start_off..end_off])
                } else {
                    data.slice(start_off..end_off)
                };
                Some((Ok::<_, Infallible>(chunk), (data, end_off, permit)))
            },
        );
        Response::builder()
            .header(header::CONTENT_TYPE, content_type)
            .header(header::CONTENT_LENGTH, content_length.to_string())
            .body(Body::wrap_stream(stream))
            .unwrap()
    }

    fn read_dfs_file(
        engine: &kvengine::Engine,
        id: u64,
        file_type: FileType,
        start_off: u64,
        end_off: Option<u64>,
        check_exists_only: bool,
        request_guard: DfsFileReadRequestGuard,
    ) -> DfsFileReadTaskResult {
        let requested_bytes = if check_exists_only {
            0
        } else if file_type == FileType::TxnChunk {
            match engine.get_txn_chunk_manager().local_chunk_size(id) {
                Ok(file_size) => file_size,
                Err(err) => return DfsFileReadTaskResult::SizeError(err),
            }
        } else {
            let file_size = match engine.local_file_size(id, file_type) {
                Ok(file_size) => file_size,
                Err(err) => return DfsFileReadTaskResult::SizeError(err),
            };
            let read_end_off = end_off.unwrap_or(file_size);
            if read_end_off > file_size {
                return DfsFileReadTaskResult::InvalidRange(DfsFileReadRangeError::EndAfterFile);
            }
            match Self::calculate_dfs_file_read_bytes(start_off, read_end_off) {
                Ok(requested_bytes) => requested_bytes,
                Err(err) => return DfsFileReadTaskResult::InvalidRange(err),
            }
        };

        if !check_exists_only {
            DFS_FILE_READ_REQUESTED_BYTES.observe(requested_bytes as f64);
        }
        let memory_guard = match engine.try_acquire_dfs_cache_memory(requested_bytes) {
            Ok(guard) => guard,
            Err(rejection) => return DfsFileReadTaskResult::Rejected(rejection),
        };
        let admission =
            DfsFileReadAdmissionGuard::new(request_guard, memory_guard, requested_bytes);

        let read_res = match file_type {
            FileType::TxnChunk => {
                debug_assert_eq!(start_off, 0);
                engine
                    .get_txn_chunk_manager()
                    .read_local_chunk(id, check_exists_only)
            }
            _ => engine.read_local_file(id, file_type, start_off, end_off, check_exists_only),
        };
        let data = match read_res {
            Ok(data) => data,
            Err(err) => return DfsFileReadTaskResult::ReadError(err),
        };
        let actual_bytes = data.len() as u64;
        // Local files are immutable once published; keep this check to defend the
        // charge.
        if actual_bytes != requested_bytes {
            return DfsFileReadTaskResult::SizeChanged {
                expected: requested_bytes,
                actual: actual_bytes,
            };
        }
        DfsFileReadTaskResult::Success { data, admission }
    }

    fn make_dfs_file_read_task_response(
        id: u64,
        check_exists_only: bool,
        start: Instant,
        task_result: DfsFileReadTaskResult,
    ) -> Response<Body> {
        let (data, admission) = match task_result {
            DfsFileReadTaskResult::Success { data, admission } => (data, admission),
            DfsFileReadTaskResult::Rejected(rejection) => {
                DFS_FILE_READ_UNSUCCESSFUL_TOTAL
                    .with_label_values(&[rejection.metric_label()])
                    .inc();
                debug!(
                    "DFS file read admission rejected";
                    "reason" => rejection.metric_label(),
                );
                return make_response(
                    StatusCode::SERVICE_UNAVAILABLE,
                    "DFS file read admission limit exceeded",
                );
            }
            DfsFileReadTaskResult::InvalidRange(range_err) => {
                DFS_FILE_READ_UNSUCCESSFUL_TOTAL
                    .with_label_values(&["invalid_range"])
                    .inc();
                return make_response(
                    StatusCode::BAD_REQUEST,
                    format!("invalid DFS file read range {:?}", range_err),
                );
            }
            DfsFileReadTaskResult::SizeError(size_err) => {
                DFS_FILE_READ_UNSUCCESSFUL_TOTAL
                    .with_label_values(&["size_lookup_error"])
                    .inc();
                error!("Failed to get DFS file size: {:?}", size_err);
                return make_response(
                    StatusCode::INTERNAL_SERVER_ERROR,
                    format!("failed to get DFS file size {:?}", size_err),
                );
            }
            DfsFileReadTaskResult::ReadError(read_err) => {
                if check_exists_only && is_not_found_error(&read_err) {
                    return make_response(StatusCode::NOT_FOUND, "");
                }
                DFS_FILE_READ_UNSUCCESSFUL_TOTAL
                    .with_label_values(&["read_error"])
                    .inc();
                error!("Failed to read file: {:?}", read_err);
                return make_response(
                    StatusCode::INTERNAL_SERVER_ERROR,
                    format!("failed to read file {:?}", read_err),
                );
            }
            DfsFileReadTaskResult::SizeChanged { expected, actual } => {
                DFS_FILE_READ_UNSUCCESSFUL_TOTAL
                    .with_label_values(&["size_changed"])
                    .inc();
                error!(
                    "DFS file size changed while reading";
                    "expected_bytes" => expected,
                    "actual_bytes" => actual,
                );
                return make_response(
                    StatusCode::INTERNAL_SERVER_ERROR,
                    "DFS file size changed while reading",
                );
            }
        };
        info!(
            "read local dfs file";
            "file_id" => id,
            "size" => data.len(),
            "takes" => ?start.saturating_elapsed(),
        );
        Self::make_dfs_file_read_response(data, admission)
    }

    async fn handle_dfs_file_read(
        req: Request<Body>,
        ctx: StatusContext,
    ) -> hyper::Result<Response<Body>> {
        let Some(id) = Self::get_dfs_file_id(&req) else {
            return Ok(make_response(StatusCode::BAD_REQUEST, "invalid file id"));
        };
        let (file_type, start_off, end_off, check_exists_only, meta) =
            Self::get_dfs_read_args(&req);

        if meta {
            // Handle meta query: return file metadata as JSON
            return Self::handle_meta(id, file_type, ctx).await;
        }
        let handle = match tokio::runtime::Handle::try_current() {
            Ok(handle) => handle,
            Err(handle_err) => {
                error!("Failed to get tokio runtime handle: {:?}", handle_err);
                return Ok(make_response(
                    StatusCode::INTERNAL_SERVER_ERROR,
                    "failed to get tokio runtime handle",
                ));
            }
        };
        let request_limit = ctx.server_config.value().dfs_file_read_concurrency;
        let request_guard =
            match DfsFileReadRequestGuard::try_acquire(&ctx.dfs_file_read_in_flight, request_limit)
            {
                Err(in_flight) => {
                    DFS_FILE_READ_UNSUCCESSFUL_TOTAL
                        .with_label_values(&["request_limit"])
                        .inc();
                    debug!(
                        "too many concurrent DFS file read requests";
                        "in_flight" => in_flight,
                        "limit" => request_limit,
                    );
                    return Ok(make_response(
                        StatusCode::SERVICE_UNAVAILABLE,
                        "too many concurrent DFS file read requests",
                    ));
                }
                Ok(guard) => guard,
            };
        let start = Instant::now_coarse();
        let engine = ctx.kvengine.clone();
        let task_result = Self::spawn_blocking_with_dfs_read_guard(
            &handle,
            request_guard,
            move |request_guard| {
                Self::read_dfs_file(
                    &engine,
                    id,
                    file_type,
                    start_off,
                    end_off,
                    check_exists_only,
                    request_guard,
                )
            },
        )
        .await;
        let task_result = match task_result {
            Ok(task_result) => task_result,
            Err(handle_err) => {
                error!("Failed to execute blocking task: {:?}", handle_err);
                return Ok(make_response(
                    StatusCode::INTERNAL_SERVER_ERROR,
                    "failed to execute blocking task",
                ));
            }
        };
        Ok(Self::make_dfs_file_read_task_response(
            id,
            check_exists_only,
            start,
            task_result,
        ))
    }

    async fn handle_dfs_file_create(
        req: Request<Body>,
        engine: kvengine::Engine,
    ) -> hyper::Result<Response<Body>> {
        let id_opt = Self::get_dfs_file_id(&req);
        if id_opt.is_none() {
            return Ok(make_response(StatusCode::BAD_REQUEST, "invalid file id"));
        }
        let id = id_opt.unwrap();
        let (file_type, ..) = Self::get_dfs_read_args(&req);
        let data = hyper::body::to_bytes(req.into_body()).await?;
        let (callback, future) = paired_future_callback();
        std::thread::spawn(move || {
            let res = match file_type {
                FileType::TxnChunk => engine.get_txn_chunk_manager().write_local_chunk(id, data),
                _ => engine.write_local_file_if_not_exists(id, data, file_type),
            };
            callback(res);
        });
        let res = future.await.unwrap();
        Ok(match res {
            Ok(_) => make_response(StatusCode::OK, ""),
            Err(err) => make_response(
                StatusCode::INTERNAL_SERVER_ERROR,
                format!("Internal Server Error {}", err),
            ),
        })
    }

    async fn handle_recovery(
        req: Request<Body>,
        engine: &kvengine::Engine,
        data_dir: &str,
    ) -> hyper::Result<Response<Body>> {
        let path = req.uri().path();
        match path {
            "/recovery/mode" => Ok(recovery::handle_mode(req, || {
                engine.unblock_keyspace_compaction()
            })),
            "/recovery/rejected" => {
                let pending_compact_shards = engine.get_pending_compaction_shards();
                Ok(recovery::handle_rejected(req, pending_compact_shards))
            }
            "/recovery/white_list" => Ok(recovery::handle_while_list(req, || {
                engine.unblock_keyspace_compaction()
            })),
            "/recovery/black_list" => Ok(recovery::handle_black_list(req, data_dir)),
            _ => Ok(make_response(StatusCode::NOT_FOUND, "Not Found")),
        }
    }

    async fn generate_encryption_meta(
        req: Request<Body>,
        engine: kvengine::Engine,
        pd_client: Arc<dyn PdClient>,
    ) -> hyper::Result<Response<Body>> {
        let body = hyper::body::to_bytes(req.into_body()).await?;

        let request: EncryptionMetaGenRequest = match serde_json::from_slice(&body) {
            Ok(r) => r,
            Err(e) => {
                return Ok(make_response(
                    StatusCode::BAD_REQUEST,
                    format!("invalid request body: {}", e),
                ));
            }
        };

        let shard = match engine.get_shard(request.region_id) {
            Some(s) => s,
            None => {
                return Ok(make_response(
                    StatusCode::NOT_FOUND,
                    format!("shard {} not found", request.region_id),
                ));
            }
        };

        // Validate that region_id belongs to the specified keyspace_id
        if shard.range.keyspace_id != request.keyspace_id {
            return Ok(make_response(
                StatusCode::BAD_REQUEST,
                format!(
                    "region {} belongs to keyspace {}, not keyspace {}",
                    request.region_id, shard.range.keyspace_id, request.keyspace_id
                ),
            ));
        }

        let existing_meta_bytes = shard.get_property(ENCRYPTION_META_KEY);

        let existing_meta = match parse_existing_meta(existing_meta_bytes) {
            Ok(meta) => meta,
            Err(e) => {
                return Ok(make_response(e.status_code(), e.to_string()));
            }
        };

        let encryption_key_manager = engine.get_encryption_key_manager();
        let result = match request.mode {
            EncryptionMetaGenMode::RotateDataKey { max_data_key_age } => {
                encryption_meta::derive_rotate_data_key(
                    existing_meta,
                    request.keyspace_id,
                    max_data_key_age,
                    encryption_key_manager.as_ref(),
                    &pd_client,
                )
                .await
            }
            EncryptionMetaGenMode::EnableCmek {
                cmek_id,
                vendor,
                region,
                endpoint,
                encryption_context,
            } => {
                encryption_meta::derive_enable_cmek(
                    existing_meta,
                    request.keyspace_id,
                    cmek_id,
                    vendor,
                    region,
                    endpoint,
                    encryption_context,
                    encryption_key_manager.as_ref(),
                    &pd_client,
                )
                .await
            }
        };

        // Handle derive result
        let result = match result {
            Ok(r) => r,
            Err(e) => {
                return Ok(make_response(e.status_code(), e.to_string()));
            }
        };

        match result {
            DeriveResult::NewMeta {
                meta,
                file_id,
                data_key_id,
                new_data_key_generated,
            } => {
                let mut opts = kvengine::dfs::Options::default();
                opts.file_type = kvengine::dfs::FileType::EncryptionDict;
                let content = meta.write_to_bytes().unwrap();

                if let Err(e) = engine.core.fs.create(file_id, content.into(), opts).await {
                    return Ok(make_response(
                        StatusCode::INTERNAL_SERVER_ERROR,
                        format!("failed to upload to S3: {}", e),
                    ));
                }

                let response = EncryptionMetaGenResponse {
                    file_id,
                    data_key_id,
                    new_data_key_generated,
                };
                Ok(Response::builder()
                    .status(StatusCode::OK)
                    .header(header::CONTENT_TYPE, "application/json")
                    .body(Body::from(serde_json::to_string(&response).unwrap()))
                    .unwrap())
            }
            DeriveResult::Skipped {
                file_id,
                data_key_id,
            } => {
                let response = EncryptionMetaGenResponse {
                    file_id,
                    data_key_id,
                    new_data_key_generated: Some(false),
                };
                Ok(Response::builder()
                    .status(StatusCode::OK)
                    .header(header::CONTENT_TYPE, "application/json")
                    .body(Body::from(serde_json::to_string(&response).unwrap()))
                    .unwrap())
            }
        }
    }

    async fn backup_rfengine(
        req: Request<Body>,
        ctx: StatusContext,
    ) -> hyper::Result<Response<Body>> {
        let body = hyper::body::to_bytes(req.into_body()).await?;
        let backup_config: serde_json::Result<rfengine::BackupConfig> =
            serde_json::from_slice(&body);
        if backup_config.is_err() {
            return Ok(make_response(StatusCode::BAD_REQUEST, "Bad request body"));
        }
        let backup_config = backup_config.unwrap();

        // Check if lightweight backup enabled.
        if backup_config.lightweight && !ctx.rfengine.is_lightweight_backup_enabled() {
            return Ok(make_response(
                StatusCode::BAD_REQUEST,
                "lightweight backup not enabled",
            ));
        }

        let cluster_id = backup_config.cluster_id;
        let mut store_ident = StoreIdent::default();
        let data = ctx
            .rfengine
            .get_state(0, rfengine::STORE_IDENT_KEY)
            .unwrap_or_default();
        store_ident.merge_from_bytes(data.chunk()).unwrap();
        if store_ident.cluster_id != cluster_id {
            warn!(
                "{}: backup id mismatch expect {:?}, got {}",
                store_ident.store_id, store_ident.cluster_id, cluster_id
            );
            return Ok(make_response(
                StatusCode::BAD_REQUEST,
                "cluster id mismatch",
            ));
        }

        let (wait_backup_ts_ok, wait_backup_ts_dur) =
            if let Some(backup_ts) = backup_config.backup_ts {
                if let Err(e) = ctx
                    .concurrency_manager
                    .update_max_ts(backup_ts.into(), "status_server_backup")
                {
                    warn!("failed to update max_ts in concurrency manager"; "err" => ?e);
                }

                ctx.concurrency_manager.replace_backup_ts(backup_ts.into());
                let start_time = Instant::now_coarse();
                let ok = ctx
                    .concurrency_manager
                    .wait_old_backup_ts_released(
                        backup_ts.into(),
                        backup_config.backup_ts_wait_timeout(),
                        backup_config.backup_ts_ttl(),
                        BACKUP_TS_WAIT_RETRY_INTERVAL,
                    )
                    .await;
                (ok, Some(start_time.saturating_elapsed()))
            } else {
                (true, None)
            };
        let (callback, future) = paired_future_callback();
        let task = rfengine::BackupTask::new(callback, backup_config);
        ctx.rfengine.backup(task);
        Ok(match future.await {
            Ok(resp) => match resp {
                Ok(mut meta) => {
                    info!("{}: backup finished", meta.store_id; "wait_backup_ts" => ?wait_backup_ts_dur);
                    estimate_backup_size_by(&ctx.kvengine, meta.mut_keyspace_size());
                    meta.set_has_missing_commit_record(!wait_backup_ts_ok);
                    Response::builder()
                        .body(Body::from(meta.write_to_bytes().unwrap()))
                        .unwrap()
                }
                Err(err) => {
                    error!("{}: backup failed {:?}", store_ident.store_id, &err);
                    make_response(
                        StatusCode::INTERNAL_SERVER_ERROR,
                        format!("Internal Server Error {}", err),
                    )
                }
            },
            Err(e) => make_response(
                StatusCode::INTERNAL_SERVER_ERROR,
                format!("Internal Server Error {}", e),
            ),
        })
    }

    async fn rfengine_track_wal_progress(
        req: Request<Body>,
        engine: &RfEngine,
    ) -> hyper::Result<Response<Body>> {
        let body = hyper::body::to_bytes(req.into_body()).await?;
        let _track_req: TrackWalProgressRequest = match serde_json::from_slice(&body) {
            Ok(track_req) => track_req,
            Err(err) => {
                return Ok(make_response(
                    StatusCode::BAD_REQUEST,
                    format!("bad request {}", err),
                ));
            }
        };
        // TODO: use track_req.ts to track async commit locks.

        let (callback, future) = paired_future_callback();
        engine.get_wal_progress(callback);

        Ok(match future.await {
            Ok(resp) => match resp {
                Ok(progress) => Response::builder()
                    .status(StatusCode::OK)
                    .body(Body::from(serde_json::to_string(&progress).unwrap()))
                    .unwrap(),
                Err(err) => {
                    error!("rfengine_track_wal_progress error: {:?}", err);
                    make_response(StatusCode::INTERNAL_SERVER_ERROR, format!("{}", err))
                }
            },
            Err(e) => make_response(
                StatusCode::INTERNAL_SERVER_ERROR,
                format!("Internal Server Error {}", e),
            ),
        })
    }

    async fn flush_rfengine_wal(engine: &rfengine::RfEngine) -> hyper::Result<Response<Body>> {
        engine.upload_wal_chunk().await;
        let healthy = engine.is_dfs_worker_healthy();
        if healthy {
            Ok(make_response(StatusCode::OK, ""))
        } else {
            Ok(make_response(
                StatusCode::SERVICE_UNAVAILABLE,
                json!("DFS worker isn't healthy, please check the log").to_string(),
            ))
        }
    }

    async fn get_rfengine_dfs_worker_status(
        engine: &rfengine::RfEngine,
    ) -> hyper::Result<Response<Body>> {
        Ok(match engine.get_dfs_worker_status().await {
            Ok(status) => json_response(StatusCode::OK, &status),
            Err(err) => {
                error!("rfengine dfs worker status error: {:?}", err);
                make_response(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
            }
        })
    }

    fn get_covered_shards_by_range(
        range: Option<(Vec<u8>, Vec<u8>)>,
        engine: &kvengine::Engine,
    ) -> Result<Vec<Arc<Shard>>> {
        let shards = engine.get_all_shard_id_vers();
        let mut partial_covered_shard = None;
        let ret: Vec<Arc<Shard>> = shards
            .into_iter()
            .filter_map(|s| engine.get_shard(s.id))
            .filter(|s| {
                if range.is_none() {
                    return true;
                }
                let range = range.as_ref().unwrap();
                let full_cover = s.outer_start >= range.0 && s.outer_end <= range.1;
                if !full_cover && s.outer_start < range.1 && s.outer_end > range.0 {
                    partial_covered_shard = Some(s.clone());
                }
                full_cover
            })
            .collect();
        if let Some(s) = partial_covered_shard {
            return Err(box_err!(
                "Shard {} [{:?}-{:?}) is partial covered by range {:?}",
                s.tag(),
                s.outer_start,
                s.outer_end,
                range
            ));
        }
        Ok(ret)
    }

    async fn get_restore_shard_request(
        raw_req: Request<Body>,
        expected_conf_ver: Option<u64>,
    ) -> hyper::Result<RestoreShardRequest> {
        let mut body = Vec::new();
        raw_req
            .into_body()
            .try_for_each(|bytes| {
                body.extend(bytes);
                ok(())
            })
            .await?;
        let mut cs = kvenginepb::ChangeSet::default();
        cs.merge_from_bytes(&body).unwrap();
        Ok(RestoreShardRequest {
            cs,
            expected_conf_ver,
        })
    }

    async fn restore_shard(
        req: Request<Body>,
        ctx: StatusContext,
    ) -> hyper::Result<Response<Body>> {
        let accept_pb = req.headers().is_accept_protobuf();
        let query = req.uri().query().unwrap_or("");
        let query_pairs: HashMap<_, _> = url::form_urlencoded::parse(query.as_bytes()).collect();
        let expected_conf_ver = match get_optional_uint_param(&query_pairs, "expected_conf_ver") {
            Ok(value) => value,
            Err(err) => return Ok(make_response(StatusCode::BAD_REQUEST, err)),
        };
        let req = Self::get_restore_shard_request(req, expected_conf_ver).await?;
        let tag = tag_from_cs(&ctx.kvengine, &req.cs);
        let shard_id = req.cs.get_shard_id();
        debug!("[{}] receive restore_shard request: {:?}", tag, req);

        if let Err(errpb) = check_available_space(&ctx.kvengine) {
            warn!("[{}] reject restore shard, low space", tag; "err" => ?errpb);
            return Ok(make_errpb_response(
                StatusCode::INTERNAL_SERVER_ERROR,
                &errpb,
                accept_pb,
            ));
        }
        if let Err(errpb) =
            check_follower_available_space(ctx.clone(), shard_id, req.cs.shard_ver).await
        {
            warn!("[{}] reject restore shard, follower low space", tag; "err" => ?errpb);
            return Ok(make_errpb_response(
                StatusCode::INTERNAL_SERVER_ERROR,
                &errpb,
                accept_pb,
            ));
        }

        let (cb, fut) = paired_future_callback();
        let callback = Callback::write(Box::new(move |res| {
            cb(res);
        }));
        ctx.router.send_casual_msg(
            req.cs.get_shard_id(),
            CasualMessage::RestoreShard {
                cs: req.cs,
                expected_conf_ver: req.expected_conf_ver,
                callback,
            },
        );

        let res = match fut.await {
            Ok(res) => res,
            Err(e) => {
                let err_msg = format!("{} restore_shard channel error: {:?}", shard_id, e);
                error!("{}", err_msg);
                return Ok(make_response(StatusCode::INTERNAL_SERVER_ERROR, err_msg));
            }
        };
        if res.response.get_header().has_error() {
            error!(
                "{} restore_shard error: {:?}",
                tag,
                res.response.get_header().get_error()
            );
            Ok(make_errpb_response(
                StatusCode::INTERNAL_SERVER_ERROR,
                res.response.get_header().get_error(),
                accept_pb,
            ))
        } else {
            let stat = ctx.kvengine.get_shard_stat(shard_id);
            let resp = RestoreShardResponse {
                shard_id,
                restore_bytes: stat.kv_size,
            };
            let json = serde_json::to_string_pretty(&resp).unwrap();
            Ok(Response::builder()
                .header(header::CONTENT_TYPE, "application/json")
                .body(Body::from(json))
                .unwrap())
        }
    }

    async fn handle_schema_file(
        req: Request<Body>,
        ctx: StatusContext,
    ) -> hyper::Result<Response<Body>> {
        let bad_request_resp = |msg: &str| make_response(StatusCode::BAD_REQUEST, msg.to_owned());
        let query = req.uri().query().unwrap_or("");
        let query_pairs: HashMap<_, _> = url::form_urlencoded::parse(query.as_bytes()).collect();
        let keyspace_id = match get_uint_param(&query_pairs, "keyspace_id") {
            Some(keyspace_id) => keyspace_id,
            None => {
                return Ok(bad_request_resp("keyspace_id not found"));
            }
        };
        let file_id = match get_uint_param(&query_pairs, "file_id") {
            Some(file_id) => file_id,
            None => {
                return Ok(bad_request_resp("file_id not found"));
            }
        };
        let schema_file = match ctx.kvengine.load_schema_file(file_id, IoType::Other).await {
            Ok(file) => file,
            Err(e) => {
                let msg = format!("failed to load schema file {:?}", e);
                return Ok(make_response(StatusCode::INTERNAL_SERVER_ERROR, msg));
            }
        };
        let (start, end) = ApiV2::get_txn_keyspace_range(keyspace_id as u32);
        let (callback, future) = paired_future_callback();
        ctx.router.send_store_msg(StoreMsg::GetRegionsInRange {
            start,
            end,
            callback,
        });
        let res = future.await;
        if let Err(err) = res {
            return Ok(make_response(
                StatusCode::INTERNAL_SERVER_ERROR,
                format!("failed to get regions in range: {err:?}"),
            ));
        }
        let region_id_vers = res.unwrap();
        for region_id_ver in region_id_vers {
            ctx.router.send_casual_msg(
                region_id_ver.id(),
                CasualMessage::UpdateSchemaFile(schema_file.clone()),
            );
        }
        Ok(Response::new(Body::empty()))
    }

    /// Clear columnar data of the specified keyspace or shard.
    ///
    /// If keyspace_id is 0, clear all columnar data of all shards.
    /// POST /clear_columnar?[restore_version=xxxx]
    /// [&keyspace_id=xxx][&shard_id=1][&confirm_all=true|false]
    async fn handle_clear_columnar(
        req: Request<Body>,
        ctx: StatusContext,
    ) -> hyper::Result<Response<Body>> {
        let bad_request_resp = |msg: &str| make_response(StatusCode::BAD_REQUEST, msg.to_owned());
        let query = req.uri().query().unwrap_or("");
        let query_pairs: HashMap<_, _> = url::form_urlencoded::parse(query.as_bytes()).collect();
        if !query_pairs.contains_key("keyspace_id") && !query_pairs.contains_key("shard_id") {
            return Ok(bad_request_resp("keyspace_id or shard_id not found"));
        }
        // If restore_version is not set, do no update the restore version.
        let restore_version = get_uint_param(&query_pairs, "restore_version");
        let mut shard_ids = vec![];
        if query_pairs.contains_key("keyspace_id") {
            let keyspace_id = match get_uint_param(&query_pairs, "keyspace_id") {
                Some(keyspace_id) => keyspace_id,
                None => {
                    return Ok(bad_request_resp("keyspace_id not found"));
                }
            };
            // If keyspace_id is 0, clear all columnar data of all shards. To avoid mistake,
            // we require confirm_all to be true.
            let (start, end) = if keyspace_id == 0 {
                // Check confirm_all to avoid mistake.
                if !query_pairs.contains_key("confirm_all") {
                    return Ok(bad_request_resp("param confirm_all not found"));
                }
                let confirmed = match get_bool_param(&query_pairs, "confirm_all") {
                    Some(confirmed) => confirmed,
                    None => return Ok(bad_request_resp("param confirm_all must be true or false")),
                };
                if !confirmed {
                    return Ok(bad_request_resp(
                        "please set param confirm_all to true to clear all columnar data",
                    ));
                }
                (vec![TXN_KEY_PREFIX], GLOBAL_SHARD_END_KEY.to_vec())
            } else {
                ApiV2::get_txn_keyspace_range(keyspace_id as u32)
            };
            let (callback, future) = paired_future_callback();
            ctx.router.send_store_msg(StoreMsg::GetRegionsInRange {
                start,
                end,
                callback,
            });
            let res = future.await;
            if let Err(err) = res {
                return Ok(make_response(
                    StatusCode::INTERNAL_SERVER_ERROR,
                    format!("failed to get regions in range: {err:?}"),
                ));
            }
            let region_id_vers = res.unwrap();
            for region_id_ver in region_id_vers {
                shard_ids.push(region_id_ver.id());
                ctx.router.send_casual_msg(
                    region_id_ver.id(),
                    CasualMessage::ClearColumnar { restore_version },
                );
            }
        } else {
            let shard_id = match get_uint_param(&query_pairs, "shard_id") {
                Some(shard_id) => shard_id,
                None => {
                    return Ok(bad_request_resp("shard_id not found"));
                }
            };
            shard_ids.push(shard_id);
            ctx.router
                .send_casual_msg(shard_id, CasualMessage::ClearColumnar { restore_version });
        }

        Ok(Response::new(Body::from(format!(
            "clear columnar success, shard_ids: {:?}\n",
            shard_ids
        ))))
    }

    /// Enable/disable to build columnar tables
    ///
    /// POST /build_columnar?switch=true|false
    async fn handle_build_columnar(
        req: Request<Body>,
        engine: &kvengine::Engine,
    ) -> hyper::Result<Response<Body>> {
        let query = req.uri().query().unwrap_or("");
        let query_pairs: HashMap<_, _> = url::form_urlencoded::parse(query.as_bytes()).collect();
        if let Some(switch) = get_bool_param(&query_pairs, "switch") {
            let previous = engine.opts.set_build_columnar(switch);
            info!("build columnar switch: {} -> {}", previous, switch);
            Ok(make_response(
                StatusCode::OK,
                format!("build columnar: {} -> {}", previous, switch),
            ))
        } else {
            Ok(make_response(
                StatusCode::BAD_REQUEST,
                "param switch not found",
            ))
        }
    }

    /// Clear fts index of the specified keyspace or shard.
    ///
    /// If keyspace_id is 0, clear all fts index of all shards.
    /// POST /clear_fts_index?
    /// [&keyspace_id=xxx][&shard_id=1][&confirm_all=true|false]
    async fn handle_clear_fts_index(
        req: Request<Body>,
        ctx: StatusContext,
    ) -> hyper::Result<Response<Body>> {
        let bad_request_resp = |msg: &str| make_response(StatusCode::BAD_REQUEST, msg.to_owned());
        let query = req.uri().query().unwrap_or("");
        let query_pairs: HashMap<_, _> = url::form_urlencoded::parse(query.as_bytes()).collect();
        if !query_pairs.contains_key("keyspace_id") && !query_pairs.contains_key("shard_id") {
            return Ok(bad_request_resp("keyspace_id or shard_id not found"));
        }
        let mut shard_ids = vec![];
        if query_pairs.contains_key("keyspace_id") {
            let keyspace_id = match get_uint_param(&query_pairs, "keyspace_id") {
                Some(keyspace_id) => keyspace_id,
                None => {
                    return Ok(bad_request_resp("keyspace_id not found"));
                }
            };
            // If keyspace_id is 0, clear all fts index of all shards. To avoid mistake,
            // we require confirm_all to be true.
            let (start, end) = if keyspace_id == 0 {
                // Check confirm_all to avoid mistake.
                if !query_pairs.contains_key("confirm_all") {
                    return Ok(bad_request_resp("param confirm_all not found"));
                }
                let confirmed = match get_bool_param(&query_pairs, "confirm_all") {
                    Some(confirmed) => confirmed,
                    None => return Ok(bad_request_resp("param confirm_all must be true or false")),
                };
                if !confirmed {
                    return Ok(bad_request_resp(
                        "please set param confirm_all to true to clear all fts index",
                    ));
                }
                (vec![TXN_KEY_PREFIX], GLOBAL_SHARD_END_KEY.to_vec())
            } else {
                ApiV2::get_txn_keyspace_range(keyspace_id as u32)
            };
            let (callback, future) = paired_future_callback();
            ctx.router.send_store_msg(StoreMsg::GetRegionsInRange {
                start,
                end,
                callback,
            });
            let res = future.await;
            if let Err(err) = res {
                return Ok(make_response(
                    StatusCode::INTERNAL_SERVER_ERROR,
                    format!("failed to get regions in range: {err:?}"),
                ));
            }
            let region_id_vers = res.unwrap();
            for region_id_ver in region_id_vers {
                shard_ids.push(region_id_ver.id());
                ctx.router
                    .send_casual_msg(region_id_ver.id(), CasualMessage::ClearFtsIndex);
            }
        } else {
            let shard_id = match get_uint_param(&query_pairs, "shard_id") {
                Some(shard_id) => shard_id,
                None => {
                    return Ok(bad_request_resp("shard_id not found"));
                }
            };
            shard_ids.push(shard_id);
            ctx.router
                .send_casual_msg(shard_id, CasualMessage::ClearFtsIndex);
        }

        Ok(Response::new(Body::from(format!(
            "clear fts index success, shard_ids: {:?}\n",
            shard_ids
        ))))
    }

    /// Enable/disable to build fts index
    ///
    /// POST /build_fts_index?switch=true|false
    async fn handle_build_fts_index(
        req: Request<Body>,
        engine: &kvengine::Engine,
    ) -> hyper::Result<Response<Body>> {
        let query = req.uri().query().unwrap_or("");
        let query_pairs: HashMap<_, _> = url::form_urlencoded::parse(query.as_bytes()).collect();
        if let Some(switch) = get_bool_param(&query_pairs, "switch") {
            let previous = engine.opts.set_build_fts_index(switch);
            info!("build fts index switch: {} -> {}", previous, switch);
            Ok(make_response(
                StatusCode::OK,
                format!("build fts index: {} -> {}", previous, switch),
            ))
        } else {
            Ok(make_response(
                StatusCode::BAD_REQUEST,
                "param switch not found",
            ))
        }
    }

    /// Get `EncryptionMeta` for the given keyspace_id or region id.
    ///
    /// This endpoint is intended for other components (e.g. CDC) to fetch
    /// encryption meta from TiKV for encrypting or decrypting external data.
    ///
    /// For keyspace requests, `latest_unstable` defaults to `false`, which
    /// returns the old stable encryption config if a rollout is still in
    /// progress. When `latest_unstable=true`, the latest pending-rollout meta
    /// is returned even if it has not been rolled out to all regions in the
    /// keyspace yet. For external components, sticking to the default is
    /// usually the right choice.
    ///
    /// Response:
    /// - Default: protobuf (`application/protobuf`)
    /// - If there is no encryption for the target: `204 No Content`
    ///
    /// GET /encryption/get-meta?keyspace_id=xxx
    /// GET /encryption/get-meta?keyspace_id=xxx&latest_unstable=true
    /// GET /encryption/get-meta?region_id=xxx
    async fn handle_get_encryption_meta(
        req: Request<Body>,
        engine: &kvengine::Engine,
        pd_client: Arc<dyn PdClient>,
    ) -> hyper::Result<Response<Body>> {
        let bad_request_resp = |msg: &str| make_response(StatusCode::BAD_REQUEST, msg.to_owned());

        let query = req.uri().query().unwrap_or("");
        let query_pairs: HashMap<_, _> = url::form_urlencoded::parse(query.as_bytes()).collect();

        let has_keyspace_id = query_pairs.contains_key("keyspace_id");
        let has_region_id = query_pairs.contains_key("region_id");
        if !has_keyspace_id && !has_region_id {
            return Ok(bad_request_resp("keyspace_id or region_id not found"));
        }
        if has_keyspace_id && has_region_id {
            return Ok(bad_request_resp(
                "please specify either keyspace_id or region_id",
            ));
        }

        if has_region_id {
            let region_id = match get_uint_param(&query_pairs, "region_id") {
                Some(region_id) => region_id,
                None => return Ok(bad_request_resp("region_id not found")),
            };

            let shard = match engine.get_shard(region_id) {
                Some(s) => s,
                None => {
                    return Ok(make_response(
                        StatusCode::NOT_FOUND,
                        format!("region {region_id} not found"),
                    ));
                }
            };

            let meta_bytes = match shard.get_property(ENCRYPTION_META_KEY) {
                Some(b) => b,
                None => {
                    // No encryption for this region.
                    return Ok(Response::builder()
                        .status(StatusCode::NO_CONTENT)
                        .body(Body::empty())
                        .unwrap());
                }
            };

            return Ok(Response::builder()
                .status(StatusCode::OK)
                .header(CONTENT_TYPE, CONTENT_TYPE_PROTOBUF)
                .body(meta_bytes.into())
                .unwrap());
        }

        let keyspace_id = {
            let keyspace_id = match get_uint_param(&query_pairs, "keyspace_id") {
                Some(keyspace_id) => keyspace_id,
                None => return Ok(bad_request_resp("keyspace_id not found")),
            };
            let Ok(keyspace_id) = keyspace_id.try_into() else {
                return Ok(bad_request_resp("keyspace_id is too large"));
            };
            keyspace_id
        };

        let latest_unstable = if query_pairs.contains_key("latest_unstable") {
            match get_bool_param(&query_pairs, "latest_unstable") {
                Some(v) => v,
                None => return Ok(bad_request_resp("invalid latest_unstable")),
            }
        } else {
            false
        };

        let cached_keyspace_meta = engine
            .get_encryption_key_manager()
            .encryption_meta_by_keyspace(keyspace_id);

        let keyspace_id_for_pd = keyspace_id;
        let pd_status_task = tokio::task::spawn_blocking(move || {
            let mgr = encryption_conf_change::EncryptionConfChangeManager::new(
                pd_client,
                OwnerCheckMode::Disabled,
            );
            mgr.load_keyspace_encryption_status_from_pd(keyspace_id_for_pd)
        });

        let pd_status = match pd_status_task.await {
            Ok(Ok(s)) => s,
            Ok(Err(e)) if e.is_keyspace_not_found() => {
                return Ok(make_response(
                    StatusCode::NOT_FOUND,
                    format!("keyspace_id {keyspace_id} not found in PD"),
                ));
            }
            Ok(Err(e)) => {
                return Ok(make_response(
                    StatusCode::INTERNAL_SERVER_ERROR,
                    format!("failed to load keyspace config from PD: {e}"),
                ));
            }
            Err(e) => {
                return Ok(make_response(
                    StatusCode::INTERNAL_SERVER_ERROR,
                    format!("failed to load keyspace config from PD: {e}"),
                ));
            }
        };

        let selection = pd_status.select_meta_id(latest_unstable);

        use encryption_conf_change::KeyspaceEncryptionMetaId;

        match selection {
            KeyspaceEncryptionMetaId::Unencrypted => Ok(Response::builder()
                .status(StatusCode::NO_CONTENT)
                .body(Body::empty())
                .unwrap()),
            KeyspaceEncryptionMetaId::UnavailableInPd => {
                let Some(meta) = cached_keyspace_meta else {
                    return Ok(make_response(
                        StatusCode::NOT_FOUND,
                        format!(
                            "keyspace_id {keyspace_id} encryption is enabled but unavailable in PD and no cached encryption meta found on this node"
                        ),
                    ));
                };
                let body = match meta.write_to_bytes() {
                    Ok(body) => body,
                    Err(e) => {
                        let msg = format!(
                            "failed to serialize encryption meta for keyspace {keyspace_id}: {e}"
                        );
                        error!("{}", msg);
                        return Ok(make_response(StatusCode::INTERNAL_SERVER_ERROR, msg));
                    }
                };
                Ok(Response::builder()
                    .status(StatusCode::OK)
                    .header(CONTENT_TYPE, CONTENT_TYPE_PROTOBUF)
                    .body(body.into())
                    .unwrap())
            }
            KeyspaceEncryptionMetaId::MetaFileId(file_id) => {
                let meta = if let Some(meta) = cached_keyspace_meta
                    .as_ref()
                    .filter(|meta| meta.get_current().get_file_id() == file_id)
                {
                    meta.clone()
                } else {
                    match encryption_conf_change::load_encryption_meta_from_dfs(
                        engine.fs.clone(),
                        file_id,
                    )
                    .await
                    {
                        Ok(meta) => meta,
                        Err(encryption_conf_change::EncryptionMetaLoadError::Dfs(
                            kvengine::dfs::Error::NotExists(_),
                        )) => {
                            return Ok(make_response(
                                StatusCode::NOT_FOUND,
                                format!("encryption meta file_id {file_id} not found in DFS"),
                            ));
                        }
                        Err(encryption_conf_change::EncryptionMetaLoadError::Dfs(e)) => {
                            return Ok(make_response(
                                StatusCode::INTERNAL_SERVER_ERROR,
                                format!(
                                    "failed to read encryption meta file_id {file_id} from DFS: {e}"
                                ),
                            ));
                        }
                        Err(encryption_conf_change::EncryptionMetaLoadError::Decode {
                            source,
                            ..
                        }) => {
                            return Ok(make_response(
                                StatusCode::INTERNAL_SERVER_ERROR,
                                format!(
                                    "failed to parse encryption meta file_id {file_id}: {source}"
                                ),
                            ));
                        }
                    }
                };

                let body = match meta.write_to_bytes() {
                    Ok(body) => body,
                    Err(e) => {
                        let msg =
                            format!("failed to serialize encryption meta file_id {file_id}: {e}");
                        error!("{}", msg);
                        return Ok(make_response(StatusCode::INTERNAL_SERVER_ERROR, msg));
                    }
                };
                Ok(Response::builder()
                    .status(StatusCode::OK)
                    .header(CONTENT_TYPE, CONTENT_TYPE_PROTOBUF)
                    .body(body.into())
                    .unwrap())
            }
        }
    }

    pub fn stop(self) {
        let _ = self.close_tx.send(());
        let _ = self.close_handle.unwrap().join();
        shutdown_runtimes!(self.thread_pool);
    }

    // Return listening address, this may only be used for outer test
    // to get the real address because we may use "127.0.0.1:0"
    // in test to avoid port conflict.
    #[allow(unused)]
    pub fn listening_addr(&self) -> SocketAddr {
        self.addr.unwrap()
    }
}

#[cfg(debug_assertions)]
impl StatusServer {
    async fn handle_debug_sleep() -> hyper::Result<Response<Body>> {
        info!("sleep for 5 seconds");
        std::thread::sleep(Duration::from_secs(5));
        info!("sleep done");
        Ok(Response::default())
    }
}

fn get_last_path_segment(path: &str) -> &str {
    let mut last = "";
    for i in (0..path.len()).rev() {
        if path.as_bytes()[i] == b'/' {
            last = &path[i + 1..];
            break;
        }
    }
    last
}

fn get_uint_param(query_pairs: &HashMap<Cow<'_, str>, Cow<'_, str>>, key: &str) -> Option<u64> {
    query_pairs.get(key).and_then(|v| u64::from_str(v).ok())
}

fn get_optional_uint_param(
    query_pairs: &HashMap<Cow<'_, str>, Cow<'_, str>>,
    key: &str,
) -> std::result::Result<Option<u64>, String> {
    query_pairs
        .get(key)
        .map(|value| {
            u64::from_str(value).map_err(|_| format!("invalid {key}: expected an unsigned integer"))
        })
        .transpose()
}

fn get_online_snapshot_branch_ts(
    query_pairs: &HashMap<Cow<'_, str>, Cow<'_, str>>,
) -> std::result::Result<Option<u64>, String> {
    let Some(value) = query_pairs.get("branch_ts") else {
        return Ok(None);
    };
    let branch_ts =
        u64::from_str(value).map_err(|_| "branch_ts must be an unsigned integer".to_owned())?;
    let branch_ts = TimeStamp::new(branch_ts);
    if branch_ts.is_zero() || branch_ts.is_max() {
        return Err("branch_ts must be greater than zero and less than u64::MAX".to_owned());
    }
    Ok(Some(branch_ts.into_inner()))
}

fn get_bounded_duration_param(
    query_pairs: &HashMap<Cow<'_, str>, Cow<'_, str>>,
    key: &str,
    default: Duration,
    max: Duration,
) -> std::result::Result<Duration, String> {
    let Some(value) = query_pairs.get(key) else {
        return Ok(default);
    };
    let seconds = u64::from_str(value).map_err(|_| format!("{key} must be an unsigned integer"))?;
    let duration = Duration::from_secs(seconds);
    if duration.is_zero() || duration > max {
        return Err(format!(
            "{key} must be between 1 and {} seconds",
            max.as_secs()
        ));
    }
    Ok(duration)
}

fn get_bool_param(query_pairs: &HashMap<Cow<'_, str>, Cow<'_, str>>, key: &str) -> Option<bool> {
    query_pairs.get(key).and_then(|v| bool::from_str(v).ok())
}

impl StatusServer {
    pub async fn dump_region_meta(
        _req: Request<Body>,
        _router: RaftRouter,
    ) -> hyper::Result<Response<Body>> {
        // TODO(x)
        Ok(hyper::Response::new(Body::empty()))
    }

    async fn handle_sync_region_by_id(
        req: Request<Body>,
        ctx: StatusContext,
    ) -> hyper::Result<Response<Body>> {
        let mut body = Vec::new();
        req.into_body()
            .try_for_each(|bytes| {
                body.extend(bytes);
                ok(())
            })
            .await?;

        let SyncRegionByIdRequest { region_id, debug } = match serde_json::from_slice(&body) {
            Ok(req) => req,
            Err(e) => {
                return Ok(make_response(
                    StatusCode::BAD_REQUEST,
                    format!("invalid request body: {:?}", e),
                ));
            }
        };

        let resp = ctx.region_info_accessor.find_region_by_id(region_id);
        let body = if debug {
            format!("{:?}", resp).into_bytes()
        } else {
            Self::make_sync_region_resp(
                ctx.stores_info.pd_client.get_cluster_id().unwrap(),
                ctx.rfengine.get_engine_id(),
                resp.as_slice(),
            )
            .write_to_bytes()
            .unwrap()
        };
        Ok(Response::new(body.into()))
    }

    async fn handle_sync_region(
        req: Request<Body>,
        ctx: StatusContext,
    ) -> hyper::Result<Response<Body>> {
        let mut body = Vec::new();
        req.into_body()
            .try_for_each(|bytes| {
                body.extend(bytes);
                ok(())
            })
            .await?;

        let SyncRegionRequest {
            start,
            end,
            limit,
            reverse,
            debug,
        } = match serde_json::from_slice(&body) {
            Ok(req) => req,
            Err(e) => {
                return Ok(make_response(
                    StatusCode::BAD_REQUEST,
                    format!("invalid request body: {:?}", e),
                ));
            }
        };

        let (start, mut end) = match (hex::decode(&start), hex::decode(&end)) {
            (Ok(start), Ok(end)) if end.is_empty() || start < end => (start, end),
            _ => {
                return Ok(make_response(
                    StatusCode::BAD_REQUEST,
                    format!("invalid range: [{}, {})", start, end),
                ));
            }
        };

        let start = encode_bytes(&start);
        if end.is_empty() {
            end.extend(GLOBAL_SHARD_END_KEY);
        }
        let end = encode_bytes(&end);
        let resp = ctx
            .region_info_accessor
            .get_regions_in_range_opt(&start, &end, true, reverse, limit);
        let body = if debug {
            format!("{:#?}", resp).into_bytes()
        } else {
            Self::make_sync_region_resp(
                ctx.stores_info.pd_client.get_cluster_id().unwrap(),
                ctx.rfengine.get_engine_id(),
                &resp,
            )
            .write_to_bytes()
            .unwrap()
        };
        Ok(Response::new(body.into()))
    }

    fn make_sync_region_resp(
        cluster_id: u64,
        store_id: u64,
        region_infos: &[RegionInfo],
    ) -> SyncRegionResponse {
        let cap = region_infos.len();
        let mut resp_regions = Vec::with_capacity(cap);
        let mut resp_stats = Vec::with_capacity(cap);
        let mut resp_leaders = Vec::with_capacity(cap);
        let mut resp_buckets = Vec::with_capacity(cap);
        let mut resp_down_peers: Vec<pdpb::PeersStats> = Vec::with_capacity(cap);
        let mut resp_pending_peers: Vec<Peers> = Vec::with_capacity(cap);
        for info in region_infos {
            let region = &info.region;
            resp_regions.push(region.clone());
            // The stats are used along with region, we need to push a default one if not
            // found.
            let region_stat = pdpb::RegionStat::new();
            let down_peers = pdpb::PeersStats::new();
            let pending_peers = Peers::new();
            resp_stats.push(region_stat);
            resp_down_peers.push(down_peers);
            resp_pending_peers.push(pending_peers);
            let leader_peer = find_peer(region, store_id).cloned().unwrap_or_default();
            resp_leaders.push(leader_peer);
            let mut bucket = metapb::Buckets::new();
            bucket.set_region_id(region.id);
            bucket.set_version(info.buckets.version);
            bucket.set_keys(info.buckets.keys.clone().into());
            bucket.set_stats(BucketStats::new());
            resp_buckets.push(bucket);
        }
        let mut resp = SyncRegionResponse::new();
        resp.mut_header().set_cluster_id(cluster_id);
        resp.set_regions(resp_regions.into());
        resp.set_region_leaders(resp_leaders.into());
        resp.set_region_stats(resp_stats.into());
        resp.set_buckets(resp_buckets.into());
        resp.set_down_peers(resp_down_peers.into());
        resp.set_pending_peers(resp_pending_peers.into());
        resp
    }

    fn handle_get_metrics(
        req: Request<Body>,
        mgr: &ConfigController,
    ) -> hyper::Result<Response<Body>> {
        let should_simplify = mgr.get_current().server.simplify_metrics;
        let gz_encoding = client_accept_gzip(&req);
        let metrics = if gz_encoding {
            // gzip can reduce the body size to less than 1/10.
            let mut encoder = GzEncoder::new(vec![], Compression::default());
            dump_to(&mut encoder, should_simplify);
            encoder.finish().unwrap()
        } else {
            dump(should_simplify).into_bytes()
        };
        let mut resp = Response::new(metrics.into());
        resp.headers_mut()
            .insert(CONTENT_TYPE, HeaderValue::from_static(TEXT_FORMAT));
        if gz_encoding {
            resp.headers_mut()
                .insert(CONTENT_ENCODING, HeaderValue::from_static("gzip"));
        }

        Ok(resp)
    }

    fn handle_ready_request(req: Request<Body>) -> hyper::Result<Response<Body>> {
        let verbose = req
            .uri()
            .query()
            .is_some_and(|query| query.contains("verbose"));

        let status_code = if GLOBAL_SERVER_READINESS.is_ready() {
            StatusCode::OK
        } else {
            StatusCode::INTERNAL_SERVER_ERROR
        };

        let body = if verbose {
            GLOBAL_SERVER_READINESS.to_json()
        } else {
            "".to_string()
        };
        Ok(make_response(status_code, body))
    }

    fn start_serve<I, C>(&mut self, builder: HyperBuilder<I>)
    where
        I: Accept<Conn = C, Error = std::io::Error> + Send + 'static,
        I::Error: Into<Box<dyn StdError + Send + Sync>>,
        I::Conn: AsyncRead + AsyncWrite + Unpin + Send + 'static,
        C: ServerConnection,
    {
        let owner_check = match load_store_ident(&self.rfengine) {
            Some(ident) if ident.store_id != 0 => OwnerCheckMode::Enforced {
                owner_store_id: ident.store_id,
            },
            Some(_) => {
                warn!("encryption conf-change owner check disabled: store_id is 0");
                OwnerCheckMode::Disabled
            }
            None => {
                warn!("encryption conf-change owner check disabled: store ident not found");
                OwnerCheckMode::Disabled
            }
        };
        let encryption_conf_change_mgr =
            EncryptionConfChangeManager::new(self.pd_client.clone(), owner_check);

        let ctx_inner = StatusContextInner {
            security_config: self.security_config.clone(),
            cfg_controller: self.cfg_controller.clone(),
            kvengine: self.kvengine.clone(),
            rfengine: self.rfengine.clone(),
            concurrency_manager: self.concurrency_manager.clone(),
            server_config: self.server_config.clone(),
            dfs_file_read_in_flight: Arc::new(AtomicUsize::new(0)),
            online_snapshot_dump_semaphore: Arc::new(Semaphore::new(
                online_snapshot_dump_concurrency(SysQuota::cpu_cores_quota()),
            )),
            online_snapshot_memory_semaphore: Arc::new(Semaphore::new(
                ONLINE_SNAPSHOT_MEMORY_PERMITS,
            )),
            online_snapshot_prepared_ts: AtomicU64::new(0),
            stores_info: Arc::new(StoresInfo::new(self.pd_client.clone())),
            thread_pool: self.thread_pool.handle().clone(),
            region_info_accessor: self.region_info_accessor.clone(),
            pd_client: self.pd_client.clone(),
            encryption_conf_change_mgr,
        };
        let ctx = StatusContext {
            inner: Arc::new(ctx_inner),
            router: self.router.clone(),
        };
        // Start to serve.
        let server = builder.serve(make_service_fn(move |conn: &C| {
            let x509 = conn.get_x509();
            let ctx = ctx.clone();
            async move {
                // Create a status service.
                Ok::<_, hyper::Error>(service_fn(move |req: Request<Body>| {
                    let x509 = x509.clone();
                    let ctx = ctx.clone();
                    tikv_util::init_task_local(async move {
                        let path = req.uri().path().to_owned();
                        let method = req.method().to_owned();

                        #[cfg(feature = "failpoints")]
                        {
                            if path.starts_with(FAIL_POINTS_REQUEST_PATH) {
                                return handle_fail_points_request(req).await;
                            }
                        }

                        // 1. POST "/config" will modify the configuration of TiKV.
                        // 2. GET "/region" will get start key and end key. These keys could be
                        // actual user data since in some cases the data
                        // itself is stored in the key.
                        let should_check_cert = !matches!(
                            (&method, path.as_ref()),
                            (&Method::GET, "/metrics")
                                | (&Method::GET, "/status")
                                | (&Method::GET, "/config")
                                | (&Method::GET, "/debug/pprof/profile")
                        );

                        if should_check_cert && !check_cert(&ctx.security_config, x509) {
                            return Ok(make_response(
                                StatusCode::FORBIDDEN,
                                "certificate role error",
                            ));
                        }

                        let mut is_unknown_path = false;
                        let start = Instant::now();
                        let resp = match (method.clone(), path.as_ref()) {
                            (Method::GET, "/metrics") => {
                                Self::handle_get_metrics(req, &ctx.cfg_controller)
                            }
                            (Method::GET, "/status") => Ok(Response::default()),
                            (Method::GET, "/ready") => {
                                Self::handle_ready_request(req)
                            }
                            (Method::GET, "/debug/pprof/heap_list") => {
                                Ok(make_response(
                                    StatusCode::GONE,
                                    "Deprecated, heap profiling is always enabled by default, just use /debug/pprof/heap to get the heap profile when needed",
                                ))
                            }
                            (Method::GET, "/debug/pprof/heap_activate") => {
                                Ok(make_response(
                                    StatusCode::GONE,
                                    "Deprecated, use config `memory.enable_heap_profiling` to toggle",
                                ))
                            }
                            (Method::GET, "/debug/pprof/heap_deactivate") => {
                                Ok(make_response(
                                    StatusCode::GONE,
                                    "Deprecated, use config `memory.enable_heap_profiling` to toggle",
                                ))
                            }
                            (Method::GET, "/debug/pprof/heap") => {
                                Self::dump_heap_prof_to_resp(req)
                            }
                            (Method::GET, "/debug/pprof/cmdline") => Self::get_cmdline(req),
                            (Method::GET, "/debug/pprof/symbol") => {
                                Self::get_symbol_count(req)
                            }
                            (Method::POST, "/debug/pprof/symbol") => Self::get_symbol(req).await,
                            (Method::GET, "/config") => {
                                Self::get_config(req, &ctx.cfg_controller).await
                            }
                            (Method::POST, "/config") => {
                                Self::update_config(ctx.cfg_controller.clone(), req).await
                            }
                            (Method::GET, "/debug/pprof/profile") => {
                                Self::dump_cpu_prof_to_resp(req).await
                            }
                            (Method::GET, "/debug/fail_point") => {
                                info!("debug fail point API start");
                                fail_point!("debug_fail_point");
                                info!("debug fail point API finish");
                                Ok(Response::default())
                            }
                            #[cfg(debug_assertions)]
                            (Method::GET, "/debug/sleep") => Self::handle_debug_sleep().await,
                            (Method::GET, path) if path.starts_with("/region") => {
                                Self::dump_region_meta(req, ctx.router).await
                            }
                            (Method::GET, path) if path.starts_with("/sync_region_by_id") => {
                                Self::handle_sync_region_by_id(req, ctx).await
                            }
                            (Method::GET, path) if path.starts_with("/sync_region") => {
                                Self::handle_sync_region(req, ctx).await
                            }
                            (Method::PUT, path) if path.starts_with("/log-level") => {
                                Self::change_log_level(req).await
                            }
                            (Method::GET, path) if path.starts_with("/kvengine") => {
                                if path.starts_with("/kvengine/online-snapshot/file-refs/") {
                                    Self::get_online_snapshot_file_refs(req, ctx).await
                                } else if path.starts_with("/kvengine/online-snapshot/capture/") {
                                    Self::capture_online_snapshot(req, ctx).await
                                } else if path.starts_with("/kvengine/snapshot/") {
                                    Self::dump_kvengine_snapshot(req, ctx).await
                                } else if path.starts_with("/kvengine/columnar_status") {
                                    Self::collect_columnar_status(req, &ctx)
                                } else if path.starts_with("/kvengine/storage_class_status") {
                                    Self::collect_storage_class_status(req, &ctx)
                                } else if path.starts_with("/kvengine/columnar_index_stats") {
                                    Self::collect_columnar_index_stats(req, ctx).await
                                } else if path.starts_with("/kvengine/meta/") {
                                    Self::dump_kvengine_meta(req, &ctx.kvengine).await
                                } else {
                                    Self::dump_kvengine_stats(req, &ctx.kvengine).await
                                }
                            }
                            (Method::POST, "/kvengine/online-snapshot/prepare") => {
                                Self::prepare_online_snapshot(req, ctx).await
                            }
                            (Method::GET, path) if path.starts_with("/rfengine") => {
                                if path == "/rfengine/dfs_worker" {
                                    Self::get_rfengine_dfs_worker_status(&ctx.rfengine).await
                                } else if path.starts_with("/rfengine/wal_chunk") {
                                    Self::rfengine_wal_chunk(req, &ctx.rfengine).await
                                } else {
                                    Self::dump_rfengine_stats(req, &ctx.rfengine).await
                                }
                            }
                            (Method::POST, path) if path.starts_with("/rfengine/backup") => {
                                Self::backup_rfengine(req, ctx).await
                            }
                            (Method::POST, path) if path.starts_with("/rfengine/track_wal_progress") => {
                                Self::rfengine_track_wal_progress(req, &ctx.rfengine).await
                            }
                            (Method::POST, path) if path.starts_with("/rfengine/flush") => {
                                Self::flush_rfengine_wal(&ctx.rfengine).await
                            }
                            (Method::POST, path) if path.starts_with("/restore-shard") => {
                                Self::restore_shard(req, ctx).await
                            }
                            (Method::POST, path) if path.starts_with("/kvengine/compactor") => {
                                Self::add_remote_compactor(req, &ctx.kvengine.comp_client).await
                            }
                            (Method::POST, path) if path.starts_with("/ingest_files") => {
                                Self::ingest_files(req, ctx).await
                            }
                            (Method::POST, path) if path.starts_with("/ingest_s3") => {
                                Self::ingest_s3(req, ctx).await
                            }
                            (Method::POST, path) if path.starts_with("/unsafe_recover") => {
                                Self::unsafe_recover(req, &ctx.rfengine, &ctx.kvengine).await
                            }
                            (Method::POST, path) if path.starts_with("/major-compact") => {
                                Self::major_compact(req, ctx).await
                            }
                            (Method::POST, path) if path.starts_with("/flush") => {
                                Self::flush(req, ctx).await
                            }
                            (Method::POST, path) if path.starts_with("/schema_file") => {
                                Self::handle_schema_file(req, ctx).await
                            }
                            (Method::POST, path) if path.starts_with("/clear_columnar") => {
                                Self::handle_clear_columnar(req, ctx).await
                            }
                            (Method::POST, path) if path.starts_with("/build_columnar") => {
                                Self::handle_build_columnar(req, &ctx.kvengine).await
                            }
                            (Method::POST, path) if path.starts_with("/clear_fts_index") => {
                                Self::handle_clear_fts_index(req, ctx).await
                            }
                            (Method::POST, path) if path.starts_with("/build_fts_index") => {
                                Self::handle_build_fts_index(req, &ctx.kvengine).await
                            }
                            (Method::GET, path) if path.starts_with("/dfs/") => {
                                Self::handle_dfs_file_read(req, ctx).await
                            }
                            (Method::POST, path) if path.starts_with("/dfs/") => {
                                Self::handle_dfs_file_create(req, ctx.kvengine.clone()).await
                            }
                            (Method::GET | Method::POST, path) if path.starts_with("/recovery/") => {
                                let data_dir = &ctx.cfg_controller.get_current().storage.data_dir;
                                Self::handle_recovery(req, &ctx.kvengine.clone(), data_dir).await
                            },
                            (Method::POST, "/encryption/start_rollout_job") => {
                                handle_encryption_start_rollout_job(
                                    &ctx.encryption_conf_change_mgr,
                                    req,
                                )
                                .await
                            }
                            (Method::POST, "/encryption/cancel_rollout_job") => {
                                handle_encryption_cancel_rollout_job(
                                    &ctx.encryption_conf_change_mgr,
                                    req,
                                )
                                .await
                            }
                            (Method::POST, path) if path.starts_with("/encryption/gen-meta") => {
                                Self::generate_encryption_meta(req, ctx.kvengine.clone(), ctx.pd_client.clone()).await
                            },
                            (Method::GET, path) if path.starts_with("/encryption/get-meta") => {
                                Self::handle_get_encryption_meta(
                                    req,
                                    &ctx.kvengine,
                                    ctx.pd_client.clone(),
                                )
                                .await
                            },
                            (Method::POST, path) if path.starts_with("/encryption/sync-latest") => {
                                Self::sync_latest_encryption_meta(req, ctx).await
                            },
                            _ => {
                                is_unknown_path = true;
                                Ok(make_response(StatusCode::NOT_FOUND, "path not found"))
                            }
                        };
                        let dur = start.saturating_elapsed();
                        // Using a first-segment path label (or "unknown") to avoid high cardinality.
                        let path_label = if is_unknown_path {
                            "unknown"
                        } else {
                            get_first_path_segment(path.as_str())
                        };
                        STATUS_REQUEST_DURATION
                            .with_label_values(&[method.as_str(), path_label])
                            .observe(dur.as_secs_f64());
                        if dur > Duration::from_millis(100) {
                            info!(
                                "slow status request";
                                "path" => %path,
                                "method" => ?method,
                                "elapsed" => ?dur,
                            );
                        }
                        resp
                    })
                }))
            }
        }));

        let close_rx = self.close_rx.take().unwrap();
        let thread_pool = self.thread_pool.clone();
        let close_handle = std::thread::spawn(move || {
            thread_pool.block_on(async move {
                tokio::select! {
                    res = server => {
                        res.expect("status server exit with error");
                    }
                    _ = close_rx => {}
                }
            })
        });
        self.close_handle = Some(close_handle);
    }

    pub fn start(&mut self, status_addr: String) -> Result<()> {
        let addr = SocketAddr::from_str(&status_addr)?;

        let mut incoming = {
            let _enter = self.thread_pool.enter();
            AddrIncoming::bind(&addr)
        }?;

        incoming.set_keepalive(Some(SERVER_TCP_KEEPALIVE));
        incoming.set_nodelay(true);

        self.addr = Some(incoming.local_addr());

        match self.pd_client.get_security_mgr().acceptor(incoming)? {
            Either::Left(addr_incoming) => {
                let server =
                    Server::builder(addr_incoming).http1_header_read_timeout(SERVER_READ_TIMEOUT);
                self.start_serve(server);
            }
            Either::Right(tls_acceptor) => {
                let server =
                    Server::builder(tls_acceptor).http1_header_read_timeout(SERVER_READ_TIMEOUT);
                self.start_serve(server);
            }
        };
        Ok(())
    }
}

struct StatusContextInner {
    security_config: Arc<SecurityConfig>,
    cfg_controller: ConfigController,
    kvengine: kvengine::Engine,
    rfengine: RfEngine,
    concurrency_manager: ConcurrencyManager,
    server_config: Arc<VersionTrack<ServerConfig>>,
    dfs_file_read_in_flight: Arc<AtomicUsize>,
    online_snapshot_dump_semaphore: Arc<Semaphore>,
    online_snapshot_memory_semaphore: Arc<Semaphore>,
    online_snapshot_prepared_ts: AtomicU64,
    stores_info: Arc<StoresInfo>,
    thread_pool: tokio::runtime::Handle,
    region_info_accessor: RegionInfoAccessor,
    pd_client: Arc<dyn PdClient>,
    encryption_conf_change_mgr: EncryptionConfChangeManager,
}

#[derive(Clone)]
struct StatusContext {
    inner: Arc<StatusContextInner>,
    router: RaftRouter,
}

impl ops::Deref for StatusContext {
    type Target = StatusContextInner;

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

// To unify TLS/Plain connection usage in start_serve function
trait ServerConnection {
    fn get_x509(&self) -> Option<X509>;
}

impl ServerConnection for TlsStream<AddrStream> {
    fn get_x509(&self) -> Option<X509> {
        // TODO: support return x509
        None
    }
}

impl ServerConnection for AddrStream {
    fn get_x509(&self) -> Option<X509> {
        None
    }
}

#[derive(Clone)]
struct X509 {
    common_name: String,
}

// Check if the peer's x509 certificate meets the requirements, this should
// be called where the access should be controlled.
//
// For now, the check only verifies the role of the peer certificate.
// If x509 is unavailable (e.g., rustls), the check is skipped.
fn check_cert(security_config: &SecurityConfig, x509: Option<X509>) -> bool {
    let Some(x509) = x509 else {
        return true;
    };
    // if `cert_allowed_cn` is empty, skip check and return true
    if !security_config.cert_allowed_cn.is_empty() {
        // Check common name in peer cert
        security::match_peer_names(&security_config.cert_allowed_cn, &x509.common_name)
    } else {
        true
    }
}
// For handling fail points related requests
#[cfg(feature = "failpoints")]
async fn handle_fail_points_request(req: Request<Body>) -> hyper::Result<Response<Body>> {
    let path = req.uri().path().to_owned();
    let method = req.method().to_owned();
    let fail_path = format!("{}/", FAIL_POINTS_REQUEST_PATH);
    let fail_path_has_sub_path: bool = path.starts_with(&fail_path);

    match (method, fail_path_has_sub_path) {
        (Method::PUT, true) => {
            let mut buf = Vec::new();
            req.into_body()
                .try_for_each(|bytes| {
                    buf.extend(bytes);
                    ok(())
                })
                .await?;
            let (_, name) = path.split_at(fail_path.len());
            if name.is_empty() {
                return Ok(Response::builder()
                    .status(StatusCode::UNPROCESSABLE_ENTITY)
                    .body(MISSING_NAME.into())
                    .unwrap());
            };

            let actions = String::from_utf8(buf).unwrap_or_default();
            if actions.is_empty() {
                return Ok(Response::builder()
                    .status(StatusCode::UNPROCESSABLE_ENTITY)
                    .body(MISSING_ACTIONS.into())
                    .unwrap());
            };

            if let Err(e) = fail::cfg(name.to_owned(), &actions) {
                return Ok(Response::builder()
                    .status(StatusCode::BAD_REQUEST)
                    .body(e.into())
                    .unwrap());
            }
            let body = format!("Added fail point with name: {}, actions: {}", name, actions);
            Ok(Response::new(body.into()))
        }
        (Method::DELETE, true) => {
            let (_, name) = path.split_at(fail_path.len());
            if name.is_empty() {
                return Ok(Response::builder()
                    .status(StatusCode::UNPROCESSABLE_ENTITY)
                    .body(MISSING_NAME.into())
                    .unwrap());
            };

            fail::remove(name);
            let body = format!("Deleted fail point with name: {}", name);
            Ok(Response::new(body.into()))
        }
        (Method::GET, _) => {
            // In this scope the path must be like /fail...(/...), which starts with
            // FAIL_POINTS_REQUEST_PATH and may or may not have a sub path
            // Now we return 404 when path is neither /fail nor /fail/
            if path != FAIL_POINTS_REQUEST_PATH && path != fail_path {
                return Ok(Response::builder()
                    .status(StatusCode::NOT_FOUND)
                    .body(Body::empty())
                    .unwrap());
            }

            // From here path is either /fail or /fail/, return lists of fail points
            let list: Vec<String> = fail::list()
                .into_iter()
                .map(move |(name, actions)| format!("{}={}", name, actions))
                .collect();
            let list = list.join("\n");
            Ok(Response::new(list.into()))
        }
        _ => Ok(Response::builder()
            .status(StatusCode::METHOD_NOT_ALLOWED)
            .body(Body::empty())
            .unwrap()),
    }
}

// Reports whether the client accepts gzip compression. Adapted from Prometheus:
// https://github.com/prometheus/client_golang/blob/24172847e35ba46025c49d90b8846b59eb5d9ead/prometheus/promhttp/http.go#L367-L378
// Copyright 2016 The Prometheus Authors
// Modifications Copyright 2022 PingCAP, Inc.
fn client_accept_gzip(req: &Request<Body>) -> bool {
    let encoding = req
        .headers()
        .get(ACCEPT_ENCODING)
        .map(|enc| enc.to_str().unwrap_or_default())
        .unwrap_or_default();
    encoding
        .split(',')
        .map(|s| s.trim())
        .any(|s| s == "gzip" || s.starts_with("gzip;"))
}

// Decode different type of json value to string value
fn decode_json(
    data: &[u8],
) -> std::result::Result<std::collections::HashMap<String, String>, Box<dyn std::error::Error>> {
    let json: Value = serde_json::from_slice(data)?;
    if let Value::Object(map) = json {
        let mut dst = std::collections::HashMap::new();
        for (k, v) in map.into_iter() {
            let v = match v {
                Value::Bool(v) => format!("{}", v),
                Value::Number(v) => format!("{}", v),
                Value::String(v) => v,
                Value::Array(_) => return Err("array type are not supported".to_owned().into()),
                _ => return Err("wrong format".to_owned().into()),
            };
            dst.insert(k, v);
        }
        Ok(dst)
    } else {
        Err("wrong format".to_owned().into())
    }
}

fn make_response<T>(status_code: StatusCode, message: T) -> Response<Body>
where
    T: Into<Body>,
{
    Response::builder()
        .status(status_code)
        .body(message.into())
        .unwrap()
}

#[derive(Debug, serde::Serialize)]
struct EncryptionConfChangeErrorResponse {
    error: String,
    message: String,
}

fn json_response<T: serde::Serialize>(status_code: StatusCode, value: &T) -> Response<Body> {
    let body = match serde_json::to_vec(value) {
        Ok(body) => body,
        Err(e) => {
            return Response::builder()
                .status(StatusCode::INTERNAL_SERVER_ERROR)
                .header(header::CONTENT_TYPE, "text/plain")
                .body(Body::from(format!(
                    "failed to serialize json response: {}",
                    e
                )))
                .unwrap();
        }
    };

    Response::builder()
        .status(status_code)
        .header(header::CONTENT_TYPE, "application/json")
        .body(Body::from(body))
        .unwrap()
}

async fn handle_encryption_start_rollout_job(
    mgr: &EncryptionConfChangeManager,
    req: Request<Body>,
) -> hyper::Result<Response<Body>> {
    let body = hyper::body::to_bytes(req.into_body()).await?;
    let start_req: StartRolloutJobRequest = match serde_json::from_slice(&body) {
        Ok(v) => v,
        Err(e) => {
            return Ok(json_response(
                StatusCode::BAD_REQUEST,
                &EncryptionConfChangeErrorResponse {
                    error: "invalid_json".to_owned(),
                    message: e.to_string(),
                },
            ));
        }
    };

    match mgr.handle_start_rollout_job(start_req).await {
        Ok(result) => Ok(json_response(StatusCode::OK, &result.response)),
        Err(e) => Ok(conf_change_error_to_response(e)),
    }
}

async fn handle_encryption_cancel_rollout_job(
    mgr: &EncryptionConfChangeManager,
    req: Request<Body>,
) -> hyper::Result<Response<Body>> {
    let body = hyper::body::to_bytes(req.into_body()).await?;
    let cancel_req: CancelRolloutJobRequest = match serde_json::from_slice(&body) {
        Ok(v) => v,
        Err(e) => {
            return Ok(json_response(
                StatusCode::BAD_REQUEST,
                &EncryptionConfChangeErrorResponse {
                    error: "invalid_json".to_owned(),
                    message: e.to_string(),
                },
            ));
        }
    };

    match mgr.handle_cancel_rollout_job(cancel_req).await {
        Ok(resp) => Ok(json_response(StatusCode::OK, &resp)),
        Err(e) => Ok(conf_change_error_to_response(e)),
    }
}

fn conf_change_error_to_response(e: EncryptionConfChangeError) -> Response<Body> {
    let (status_code, error) = match &e {
        EncryptionConfChangeError::WrongOwner { .. } => (StatusCode::CONFLICT, "wrong_owner"),
        _ => (StatusCode::INTERNAL_SERVER_ERROR, "internal_error"),
    };
    json_response(
        status_code,
        &EncryptionConfChangeErrorResponse {
            error: error.to_owned(),
            message: e.to_string(),
        },
    )
}

/// Create a meta response with file metadata in JSON format.
///
/// Returns a JSON response containing file metadata like size.
fn make_meta_response(file_size: u64) -> Response<Body> {
    let info = BuiltinDfsFileMeta { size: file_size };
    let json = serde_json::to_string(&info).unwrap();
    Response::builder()
        .status(StatusCode::OK)
        .header(header::CONTENT_TYPE, "application/json")
        .body(Body::from(json))
        .unwrap()
}

fn is_not_found_error(err: &kvengine::Error) -> bool {
    match err {
        kvengine::Error::Io { err, .. } if err.kind() == std::io::ErrorKind::NotFound => true,
        kvengine::Error::TableError(e) => match e {
            kvengine::table::Error::NotFound => true,
            kvengine::table::Error::Io(msg) => {
                msg.contains("NotFound") || msg.contains("No such file")
            }
            _ => false,
        },
        _ => false,
    }
}

fn make_errpb_response(
    status_code: StatusCode,
    errpb: &kvproto::errorpb::Error,
    accept_pb: bool,
) -> Response<Body> {
    if !accept_pb {
        make_response(status_code, errpb.get_message().to_string())
    } else {
        Response::builder()
            .status(status_code)
            .header(CONTENT_TYPE, CONTENT_TYPE_PROTOBUF)
            .body(errpb.write_to_bytes().unwrap().into())
            .unwrap()
    }
}

fn check_available_space(
    engine: &kvengine::Engine,
) -> std::result::Result<(), kvproto::errorpb::Error> {
    if engine.is_low_space() {
        Err(make_errpb_disk_full(
            vec![engine.get_engine_id()],
            format!(
                "available space is low: {}",
                ReadableSize(engine.available_space()).as_mb_f64()
            ),
        ))
    } else {
        Ok(())
    }
}

async fn check_follower_available_space(
    ctx: StatusContext,
    region_id: u64,
    region_ver: u64,
) -> std::result::Result<(), errorpb::Error> {
    let (cb, fut) = paired_future_callback();
    ctx.router.send_store_msg(StoreMsg::GetRegionById {
        region_id,
        callback: cb,
    });
    let region = fut
        .await
        .map_err(|e| {
            // Return not leader to let caller retry.
            make_errpb_not_leader(
                region_id,
                None,
                format!("failed to get region {}: {}", region_id, e),
            )
        })?
        .ok_or_else(|| make_errpb_region_not_found(region_id))?;
    if region.get_region_epoch().version != region_ver {
        return Err(make_errpb_epoch_not_match(region));
    }

    let current_store_id = ctx.kvengine.get_engine_id();
    let followers = region
        .get_peers()
        .iter()
        .filter(|p| p.store_id != current_store_id && p.role != metapb::PeerRole::Learner);
    let mut join_set = JoinSet::new();
    for follower in followers {
        let stores_info = ctx.stores_info.clone();
        let store_id = follower.store_id;
        join_set.spawn_on(
            async move {
                stores_info
                    .get_store_stats(store_id)
                    .await
                    .map(|stats| (store_id, stats))
            },
            &ctx.thread_pool,
        );
    }

    let low_space_threshold = ctx.cfg_controller.get_current().storage.low_space_threshold;

    let mut disk_full_stores = vec![];
    let mut disk_full_stores_available_size = vec![];
    while let Some(res) = join_set.join_next().await {
        match res {
            Ok(Ok((store_id, stats))) => {
                #[allow(unused_mut)]
                let mut stats = stats;
                #[cfg(feature = "failpoints")]
                fail::eval("mock_zero_follower_store_stats", |_| {
                    stats.capacity = 0;
                    stats.available = 0;
                });
                if stats.capacity == 0 {
                    // A store can briefly return default zero stats after PD leader
                    // switch or before a fresh store heartbeat repopulates PD.
                    // Return not leader to make caller retry.
                    return Err(make_errpb_not_leader(
                        region_id,
                        None,
                        format!("store {} stats not ready", store_id),
                    ));
                }
                if is_store_low_space(stats, low_space_threshold) {
                    disk_full_stores.push(store_id);
                    disk_full_stores_available_size.push(ReadableSize(stats.available).as_mb_f64());
                }
            }
            Ok(Err(pd_err)) => {
                // Return not leader to make caller retry.
                return Err(make_errpb_not_leader(
                    region_id,
                    None,
                    format!("failed to get store stats: {:?}", pd_err),
                ));
            }
            Err(e) => {
                // Return not leader to make caller retry.
                return Err(make_errpb_not_leader(
                    region_id,
                    None,
                    format!("failed to get store stats: {}", e),
                ));
            }
        }
    }

    if disk_full_stores.is_empty() {
        Ok(())
    } else {
        Err(make_errpb_disk_full(
            disk_full_stores,
            format!(
                "available space is low: {:?}",
                disk_full_stores_available_size
            ),
        ))
    }
}

fn is_store_low_space(stats: StoreStatsLite, low_space_threshold: AbsoluteOrPercentSize) -> bool {
    fail::fail_point!("is_store_low_space", |_| true);

    stats.available <= low_space_threshold.absolute(stats.capacity)
}

#[derive(Default, Clone, Copy, Debug)]
struct StoreStatsLite {
    update_time_secs: u64,
    capacity: u64,
    available: u64,
}

type StoresStatsCache =
    papaya::HashMap<u64 /* self */, Arc<tokio::sync::Mutex<StoreStatsLite>>>;

struct StoresInfo {
    pd_client: Arc<dyn PdClient>,
    cache: StoresStatsCache,
}

impl StoresInfo {
    fn new(pd_client: Arc<dyn PdClient>) -> Self {
        Self {
            pd_client,
            cache: StoresStatsCache::default(),
        }
    }

    async fn get_store_stats(&self, store_id: u64) -> pd_client::Result<StoreStatsLite> {
        let stats_lite = self
            .cache
            .pin()
            .get_or_insert_with(store_id, Default::default)
            .clone();
        let mut stats_lite = stats_lite.lock_owned().await;

        let now = UnixSecs::now().into_inner();
        if stats_lite.update_time_secs + STORES_INFO_CACHE_TTL.as_secs() > now {
            return Ok(*stats_lite);
        }

        let stats = match self.pd_client.get_store_stats_async(store_id).await {
            Ok(stats) => stats,
            Err(e) => {
                if stats_lite.update_time_secs + STORES_INFO_CACHE_MAX_TTL.as_secs() > now {
                    warn!("get_store_stats failed, use stale cache"; "store" => store_id, "err" => ?e);
                    return Ok(*stats_lite);
                } else {
                    error!("get_store_stats failed"; "store" => store_id, "err" => ?e);
                    return Err(e);
                }
            }
        };

        stats_lite.update_time_secs = now;
        stats_lite.capacity = stats.capacity;
        stats_lite.available = stats.available;
        info!("get_store_stats"; "store" => store_id, "stats" => ?stats_lite);
        Ok(*stats_lite)
    }
}

#[derive(Debug)]
struct RestoreShardRequest {
    pub cs: kvenginepb::ChangeSet,
    pub expected_conf_ver: Option<u64>,
}

#[derive(Default, Serialize, Deserialize, Debug, PartialEq)]
#[serde(default)]
#[serde(rename_all = "kebab-case")]
pub struct RestoreShardResponse {
    pub shard_id: u64,
    pub restore_bytes: u64,
}

#[derive(Default, Serialize, Deserialize, Debug)]
#[serde(default)]
#[serde(rename_all = "kebab-case")]
pub struct TruncateTsConfig {
    pub cluster_id: u64,
    pub ts: u64,
    pub range: Option<(Vec<u8>, Vec<u8>)>, // None means apply to all keyspaces.
}

#[derive(Default, Serialize, Deserialize, Debug)]
#[serde(default)]
#[serde(rename_all = "kebab-case")]
pub struct TrackWalProgressRequest {
    pub ts: u64,
}

fn estimate_backup_size_by(
    engine: &kvengine::Engine,
    result: &mut std::collections::HashMap<u32, rfenginepb::BackupSize>,
) {
    for shard in engine.shards() {
        let size = result.entry(shard.keyspace_id).or_default();
        size.size += shard.get_estimated_size();
    }
}

fn tag_from_cs(engine: &kvengine::Engine, cs: &kvenginepb::ChangeSet) -> ShardTag {
    ShardTag::new(
        engine.get_engine_id(),
        IdVer::new(cs.shard_id, cs.shard_ver),
    )
}

fn make_errpb_not_leader(
    region_id: u64,
    leader: Option<metapb::Peer>,
    msg: String,
) -> errorpb::Error {
    let mut errpb = errorpb::Error::default();
    errpb.set_message(msg);
    errpb.mut_not_leader().set_region_id(region_id);
    if let Some(leader) = leader {
        errpb.mut_not_leader().set_leader(leader);
    }
    errpb
}

fn make_errpb_region_not_found(region_id: u64) -> errorpb::Error {
    let mut errpb = errorpb::Error::default();
    errpb.mut_region_not_found().set_region_id(region_id);
    errpb
}

fn make_errpb_epoch_not_match(region: metapb::Region) -> errorpb::Error {
    let mut errpb = errorpb::Error::default();
    errpb
        .mut_epoch_not_match()
        .set_current_regions(vec![region].into());
    errpb
}

fn make_errpb_disk_full(stores_id: Vec<u64>, reason: String) -> errorpb::Error {
    let disk_full = errorpb::DiskFull {
        store_id: stores_id,
        reason,
        ..Default::default()
    };
    let mut errpb = errorpb::Error::default();
    errpb.set_disk_full(disk_full);
    errpb
}

fn get_first_path_segment(path: &str) -> &str {
    path.trim_start_matches('/').split('/').next().unwrap_or("")
}

#[cfg(test)]
mod tests {
    use std::{
        convert::Infallible,
        net::TcpListener,
        sync::{
            Arc, Barrier,
            atomic::{AtomicUsize, Ordering},
            mpsc,
        },
        thread,
        time::Duration,
    };

    use bytes::Bytes;
    use hyper::{
        Body, Client, Request, Server,
        body::HttpBody,
        header::{CONTENT_LENGTH, CONTENT_TYPE, TRANSFER_ENCODING},
        service::{make_service_fn, service_fn},
    };
    use tikv_util::memory::MemoryLimiter;
    use tokio::sync::Semaphore;

    use super::{
        DfsFileReadAdmissionGuard, DfsFileReadRangeError, DfsFileReadRequestGuard,
        ONLINE_SNAPSHOT_MAX_RESPONSE_BYTES, ONLINE_SNAPSHOT_MAX_TTL,
        ONLINE_SNAPSHOT_MEMORY_PERMITS, ONLINE_SNAPSHOT_MEMORY_UNIT_BYTES,
        get_bounded_duration_param, get_first_path_segment, get_optional_uint_param,
        online_snapshot_dump_concurrency, online_snapshot_memory_permits,
    };
    use crate::StatusServer;

    fn admit(
        request_in_flight: &Arc<AtomicUsize>,
        memory_limiter: &MemoryLimiter,
        requested_bytes: u64,
    ) -> DfsFileReadAdmissionGuard {
        let request = DfsFileReadRequestGuard::try_acquire(request_in_flight, 1).unwrap();
        let bytes = memory_limiter.acquire(requested_bytes).unwrap();
        DfsFileReadAdmissionGuard::new(request, bytes, requested_bytes)
    }

    #[test]
    fn test_get_first_path_segment() {
        let cases = [
            ("/", ""),
            ("/kvengine/snapshot/123", "kvengine"),
            ("/dfs/path/to/file", "dfs"),
            ("/region", "region"),
            ("/region/", "region"),
            ("kvengine/snapshot", "kvengine"),
        ];

        for (input, expected) in cases {
            assert_eq!(get_first_path_segment(input), expected);
        }
    }

    #[test]
    fn test_get_optional_uint_param_rejects_malformed_value() {
        let query_pairs = url::form_urlencoded::parse(b"expected_conf_ver=abc").collect();
        let err = get_optional_uint_param(&query_pairs, "expected_conf_ver").unwrap_err();
        assert_eq!(
            err,
            "invalid expected_conf_ver: expected an unsigned integer"
        );
    }

    #[test]
    fn test_dfs_file_read_range_is_charged_by_requested_bytes() {
        assert_eq!(StatusServer::calculate_dfs_file_read_bytes(0, 100), Ok(100));
        assert_eq!(StatusServer::calculate_dfs_file_read_bytes(10, 40), Ok(30));
        assert_eq!(
            StatusServer::calculate_dfs_file_read_bytes(40, 10),
            Err(DfsFileReadRangeError::StartAfterEnd)
        );
    }

    #[test]
    fn test_dfs_file_read_request_limit_can_shrink_and_grow() {
        let in_flight = Arc::new(AtomicUsize::new(0));

        let guard1 = DfsFileReadRequestGuard::try_acquire(&in_flight, 2).unwrap();
        let guard2 = DfsFileReadRequestGuard::try_acquire(&in_flight, 2).unwrap();
        assert_eq!(
            DfsFileReadRequestGuard::try_acquire(&in_flight, 2).unwrap_err(),
            2
        );

        // Shrinking never interrupts active responses, but blocks new ones
        // until the in-flight count falls below the latest limit.
        assert_eq!(
            DfsFileReadRequestGuard::try_acquire(&in_flight, 1).unwrap_err(),
            2
        );
        drop(guard1);
        assert_eq!(
            DfsFileReadRequestGuard::try_acquire(&in_flight, 1).unwrap_err(),
            1
        );

        // Growing takes effect on the next admission attempt.
        let guard3 = DfsFileReadRequestGuard::try_acquire(&in_flight, 2).unwrap();
        assert_eq!(in_flight.load(Ordering::Acquire), 2);
        drop((guard2, guard3));
        assert_eq!(in_flight.load(Ordering::Acquire), 0);
    }

    #[test]
    fn test_dfs_file_read_request_limit_under_concurrency() {
        const LIMIT: usize = 4;
        const REQUESTS: usize = 32;

        let in_flight = Arc::new(AtomicUsize::new(0));
        let admitted = Arc::new(AtomicUsize::new(0));
        let barrier = Arc::new(Barrier::new(REQUESTS));
        thread::scope(|scope| {
            for _ in 0..REQUESTS {
                let in_flight = in_flight.clone();
                let admitted = admitted.clone();
                let barrier = barrier.clone();
                scope.spawn(move || {
                    let guard = DfsFileReadRequestGuard::try_acquire(&in_flight, LIMIT).ok();
                    if guard.is_some() {
                        admitted.fetch_add(1, Ordering::AcqRel);
                    }
                    barrier.wait();
                });
            }
        });

        assert_eq!(admitted.load(Ordering::Acquire), LIMIT);
        assert_eq!(in_flight.load(Ordering::Acquire), 0);
    }

    #[test]
    fn test_online_snapshot_dump_concurrency_tracks_cpu_quota() {
        assert_eq!(online_snapshot_dump_concurrency(0.5), 1);
        assert_eq!(online_snapshot_dump_concurrency(1.0), 1);
        assert_eq!(online_snapshot_dump_concurrency(2.0), 1);
        assert_eq!(online_snapshot_dump_concurrency(4.0), 2);
        assert_eq!(online_snapshot_dump_concurrency(5.9), 2);
        assert_eq!(online_snapshot_dump_concurrency(6.0), 3);
    }

    #[test]
    fn test_online_snapshot_memory_permits_charge_twice_memtable_kv_size() {
        assert_eq!(online_snapshot_memory_permits(0), 1);
        assert_eq!(
            online_snapshot_memory_permits(ONLINE_SNAPSHOT_MEMORY_UNIT_BYTES / 2),
            1
        );
        assert_eq!(
            online_snapshot_memory_permits(ONLINE_SNAPSHOT_MEMORY_UNIT_BYTES / 2 + 1),
            2
        );
        assert_eq!(
            online_snapshot_memory_permits(ONLINE_SNAPSHOT_MAX_RESPONSE_BYTES),
            ONLINE_SNAPSHOT_MEMORY_PERMITS as u32
        );
        assert_eq!(
            online_snapshot_memory_permits(usize::MAX),
            ONLINE_SNAPSHOT_MEMORY_PERMITS as u32
        );
    }

    #[test]
    fn test_online_snapshot_control_duration_is_bounded() {
        let query_pairs = url::form_urlencoded::parse(b"").collect();
        assert_eq!(
            get_bounded_duration_param(
                &query_pairs,
                "ttl_secs",
                Duration::from_secs(60),
                ONLINE_SNAPSHOT_MAX_TTL,
            )
            .unwrap(),
            Duration::from_secs(60)
        );

        let max_seconds = ONLINE_SNAPSHOT_MAX_TTL.as_secs();
        let query = format!("ttl_secs={max_seconds}");
        let query_pairs = url::form_urlencoded::parse(query.as_bytes()).collect();
        assert_eq!(
            get_bounded_duration_param(
                &query_pairs,
                "ttl_secs",
                Duration::from_secs(60),
                ONLINE_SNAPSHOT_MAX_TTL,
            )
            .unwrap(),
            ONLINE_SNAPSHOT_MAX_TTL
        );

        for value in [
            "0".to_owned(),
            (max_seconds + 1).to_string(),
            "invalid".to_owned(),
        ] {
            let query = format!("ttl_secs={value}");
            let query_pairs = url::form_urlencoded::parse(query.as_bytes()).collect();
            get_bounded_duration_param(
                &query_pairs,
                "ttl_secs",
                Duration::from_secs(60),
                ONLINE_SNAPSHOT_MAX_TTL,
            )
            .unwrap_err();
        }
    }

    #[test]
    fn test_parse_online_snapshot_shard_request() {
        let request =
            Request::get("/kvengine/online-snapshot/capture/42?shard_ver=7&branch_ts=100")
                .body(Body::empty())
                .unwrap();
        let parsed = StatusServer::parse_online_snapshot_shard_request(&request).unwrap();
        assert_eq!(parsed.shard_id, 42);
        assert_eq!(parsed.shard_ver, 7);
        assert_eq!(parsed.branch_ts, Some(100));

        let request = Request::get("/kvengine/online-snapshot/file-refs/42")
            .body(Body::empty())
            .unwrap();
        assert!(StatusServer::parse_online_snapshot_shard_request(&request).is_err());

        let request = Request::get(format!(
            "/kvengine/online-snapshot/capture/42?shard_ver=7&branch_ts={}",
            u64::MAX
        ))
        .body(Body::empty())
        .unwrap();
        assert!(StatusServer::parse_online_snapshot_shard_request(&request).is_err());
    }

    #[tokio::test]
    async fn test_permit_guarded_response_holds_all_permits_until_body_drop() {
        let semaphore = Arc::new(Semaphore::new(4));
        let permit = semaphore.clone().acquire_many_owned(3).await.unwrap();
        let data = Bytes::from(vec![7; StatusServer::DFS_FILE_READ_CHUNK_SIZE + 1]);
        let response = StatusServer::make_permit_guarded_response(
            data,
            permit,
            "application/protobuf",
            StatusServer::DFS_FILE_READ_CHUNK_SIZE,
        );
        assert_eq!(semaphore.available_permits(), 1);

        let mut body = response.into_body();
        assert_eq!(
            body.data().await.unwrap().unwrap().len(),
            StatusServer::DFS_FILE_READ_CHUNK_SIZE
        );
        assert_eq!(semaphore.available_permits(), 1);

        drop(body);
        assert_eq!(semaphore.available_permits(), 4);
    }

    #[tokio::test]
    async fn test_dfs_file_read_response_streams_all_chunks() {
        let final_chunk_size = 37;
        let data_len = StatusServer::DFS_FILE_READ_CHUNK_SIZE * 2 + final_chunk_size;
        let expected = Bytes::from(
            (0..data_len)
                .map(|i| (i % u8::MAX as usize) as u8)
                .collect::<Vec<_>>(),
        );
        let admission = Arc::new(AtomicUsize::new(0));
        let memory_limiter = MemoryLimiter::new(data_len as u64, None);
        let guard = admit(&admission, &memory_limiter, data_len as u64);
        let response = StatusServer::make_dfs_file_read_response(expected.clone(), guard);

        assert_eq!(
            response.headers().get(CONTENT_LENGTH).unwrap(),
            expected.len().to_string().as_str()
        );
        assert_eq!(admission.load(Ordering::Acquire), 1);
        assert_eq!(memory_limiter.used(), data_len as u64);

        let mut body = response.into_body();
        let mut actual = Vec::with_capacity(expected.len());
        let mut chunk_sizes = Vec::new();
        let mut chunk_uniqueness = Vec::new();
        while let Some(chunk) = body.data().await {
            let chunk = chunk.unwrap();
            chunk_sizes.push(chunk.len());
            chunk_uniqueness.push(chunk.is_unique());
            actual.extend_from_slice(&chunk);
        }

        assert_eq!(
            chunk_sizes,
            vec![
                StatusServer::DFS_FILE_READ_CHUNK_SIZE,
                StatusServer::DFS_FILE_READ_CHUNK_SIZE,
                final_chunk_size,
            ]
        );
        assert_eq!(chunk_uniqueness, vec![false, false, true]);
        assert_eq!(actual, expected);
        assert_eq!(admission.load(Ordering::Acquire), 0);
        assert_eq!(memory_limiter.used(), 0);
    }

    #[tokio::test]
    async fn test_dfs_file_read_admission_outlives_detached_blocking_task() {
        let admission = Arc::new(AtomicUsize::new(0));
        let memory_limiter = MemoryLimiter::new(8, None);
        let request = DfsFileReadRequestGuard::try_acquire(&admission, 1).unwrap();
        let (started_tx, started_rx) = mpsc::channel();
        let (unblock_tx, unblock_rx) = mpsc::channel();
        let task_memory_limiter = memory_limiter.clone();

        let join_handle = StatusServer::spawn_blocking_with_dfs_read_guard(
            &tokio::runtime::Handle::current(),
            request,
            move |request| {
                let bytes = task_memory_limiter.acquire(8).unwrap();
                let _guard = DfsFileReadAdmissionGuard::new(request, bytes, 8);
                started_tx.send(()).unwrap();
                unblock_rx.recv().unwrap();
            },
        );
        started_rx.recv_timeout(Duration::from_secs(5)).unwrap();

        // Dropping the request-side handle detaches the blocking task, but the
        // task must retain both admission limits until the read itself finishes.
        drop(join_handle);
        DfsFileReadRequestGuard::try_acquire(&admission, 1).unwrap_err();
        assert_eq!(memory_limiter.used(), 8);

        unblock_tx.send(()).unwrap();
        let request = tokio::time::timeout(Duration::from_secs(5), async {
            loop {
                if let Ok(request) = DfsFileReadRequestGuard::try_acquire(&admission, 1) {
                    return request;
                }
                tokio::task::yield_now().await;
            }
        })
        .await
        .unwrap();
        assert_eq!(memory_limiter.used(), 0);
        drop(request);
    }

    #[tokio::test]
    async fn test_dfs_file_read_response_releases_admission_on_completion_and_drop() {
        let admission = Arc::new(AtomicUsize::new(0));
        let memory_limiter =
            MemoryLimiter::new((StatusServer::DFS_FILE_READ_CHUNK_SIZE + 1) as u64, None);

        let guard = admit(&admission, &memory_limiter, 0);
        let response = StatusServer::make_dfs_file_read_response(Bytes::new(), guard);
        DfsFileReadRequestGuard::try_acquire(&admission, 1).unwrap_err();
        assert!(
            hyper::body::to_bytes(response.into_body())
                .await
                .unwrap()
                .is_empty()
        );
        assert_eq!(admission.load(Ordering::Acquire), 0);
        assert_eq!(memory_limiter.used(), 0);

        let data = Bytes::from(vec![42; StatusServer::DFS_FILE_READ_CHUNK_SIZE + 1]);
        let guard = admit(&admission, &memory_limiter, data.len() as u64);
        let response = StatusServer::make_dfs_file_read_response(data, guard);
        let mut body = response.into_body();
        assert_eq!(
            body.data().await.unwrap().unwrap().len(),
            StatusServer::DFS_FILE_READ_CHUNK_SIZE
        );
        DfsFileReadRequestGuard::try_acquire(&admission, 1).unwrap_err();
        assert_eq!(
            memory_limiter.used(),
            (StatusServer::DFS_FILE_READ_CHUNK_SIZE + 1) as u64
        );

        drop(body);
        assert_eq!(admission.load(Ordering::Acquire), 0);
        assert_eq!(memory_limiter.used(), 0);
    }

    #[tokio::test]
    async fn test_dfs_file_read_response_is_compatible_with_hyper_client() {
        let expected = Bytes::from(vec![7; StatusServer::DFS_FILE_READ_CHUNK_SIZE + 17]);
        let admission = Arc::new(AtomicUsize::new(0));
        let memory_limiter = MemoryLimiter::new(expected.len() as u64, None);
        let response_data = expected.clone();
        let response_admission = admission.clone();
        let response_memory_limiter = memory_limiter.clone();
        let make_service = make_service_fn(move |_| {
            let data = response_data.clone();
            let admission = response_admission.clone();
            let memory_limiter = response_memory_limiter.clone();
            async move {
                Ok::<_, Infallible>(service_fn(move |_| {
                    let data = data.clone();
                    let guard = admit(&admission, &memory_limiter, data.len() as u64);
                    async move {
                        Ok::<_, Infallible>(StatusServer::make_dfs_file_read_response(data, guard))
                    }
                }))
            }
        });
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel();
        let server = Server::from_tcp(listener)
            .unwrap()
            .serve(make_service)
            .with_graceful_shutdown(async move {
                let _ = shutdown_rx.await;
            });
        let server_task = tokio::spawn(server);

        let client = Client::new();
        let response = client
            .get(format!("http://{addr}/dfs/1").parse().unwrap())
            .await
            .unwrap();
        assert_eq!(response.status(), hyper::StatusCode::OK);
        assert_eq!(
            response.headers().get(CONTENT_LENGTH).unwrap(),
            expected.len().to_string().as_str()
        );
        assert!(response.headers().get(TRANSFER_ENCODING).is_none());
        let actual = hyper::body::to_bytes(response.into_body()).await.unwrap();
        assert_eq!(actual, expected);
        assert_eq!(admission.load(Ordering::Acquire), 0);
        assert_eq!(memory_limiter.used(), 0);

        drop(client);
        let _ = shutdown_tx.send(());
        server_task.await.unwrap().unwrap();
    }

    #[test]
    fn test_collect_columnar_status_table_prefix_range_covers_row_keys() {
        let keyspace_id = 1u32;
        let table_id = 120i64;

        let mut raw_table_row_prefix = api_version::ApiV2::get_txn_keyspace_prefix(keyspace_id);
        raw_table_row_prefix.extend_from_slice(
            &tidb_query_datatype::codec::table::encode_row_key_prefix(table_id),
        );
        let raw_table_row_prefix_next = keys::next_key(&raw_table_row_prefix);

        let encoded_start = tikv_util::codec::bytes::encode_bytes(&raw_table_row_prefix);
        let encoded_end = tikv_util::codec::bytes::encode_bytes(&raw_table_row_prefix_next);

        let mut raw_row_key = api_version::ApiV2::get_txn_keyspace_prefix(keyspace_id);
        raw_row_key.extend_from_slice(&tidb_query_datatype::codec::table::encode_row_key(
            table_id, 42,
        ));
        let encoded_row_key = tikv_util::codec::bytes::encode_bytes(&raw_row_key);

        assert!(encoded_start <= encoded_row_key);
        assert!(encoded_row_key < encoded_end);

        // Regression: using `next_key` on `encode_bytes(prefix)` produces a too-small
        // end key when `prefix.len() % 8 != 0`, which would exclude almost all row
        // keys.
        let encoded_wrong_end = keys::next_key(&encoded_start);
        assert!(encoded_row_key >= encoded_wrong_end);
    }

    #[test]
    fn test_dump_heap_profile_resp_default_returns_symbolized_pprof() {
        let req = Request::builder()
            .uri("/debug/pprof/heap")
            .body(Body::empty())
            .unwrap();
        let resp = StatusServer::dump_heap_prof_to_resp(req).unwrap();

        assert_eq!(
            resp.headers().get(CONTENT_TYPE).unwrap(),
            "application/octet-stream"
        );
        assert_eq!(
            resp.headers().get("Content-Disposition").unwrap(),
            "attachment; filename=\"profile.pprof\""
        );
    }

    #[test]
    fn test_dump_heap_profile_resp_svg() {
        let req = Request::builder()
            .uri("/debug/pprof/heap?svg=true")
            .body(Body::empty())
            .unwrap();
        let resp = StatusServer::dump_heap_prof_to_resp(req).unwrap();

        assert_eq!(resp.headers().get(CONTENT_TYPE).unwrap(), "image/svg+xml");
        assert_eq!(
            resp.headers().get("Content-Disposition").unwrap(),
            "attachment; filename=\"profile.svg\""
        );
    }
}
#[cfg(test)]
mod encryption_meta_gen_tests;