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

use std::{
    fmt::{Debug, Formatter},
    io::Write,
    ops::{Deref, DerefMut},
    path::Path,
    sync::{
        Arc,
        atomic::{AtomicBool, AtomicUsize, Ordering},
    },
    time::Duration,
};

use async_trait::async_trait;
use bstr::ByteSlice;
use bytes::{BufMut, Bytes, BytesMut};
use engine_traits::{GetObjectOptions, ListObjectContent, ObjectCacheWithHook, ObjectStorage};
use fail::fail_point;
use futures::StreamExt;
use http::StatusCode;
use hyper_tls::HttpsConnector;
use rusoto_core::{
    HttpClient, HttpDispatchError, Region, RusotoError,
    param::{Params, ServiceParams},
    request::{BufferedHttpResponse, HttpResponse},
    signature::SignedRequest,
};
use rusoto_s3::{
    AbortMultipartUploadError, CompleteMultipartUploadError, CopyObjectError,
    CreateMultipartUploadError, DeleteObjectError, GetObjectError, GetObjectTaggingError,
    HeadObjectError, ListObjectsV2Error, PutObjectError, UploadPartError,
};
use tikv_util::{
    box_join_err,
    config::ReadableSize,
    errors::BoxError,
    sys::thread::ThreadBuildWrapper,
    time::{Instant, Limiter},
};
use tokio::runtime::Runtime;

use crate::dfs::{
    self, Dfs, Error, FileType, KVENGINE_DFS_LATENCY, KVENGINE_DFS_LATENCY_WITH_RETRY,
    KVENGINE_DFS_REQUEST_COUNTER, KVENGINE_DFS_RETRY_COUNTER, KVENGINE_DFS_THROTTLED,
    KVENGINE_DFS_THROUGHPUT, MetricsFileType, MultipartUploadPart, Options, ReservableWriter,
    config::{Config, ConnOptions},
    fingerprint64, get_file_type_from_key, retry_sleep_duration,
};

pub const STORAGE_CLASS_DEFAULT: &str = STORAGE_CLASS_INTELLIGENT_TIERING;
pub const STORAGE_CLASS_INTELLIGENT_TIERING: &str = "INTELLIGENT_TIERING";
pub const STORAGE_CLASS_STANDARD: &str = "STANDARD";
pub const STORAGE_CLASS_STANDARD_IA: &str = "STANDARD_IA";
pub const STORAGE_CLASS_GLACIER_IR: &str = "GLACIER_IR";

// In Amazon S3, the Standard storage class is referred to as STANDARD, the
// Infrequent Access (IA) storage class is referred to as STANDARD_IA, and the
// Archive storage class is referred to as GLACIER. You can convert the storage
// class of OSS objects based on your requirements.
// ref: https://www.alibabacloud.com/help/en/oss/developer-reference/compatibility-with-amazon-s3
// and https://www.alibabacloud.com/help/en/oss/user-guide/overview-53/
pub const OSS_STORAGE_CLASS_DEFAULT: &str = OSS_STORAGE_CLASS_STANDARD;
pub const OSS_STORAGE_CLASS_STANDARD: &str = "Standard";
pub const OSS_STORAGE_CLASS_IA: &str = "IA";
pub const OSS_STORAGE_CLASS_ARCHIVE: &str = "Archive";

// Tencent COS storage classes for multi-availability zone buckets.
// ref: https://www.tencentcloud.com/zh/document/product/436/30925
pub const COS_STORAGE_CLASS_DEFAULT: &str = COS_STORAGE_CLASS_STANDARD;
pub const COS_STORAGE_CLASS_STANDARD: &str = "MAZ_STANDARD";
pub const COS_STORAGE_CLASS_STANDARD_IA: &str = "MAZ_STANDARD_IA";
pub const COS_STORAGE_CLASS_ARCHIVE: &str = "MAZ_ARCHIVE";
pub const COS_STORAGE_CLASS_INTELLIGENT_TIERING: &str = "MAZ_INTELLIGENT_TIERING";

const AWS_DOMAIN_STRING: &str = "amazonaws.com";
const S3_CHECKSUM_ALGORITHM_CRC32: &str = "CRC32";
const S3_CHECKSUM_ALGORITHM_HEADER: &str = "x-amz-checksum-algorithm";
const S3_CHECKSUM_CRC32_HEADER: &str = "x-amz-checksum-crc32";

const SMALL_FILE_THRESHOLD_BYTES: u64 = 1024 * 1024; // 1MB

fn encode_checksum_crc32(checksum: u32) -> String {
    base64::encode(checksum.to_be_bytes())
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum StorageClass {
    Standard,
    InfrequentAccess,
    GlacierInstantRetrieval,
    IntelligentTiering,
    Default,
}

impl StorageClass {
    fn from_str(s: &str) -> Self {
        match s.to_ascii_uppercase().as_str() {
            "STANDARD" | "MAZ_STANDARD" => Self::Standard,
            "STANDARD_IA" | "IA" | "MAZ_STANDARD_IA" => Self::InfrequentAccess,
            "GLACIER_IR" | "ARCHIVE" | "MAZ_ARCHIVE" => Self::GlacierInstantRetrieval,
            "INTELLIGENT_TIERING" | "MAZ_INTELLIGENT_TIERING" => Self::IntelligentTiering,
            _ => Self::Default,
        }
    }
}

/// Supported S3-compatible or cloud object storage providers.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CloudProvider {
    Aws,
    Aliyun,
    Tencentcloud,
    Volcengine,
    Unknown,
}

impl CloudProvider {
    /// Try to infer the cloud provider based on the endpoint.
    pub fn from_endpoint(endpoint: &str) -> Self {
        let lower = endpoint.to_ascii_lowercase();
        if lower.contains(AWS_DOMAIN_STRING) {
            CloudProvider::Aws
        } else if lower.contains(aliyun::DOMAIN_STRING) {
            CloudProvider::Aliyun
        } else if lower.contains(tencentcloud::DOMAIN_STRING) {
            CloudProvider::Tencentcloud
        } else if lower.contains(volcengine::DOMAIN_STRING) {
            CloudProvider::Volcengine
        } else {
            CloudProvider::Unknown
        }
    }

    /// Returns the provider-specific header prefix.
    pub fn header_prefix(&self) -> &'static str {
        match self {
            CloudProvider::Aws | CloudProvider::Volcengine => "x-amz-",
            CloudProvider::Aliyun => "x-oss-",
            CloudProvider::Tencentcloud => "x-cos-",
            _ => "x-amz-",
        }
    }

    /// Builds a full header key name given a suffix.
    pub fn build_header_key(&self, key: &str) -> String {
        format!("{}{}", self.header_prefix(), key)
    }

    /// Set the corresponding cloud-specific headers.
    pub fn add_provider_header(&self, req: &mut SignedRequest, key: &str, value: &str) {
        let value = if key.ends_with("storage-class") {
            self.storage_class_str(StorageClass::from_str(value))
        } else {
            value
        };
        let key_suffix = key
            .strip_prefix(CloudProvider::Aws.header_prefix())
            .unwrap_or(key);
        req.add_header(self.build_header_key(key_suffix), value);
    }

    /// Convert to the provider-specific storage class string.
    /// Alibaba cloud only support "Standard", "IA" and "Archive",
    /// Unknown storage class will be treated as "Standard".
    /// ref: https://www.alibabacloud.com/help/en/oss/developer-reference/compatibility-with-amazon-s3
    pub fn storage_class_str(&self, storage_class: StorageClass) -> &'static str {
        match self {
            CloudProvider::Aliyun => match storage_class {
                StorageClass::Standard | StorageClass::IntelligentTiering => {
                    OSS_STORAGE_CLASS_STANDARD
                }
                StorageClass::InfrequentAccess => OSS_STORAGE_CLASS_IA,
                StorageClass::GlacierInstantRetrieval => OSS_STORAGE_CLASS_ARCHIVE,
                StorageClass::Default => OSS_STORAGE_CLASS_DEFAULT,
            },
            CloudProvider::Tencentcloud => match storage_class {
                StorageClass::Standard => COS_STORAGE_CLASS_STANDARD,
                StorageClass::InfrequentAccess => COS_STORAGE_CLASS_STANDARD_IA,
                StorageClass::GlacierInstantRetrieval => COS_STORAGE_CLASS_ARCHIVE,
                StorageClass::IntelligentTiering => COS_STORAGE_CLASS_INTELLIGENT_TIERING,
                StorageClass::Default => COS_STORAGE_CLASS_DEFAULT,
            },
            // Volcengine internal TOS only bills the standard class (its tiering is
            // a wrapper with no real cold/hot separation), so pin every class to it.
            CloudProvider::Volcengine => match storage_class {
                StorageClass::Standard => STORAGE_CLASS_STANDARD,
                StorageClass::InfrequentAccess => STORAGE_CLASS_STANDARD,
                StorageClass::GlacierInstantRetrieval => STORAGE_CLASS_STANDARD,
                StorageClass::IntelligentTiering => STORAGE_CLASS_STANDARD,
                StorageClass::Default => STORAGE_CLASS_STANDARD,
            },
            _ => match storage_class {
                StorageClass::Standard => STORAGE_CLASS_STANDARD,
                StorageClass::InfrequentAccess => STORAGE_CLASS_STANDARD_IA,
                StorageClass::GlacierInstantRetrieval => STORAGE_CLASS_GLACIER_IR,
                StorageClass::IntelligentTiering => STORAGE_CLASS_INTELLIGENT_TIERING,
                StorageClass::Default => STORAGE_CLASS_DEFAULT,
            },
        }
    }
}

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

impl S3Fs {
    pub fn new(
        prefix: String,
        endpoint: String,
        key_id: String,
        secret_key: String,
        region: String,
        bucket: String,
        options: ConnOptions,
        read_only: bool,
    ) -> Self {
        let core = Arc::new(S3FsCore::new(
            endpoint,
            key_id,
            secret_key,
            region,
            bucket,
            prefix,
            options,
            read_only,
            ReadableSize(0),
        ));
        Self { core }
    }

    pub fn new_from_config(conf: Config) -> Self {
        let core = Arc::new(S3FsCore::new(
            conf.s3_endpoint,
            conf.s3_key_id,
            conf.s3_secret_key,
            conf.s3_region,
            conf.s3_bucket,
            conf.prefix,
            conf.conn_options,
            conf.read_only,
            conf.max_read_throughput,
        ));
        Self { core }
    }

    #[cfg(any(test, feature = "testexport"))]
    pub fn new_for_test(s3c: rusoto_core::Client, bucket: String, prefix: String) -> Self {
        Self {
            core: Arc::new(S3FsCore::new_with_s3_client(
                s3c,
                "".to_string(),
                "local".to_string(),
                bucket,
                prefix,
                ConnOptions::default(),
                false,
            )),
        }
    }

    #[cfg(any(test, feature = "testexport"))]
    pub fn new_for_test_with_opts(
        s3c: rusoto_core::Client,
        bucket: String,
        prefix: String,
        opts: ConnOptions,
    ) -> Self {
        Self {
            core: Arc::new(S3FsCore::new_with_s3_client(
                s3c,
                "".to_string(),
                "local".to_string(),
                bucket,
                prefix,
                opts,
                false,
            )),
        }
    }

    /// Set whether to move removed files to STANDARD_IA storage class.
    /// If false, removed files will keep their current storage class.
    pub fn set_removed_files_to_ia(&self, removed_files_to_ia: bool) {
        self.core
            .removed_files_to_ia
            .store(removed_files_to_ia, Ordering::Relaxed);
    }
}

impl Deref for S3Fs {
    type Target = S3FsCore;

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

pub struct S3FsCore {
    s3c: rusoto_core::Client,
    hostname: String,
    region: Region,
    bucket: String,
    prefix: String,
    runtime: Option<tokio::runtime::Runtime>,
    provider: CloudProvider,
    virtual_host: bool,
    // `rusoto` uses https by default, but cse uses http as default protocol.
    // This marker indicates the original endpoint has no protocol hence http should be used.
    use_http: bool,
    opts: ConnOptions,
    read_only: bool, // For safety when used for recovery.
    read_speed_limiter: Option<Limiter>,
    verifier: OperationVerifier,
    // Whether to move removed files to STANDARD_IA storage class.
    // If false, removed files will keep their current storage class.
    removed_files_to_ia: AtomicBool,
}

impl S3FsCore {
    pub fn new(
        endpoint: String,
        key_id: String,
        secret_key: String,
        region: String,
        bucket: String,
        prefix: String,
        options: ConnOptions,
        read_only: bool,
        max_read_throughput: ReadableSize,
    ) -> Self {
        let mut config = rusoto_core::HttpConfig::new();
        config.read_buf_size(256 * 1024);
        config.pool_idle_timeout(Some(options.pool_idle_timeout.0));
        let endpoint = if endpoint.is_empty() {
            format!("http://s3.{}.amazonaws.com", region.as_str())
        } else {
            endpoint
        };
        let use_tls = endpoint.starts_with("https");
        let default_provider = aws::new_credentials_provider_rusoto_wrapper();
        let static_provider =
            rusoto_credential::StaticProvider::new(key_id.clone(), secret_key, None, None);
        let mut http_connector = hyper::client::connect::HttpConnector::new();
        http_connector.set_connect_timeout(Some(options.conn_timeout.0));
        http_connector.set_keepalive(Some(options.keep_alive_duration.0));
        let s3c = if use_tls {
            http_connector.enforce_http(false);
            let https_connector = HttpsConnector::new_with_connector(http_connector);
            let http_client = HttpClient::from_connector_with_config(https_connector, config);
            if key_id.is_empty() {
                if endpoint.contains(aliyun::DOMAIN_STRING) {
                    let ali_provider = aliyun::new_credential_provider().unwrap();
                    rusoto_core::Client::new_with(ali_provider, http_client)
                } else if endpoint.contains(tencentcloud::DOMAIN_STRING) {
                    let tc_provider = tencentcloud::new_credential_provider().unwrap();
                    rusoto_core::Client::new_with(tc_provider, http_client)
                } else if endpoint.contains(volcengine::DOMAIN_STRING) {
                    let volc_provider = volcengine::new_credential_provider().unwrap();
                    rusoto_core::Client::new_with(volc_provider, http_client)
                } else {
                    rusoto_core::Client::new_with(default_provider, http_client)
                }
            } else {
                rusoto_core::Client::new_with(static_provider, http_client)
            }
        } else {
            let http_client = HttpClient::from_connector_with_config(http_connector, config);
            if key_id.is_empty() {
                if endpoint.contains(aliyun::DOMAIN_STRING) {
                    let ali_provider = aliyun::new_credential_provider().unwrap();
                    rusoto_core::Client::new_with(ali_provider, http_client)
                } else if endpoint.contains(tencentcloud::DOMAIN_STRING) {
                    let tc_provider = tencentcloud::new_credential_provider().unwrap();
                    rusoto_core::Client::new_with(tc_provider, http_client)
                } else if endpoint.contains(volcengine::DOMAIN_STRING) {
                    let volc_provider = volcengine::new_credential_provider().unwrap();
                    rusoto_core::Client::new_with(volc_provider, http_client)
                } else {
                    rusoto_core::Client::new_with(default_provider, http_client)
                }
            } else {
                rusoto_core::Client::new_with(static_provider, http_client)
            }
        };
        let mut core =
            Self::new_with_s3_client(s3c, endpoint, region, bucket, prefix, options, read_only);
        core.read_speed_limiter = super::build_read_speed_limiter(max_read_throughput);
        core
    }

    pub fn new_with_s3_client(
        s3c: rusoto_core::Client,
        endpoint: String,
        region: String,
        bucket: String,
        mut prefix: String,
        options: ConnOptions,
        read_only: bool,
    ) -> Self {
        let runtime = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(2)
            .enable_all()
            .thread_name("s3-client")
            .with_sys_hooks()
            .build()
            .unwrap();
        if prefix.is_empty() {
            prefix.push_str("default")
        }
        let use_http = !endpoint.starts_with("https://");
        let no_schema_endpoint = endpoint
            .find("://")
            .map(|p| &endpoint[p + 3..])
            .unwrap_or(&endpoint);
        let provider = CloudProvider::from_endpoint(&endpoint);
        // local deployed s3 service like minio does not support virtual host
        // addressing, it always has a port defined at the end.
        let virtual_host = !no_schema_endpoint.contains(':');
        let hostname = if virtual_host {
            format!("{}.{}", &bucket, no_schema_endpoint)
        } else {
            no_schema_endpoint.to_string()
        };
        let region = Region::Custom {
            name: region,
            endpoint,
        };
        Self {
            s3c,
            hostname,
            region,
            bucket,
            prefix,
            runtime: Some(runtime),
            provider,
            virtual_host,
            use_http,
            opts: options,
            read_only,
            read_speed_limiter: None,
            verifier: OperationVerifier::new(1000),
            removed_files_to_ia: AtomicBool::new(false),
        }
    }

    pub fn rel_file_key(&self, file_id: u64, file_type: FileType) -> String {
        let idx = (fingerprint64(file_id.to_le_bytes().as_slice())) as u8;
        match file_type {
            FileType::Sst => {
                format!("{:02x}/{:016x}.sst", idx, file_id)
            }
            FileType::Blob => {
                format!("blob/{:02x}/{:016x}.blob", idx, file_id)
            }
            FileType::TxnChunk => {
                format!("txn/{:02x}/{:016x}.txn", idx, file_id)
            }
            FileType::Schema => {
                format!("schema/{:02x}/{:016x}.schema", idx, file_id)
            }
            FileType::Columnar => {
                format!("col/{:02x}/{:016x}.col", idx, file_id)
            }
            FileType::VectorIndex => {
                format!("vec/{:02x}/{:016x}.vec", idx, file_id)
            }
            FileType::FtsPackedFile => {
                format!("fts/{:02x}/{:016x}.ftspack", idx, file_id)
            }
            FileType::FtsDedicatedFile => {
                format!("fts/{:02x}/{:016x}.ftsded", idx, file_id)
            }
            FileType::EncryptionDict => {
                format!("encdict/{:02x}/{:016x}.encdict", idx, file_id)
            }
        }
    }
    pub fn file_key(&self, file_id: u64, file_type: FileType) -> String {
        let idx = (fingerprint64(file_id.to_le_bytes().as_slice())) as u8;
        match file_type {
            FileType::Sst => {
                format!("{}/{:02x}/{:016x}.sst", self.prefix, idx, file_id)
            }
            FileType::Blob => {
                format!("{}/blob/{:02x}/{:016x}.blob", self.prefix, idx, file_id)
            }
            FileType::TxnChunk => {
                format!("{}/txn/{:02x}/{:016x}.txn", self.prefix, idx, file_id)
            }
            FileType::Schema => {
                format!("{}/schema/{:02x}/{:016x}.schema", self.prefix, idx, file_id)
            }
            FileType::Columnar => {
                format!("{}/col/{:02x}/{:016x}.col", self.prefix, idx, file_id)
            }
            FileType::VectorIndex => {
                format!("{}/vec/{:02x}/{:016x}.vec", self.prefix, idx, file_id)
            }
            FileType::FtsPackedFile => {
                format!("{}/fts/{:02x}/{:016x}.ftspack", self.prefix, idx, file_id)
            }
            FileType::FtsDedicatedFile => {
                format!("{}/fts/{:02x}/{:016x}.ftsded", self.prefix, idx, file_id)
            }
            FileType::EncryptionDict => {
                format!(
                    "{}/encdict/{:02x}/{:016x}.encdict",
                    self.prefix, idx, file_id
                )
            }
        }
    }

    pub fn get_prefix(&self) -> String {
        self.prefix.clone()
    }

    pub fn get_bucket(&self) -> String {
        self.bucket.clone()
    }

    pub fn is_on_aws(&self) -> bool {
        self.hostname.contains(AWS_DOMAIN_STRING)
    }

    fn is_on_aliyun(&self) -> bool {
        self.provider == CloudProvider::Aliyun
    }

    fn is_on_tencentcloud(&self) -> bool {
        self.provider == CloudProvider::Tencentcloud
    }

    fn is_on_volcengine(&self) -> bool {
        self.provider == CloudProvider::Volcengine
    }

    pub fn storage_class_str(&self, storage_class: StorageClass) -> &'static str {
        self.provider.storage_class_str(storage_class)
    }

    /// Check if HTTP status code is retryable.
    ///
    /// Alibaba Cloud OSS retryable errors:
    /// - 429 Too Many Requests: QPS limit exceeded
    /// - 500 Internal Server Error: service busy or internal error
    /// - 503 Service Unavailable: bandwidth limit or QPS limit
    ///
    /// The 500 and 503 errors are covered by `is_server_error()`.
    ///
    /// Ref: <https://www.alibabacloud.com/help/zh/oss/support/429-error>
    fn is_retryable_status(status: StatusCode) -> bool {
        status.is_server_error() || status == StatusCode::TOO_MANY_REQUESTS
    }

    fn is_err_retryable<T>(&self, rustoto_err: &RusotoError<T>) -> bool {
        match rustoto_err {
            RusotoError::Service(_) => true,
            RusotoError::HttpDispatch(_) => true,
            RusotoError::InvalidDnsName(_) => false,
            RusotoError::Credentials(cred) => cred.message.contains("Timeout"),
            RusotoError::Validation(_) => false,
            RusotoError::ParseError(_) => false,
            RusotoError::Unknown(resp) => Self::is_retryable_status(resp.status),
            RusotoError::Blocking => false,
        }
    }

    fn is_multipart_upload_err_retryable<T>(&self, rustoto_err: &RusotoError<T>) -> bool {
        match rustoto_err {
            // Multipart service errors such as NoSuchUpload are deterministic
            // request-state errors. Retrying them only hides the real failure
            // behind a long exponential backoff.
            RusotoError::Service(_) => false,
            _ => self.is_err_retryable(rustoto_err),
        }
    }

    fn is_err_not_found<T>(&self, rustoto_err: &RusotoError<T>) -> bool {
        match rustoto_err {
            RusotoError::Unknown(resp) => resp.status == StatusCode::NOT_FOUND,
            _ => false,
        }
    }

    async fn sleep_for_retry(&self, retry_cnt: &mut u32, file_name: &str) -> bool {
        if *retry_cnt < self.opts.max_retry_count {
            *retry_cnt += 1;
            let retry_sleep = retry_sleep_duration(*retry_cnt, self.opts.retry_sleep_interval.0);
            tokio::time::sleep(retry_sleep).await;
            true
        } else {
            error!(
                "read file {}, reach max retry count {}",
                file_name, self.opts.max_retry_count
            );
            false
        }
    }

    /// list gets a list of file ids(full path) greater than `start_after`, with
    /// optional prefix `prefix`.
    ///
    /// The result contains:
    ///     A vector of file content with `key` & `last_modified` timestamp.
    ///     A boolean `has_more` indicate if there is more.
    ///     An optional `next_start_after` for next loop if `has_more` is true.
    ///
    /// Note:
    ///     `prefix` should NOT be contained in `start_after`.
    ///     The file ids in result are in full path, including `S3Fs.prefix` and
    /// `prefix`.
    ///
    /// Ref: https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListObjectsV2.html
    pub async fn list(
        &self,
        start_after: &str,
        prefix: Option<&str>,
        max_keys: Option<u32>,
    ) -> crate::dfs::Result<(
        Vec<ListObjectContent>,
        bool,           // has_more. Deprecated, use `next_start_after`
        Option<String>, // next_start_after
    )> {
        let prefix = format!("{}/{}", self.prefix.clone(), prefix.unwrap_or_default());
        let start_after = format!("{}{}", prefix, start_after);
        let mut retry_cnt = 0;
        let start_time_with_retry = Instant::now_coarse();
        loop {
            let start_time = Instant::now_coarse();
            let mut req = self.new_request("GET", "");
            let mut params = Params::new();
            params.put("list-type", "2");
            params.put("start-after", &start_after);
            params.put("prefix", &prefix);
            if let Some(max_keys) = max_keys {
                params.put("max-keys", max_keys.to_string());
            }
            req.set_params(params);
            let err = match self.dispatch(req, ListObjectsV2Error::from_response).await {
                Ok(mut response) => match self.read_body(&mut response, false).await {
                    Ok(body) => {
                        let body_str = body.to_str().unwrap();
                        let list: ListObjects = quick_xml::de::from_str(body_str).unwrap();
                        let next_start_after = list.is_truncated.then(|| {
                            list.contents.last().unwrap().key.as_str()[prefix.len()..].to_string()
                        });

                        let duration = start_time.saturating_elapsed();
                        KVENGINE_DFS_LATENCY
                            .list
                            .unknown
                            .observe(duration.as_secs_f64());
                        let duration_with_retry = start_time_with_retry.saturating_elapsed();
                        KVENGINE_DFS_LATENCY_WITH_RETRY
                            .list
                            .unknown
                            .observe(duration_with_retry.as_secs_f64());
                        KVENGINE_DFS_REQUEST_COUNTER.list.unknown.inc();

                        return Ok((list.contents, list.is_truncated, next_start_after));
                    }
                    Err(e) => e.into(),
                },
                Err(e) => e,
            };
            if self.is_err_not_found(&err) {
                return Ok((vec![], false, None));
            } else if self.is_err_retryable(&err) && retry_cnt < self.opts.max_retry_count {
                KVENGINE_DFS_RETRY_COUNTER.list.unknown.inc();
                retry_cnt += 1;
                let retry_sleep = retry_sleep_duration(retry_cnt, self.opts.retry_sleep_interval.0);
                tokio::time::sleep(retry_sleep).await;
                continue;
            }
            error!(
                "failed to list files start after {}, reach max retry count {}, err {:?}",
                start_after, self.opts.max_retry_count, err,
            );
            return Err(err.into());
        }
    }

    pub async fn list_folders(
        &self,
        prefix: &str,
        delimiter: Option<&str>,
    ) -> crate::dfs::Result<Vec<String>> {
        let prefix = format!("{}/{}", self.prefix.clone(), prefix);
        let delimiter = delimiter.unwrap_or("/");
        let mut retry_cnt = 0;
        let start_time_with_retry = Instant::now_coarse();
        loop {
            let start_time = Instant::now_coarse();
            let mut req = self.new_request("GET", "");
            let mut params = Params::new();
            params.put("list-type", "2");
            params.put("start-after", &prefix);
            params.put("prefix", &prefix);
            params.put("delimiter", delimiter);
            req.set_params(params);
            let err = match self.dispatch(req, ListObjectsV2Error::from_response).await {
                Ok(mut response) => match self.read_body(&mut response, false).await {
                    Ok(body) => {
                        let body_str = body.to_str().unwrap();
                        let list: ListObjects = quick_xml::de::from_str(body_str).unwrap();
                        let prefixes = list
                            .common_prefixes
                            .into_iter()
                            .map(|p| p.prefix)
                            .collect::<Vec<_>>();
                        let duration = start_time.saturating_elapsed();
                        KVENGINE_DFS_LATENCY
                            .list
                            .unknown
                            .observe(duration.as_secs_f64());
                        let duration_with_retry = start_time_with_retry.saturating_elapsed();
                        KVENGINE_DFS_LATENCY_WITH_RETRY
                            .list
                            .unknown
                            .observe(duration_with_retry.as_secs_f64());
                        KVENGINE_DFS_REQUEST_COUNTER.list.unknown.inc();
                        return Ok(prefixes);
                    }
                    Err(e) => e.into(),
                },
                Err(e) => e,
            };
            if self.is_err_not_found(&err) {
                return Ok(vec![]);
            } else if self.is_err_retryable(&err) && retry_cnt < self.opts.max_retry_count {
                KVENGINE_DFS_RETRY_COUNTER.list.unknown.inc();
                retry_cnt += 1;
                let retry_sleep = retry_sleep_duration(retry_cnt, self.opts.retry_sleep_interval.0);
                tokio::time::sleep(retry_sleep).await;
                continue;
            }
            error!(
                "failed to list folders prefix {}, reach max retry count {}, err {:?}",
                prefix, self.opts.max_retry_count, err,
            );
            return Err(err.into());
        }
    }

    pub async fn is_removed(&self, file_key: &str) -> Result<bool, dfs::Error> {
        let mut retry_cnt = 0;
        let file_type = get_file_type_from_key(file_key);
        let start_time_with_retry = Instant::now();
        loop {
            let start_time = Instant::now();
            let req = self.new_tagging_request("GET", file_key);
            let err = match self
                .dispatch(req, GetObjectTaggingError::from_response)
                .await
            {
                Ok(mut resp) => match self.read_body(&mut resp, false).await {
                    Ok(body) => {
                        let body_str = body.to_str().unwrap();
                        let tagging: Tagging = quick_xml::de::from_str(body_str).unwrap();

                        let duration = start_time.saturating_elapsed();
                        KVENGINE_DFS_LATENCY
                            .get_tagging
                            .get(MetricsFileType::from(file_type))
                            .observe(duration.as_secs_f64());
                        let duration_with_retry = start_time_with_retry.saturating_elapsed();
                        KVENGINE_DFS_LATENCY_WITH_RETRY
                            .get_tagging
                            .get(MetricsFileType::from(file_type))
                            .observe(duration_with_retry.as_secs_f64());
                        KVENGINE_DFS_REQUEST_COUNTER
                            .get_tagging
                            .get(MetricsFileType::from(file_type))
                            .inc();
                        return Ok(tagging.has_deleted_tag());
                    }
                    Err(e) => e.into(),
                },
                Err(e) => e,
            };
            if let RusotoError::Service(_) = err {
                return Err(dfs::Error::S3(format!(
                    "Get file {} tagging fail {:?}",
                    file_key, err
                )));
            }
            if self.is_err_retryable(&err) && retry_cnt < self.opts.max_retry_count {
                KVENGINE_DFS_RETRY_COUNTER
                    .get_tagging
                    .get(MetricsFileType::from(file_type))
                    .inc();
                retry_cnt += 1;
                let retry_sleep = retry_sleep_duration(retry_cnt, self.opts.retry_sleep_interval.0);
                warn!(
                    "Get file {} tagging fail {:?}, retry_cnt {}, retry after {:?}",
                    file_key, err, retry_cnt, retry_sleep
                );
                tokio::time::sleep(retry_sleep).await;
                continue;
            }
            return Err(dfs::Error::S3(format!(
                "Get file {} tagging, reach max retry count {}, err {:?}",
                file_key, self.opts.max_retry_count, err,
            )));
        }
    }

    pub async fn exist(&self, key: String, file_name: String) -> Result<bool, dfs::Error> {
        match self.head_object(key, file_name).await {
            Ok(_) => Ok(true),
            Err(dfs::Error::NoSuchKey(_)) => Ok(false),
            Err(err) => Err(err),
        }
    }

    pub async fn object_size(&self, key: String, file_name: String) -> Result<u64, dfs::Error> {
        match self.head_object(key.clone(), file_name).await {
            Ok(headers) => {
                if let Some(content_length) = headers.get("Content-Length") {
                    if let Ok(length) = content_length.parse::<u64>() {
                        return Ok(length);
                    }
                    Err(dfs::Error::Other(format!(
                        "failed to parse {} Content-Length {}",
                        key, content_length
                    )))
                } else {
                    Err(dfs::Error::Other(format!(
                        "failed to get Content-Length from {}",
                        key
                    )))
                }
            }
            Err(err) => Err(err),
        }
    }

    pub async fn head_object(
        &self,
        key: String,
        file_name: String,
    ) -> Result<http::HeaderMap<String>, dfs::Error> {
        let mut retry_cnt = 0;
        let metrics_file_type = MetricsFileType::from(get_file_type_from_key(&key));
        let start_time_with_retry = Instant::now_coarse();
        loop {
            let start_time = Instant::now_coarse();
            let req = self.new_request("HEAD", &key);
            let result = self.dispatch(req, HeadObjectError::from_response).await;
            match result {
                Ok(resp) => {
                    let duration = start_time.saturating_elapsed();
                    KVENGINE_DFS_LATENCY
                        .head
                        .get(metrics_file_type)
                        .observe(duration.as_secs_f64());
                    let duration_with_retry = start_time_with_retry.saturating_elapsed();
                    KVENGINE_DFS_LATENCY_WITH_RETRY
                        .head
                        .get(metrics_file_type)
                        .observe(duration_with_retry.as_secs_f64());
                    KVENGINE_DFS_REQUEST_COUNTER
                        .head
                        .get(metrics_file_type)
                        .inc();
                    info!(
                        "head object {}, takes {:?}, retry {}",
                        &file_name, duration_with_retry, retry_cnt
                    );
                    return Ok(resp.headers.clone());
                }
                Err(err) => {
                    if let RusotoError::Service(HeadObjectError::NoSuchKey(err_msg)) = &err {
                        warn!("file {} not exist, err msg {}", &file_name, err_msg);
                        return Err(dfs::Error::NoSuchKey(err_msg.to_string()));
                    }
                    if self.is_err_not_found(&err) {
                        warn!("file {} not exist, err {}", &file_name, err.to_string());
                        return Err(dfs::Error::NoSuchKey(err.to_string()));
                    }
                    if self.is_err_retryable(&err)
                        && self.sleep_for_retry(&mut retry_cnt, &file_name).await
                    {
                        KVENGINE_DFS_RETRY_COUNTER.head.get(metrics_file_type).inc();
                        warn!("retry head file {}, error {:?}", &file_name, &err);
                        continue;
                    }
                    return Err(dfs::Error::S3(err.to_string()));
                }
            }
        }
    }

    fn new_request(&self, method: &str, key: &str) -> SignedRequest {
        let path = if self.virtual_host {
            format!("/{}", key)
        } else {
            format!("/{}/{}", &self.bucket, key)
        };
        let mut req = SignedRequest::new(method, "s3", &self.region, &path);
        if self.use_http {
            req.scheme = Some("http".to_string());
        }
        req
    }

    fn new_tagging_request(&self, method: &str, key: &str) -> SignedRequest {
        let mut req = self.new_request(method, key);
        let mut params = Params::new();
        params.put_key("tagging");
        req.set_params(params);
        req
    }

    async fn dispatch<E>(
        &self,
        req: SignedRequest,
        from_response: fn(BufferedHttpResponse) -> RusotoError<E>,
    ) -> Result<Response, RusotoError<E>> {
        self.dispatch_with_timeout(req, self.opts.dispatch_timeout.0, from_response)
            .await
    }

    async fn dispatch_with_timeout<E>(
        &self,
        mut req: SignedRequest,
        timeout: Duration,
        from_response: fn(BufferedHttpResponse) -> RusotoError<E>,
    ) -> Result<Response, RusotoError<E>> {
        req.set_hostname(Some(self.hostname.clone()));
        let mut resp = self.s3c.sign_and_dispatch_timeout(req, timeout).await?;
        if !resp.status.is_success() {
            let buffered = resp.buffer().await.map_err(RusotoError::HttpDispatch)?;
            return Err(from_response(buffered));
        }
        Ok(Response { resp })
    }

    async fn read_body(
        &self,
        resp: &mut Response,
        need_throttle: bool,
    ) -> Result<Bytes, HttpDispatchError> {
        let (writer, ..) = self
            .read_body_to_writer(resp, &|| Ok(BytesMut::new().writer()), need_throttle)
            .await?;
        Ok(writer.into_inner().freeze())
    }

    async fn read_body_to_writer<W, F>(
        &self,
        resp: &mut Response,
        build_writer: &F,
        need_throttle: bool,
    ) -> Result<(W, u64 /* read_len */, Duration /* throttled */), HttpDispatchError>
    where
        W: ReservableWriter + Send + 'static,
        F: Fn() -> std::io::Result<W> + 'static,
    {
        let mut writer =
            build_writer().map_err(|e| HttpDispatchError::new(format!("build_writer: {:?}", e)))?;
        // Some interfaces may not return Content-Length header, e.g. ListObjects.
        let cap = resp
            .headers
            .get("Content-Length")
            .and_then(|value| value.parse::<u64>().ok())
            .unwrap_or_default();
        debug!("headers: {:?}, cap: {}", resp.headers, cap);
        if cap > 0 {
            writer.reserve_capacity(cap);
        }
        let mut read_len = 0;
        let mut throttled = Duration::from_secs(0);
        while let Some(res) = tokio::time::timeout(self.opts.read_body_timeout.0, resp.body.next())
            .await
            .map_err(|e| HttpDispatchError::new(format!("read body timeout {:?}", e)))?
        {
            let chunk = res.map_err(|e| HttpDispatchError::new(format!("{:?}", e)))?;
            read_len += chunk.len() as u64;
            writer
                .write_all(&chunk)
                .map_err(|e| HttpDispatchError::new(format!("write_all: {:?}", e)))?;
            if let Some(read_speed_limiter) = self.read_speed_limiter.as_ref() {
                if need_throttle {
                    let sleep_dur = read_speed_limiter.consume_duration(chunk.len());
                    if !sleep_dur.is_zero() {
                        tokio::time::sleep(sleep_dur).await;
                        throttled += sleep_dur;
                    }
                }
            }
        }
        if cap > 0 && read_len != cap {
            return Err(HttpDispatchError::new(format!(
                "content length mismatch: expected {}, got {}",
                cap, read_len
            )));
        }
        Ok((writer, read_len, throttled))
    }

    pub async fn get_object(
        &self,
        key: String,
        file_name: String,
        opts: GetObjectOptions,
    ) -> crate::dfs::Result<Bytes> {
        let (writer, _) = self
            .get_object_to_writer(key, file_name, opts, &|| Ok(BytesMut::new().writer()))
            .await?;
        Ok(writer.into_inner().freeze())
    }

    pub async fn get_object_to_writer<W, F>(
        &self,
        key: String,
        file_name: String,
        opts: GetObjectOptions,
        build_writer: F,
    ) -> crate::dfs::Result<(W, u64)>
    where
        W: ReservableWriter + Send + 'static,
        F: Fn() -> std::io::Result<W> + 'static,
    {
        let mut retry_cnt = 0;
        let file_type = get_file_type_from_key(&key);
        let start_time_with_retry = Instant::now_coarse();
        loop {
            let start_time = Instant::now_coarse();
            let mut req = self.new_request("GET", &key);
            if !opts.is_full_range() {
                req.add_header("Range", &format!("bytes={}", opts.range_string()));
            }
            let mut result = self.dispatch(req, GetObjectError::from_response).await;

            if result.is_ok() {
                let mut resp = result.unwrap();
                let res = self
                    .read_body_to_writer(&mut resp, &build_writer, true)
                    .await;
                match res {
                    Ok((writer, object_len, throttled)) => {
                        info!(
                            "read file {}, size {}, takes {:?}, retry {}",
                            &file_name,
                            object_len,
                            start_time.saturating_elapsed(),
                            retry_cnt
                        );

                        let duration = start_time.saturating_elapsed();
                        KVENGINE_DFS_THROUGHPUT
                            .s3
                            .get
                            .get(MetricsFileType::from(file_type))
                            .inc_by(object_len);
                        KVENGINE_DFS_LATENCY
                            .get
                            .get(MetricsFileType::from(file_type))
                            .observe(duration.as_secs_f64());
                        if !throttled.is_zero() {
                            KVENGINE_DFS_THROTTLED
                                .s3
                                .get
                                .observe(throttled.as_secs_f64());
                        }
                        let duration_with_retry = start_time_with_retry.saturating_elapsed();
                        KVENGINE_DFS_LATENCY_WITH_RETRY
                            .get
                            .get(MetricsFileType::from(file_type))
                            .observe(duration_with_retry.as_secs_f64());
                        KVENGINE_DFS_REQUEST_COUNTER
                            .get
                            .get(MetricsFileType::from(file_type))
                            .inc();
                        return Ok((writer, object_len));
                    }
                    Err(err) => result = Err(err.into()),
                }
            }
            let err = result.unwrap_err();
            if let RusotoError::Service(GetObjectError::NoSuchKey(err_msg)) = &err {
                error!("file {} not exist, err msg {}", &file_name, err_msg);
                return Err(dfs::Error::NoSuchKey(format!(
                    "file {} not exist,",
                    &file_name
                )));
            }
            if self.is_err_not_found(&err) {
                error!("file {} not exist, err {}", &file_name, err.to_string());
                return Err(dfs::Error::NoSuchKey(format!(
                    "file {} not exist,",
                    &file_name
                )));
            }
            if self.is_err_retryable(&err) && self.sleep_for_retry(&mut retry_cnt, &file_name).await
            {
                KVENGINE_DFS_RETRY_COUNTER
                    .get
                    .get(MetricsFileType::from(file_type))
                    .inc();
                warn!("retry read file {}, error {:?}", &file_name, &err);
                continue;
            }
            error!("read file {}, error {:?}", &file_name, err);
            return Err(err.into());
        }
    }

    pub async fn put_object(
        &self,
        key: String,
        data: Bytes,
        file_name: String,
    ) -> crate::dfs::Result<()> {
        self.put_object_with_options(key, data, file_name, None, None, None)
            .await
    }

    async fn put_object_with_options(
        &self,
        key: String,
        data: Bytes,
        file_name: String,
        tagging: Option<&Tagging>,
        storage_class: Option<&str>,
        checksum: Option<u32>, // checksum saved in big-endian order
    ) -> crate::dfs::Result<()> {
        if self.read_only {
            return Err(Error::ReadOnly);
        }

        fail_point!(
            "s3fs_put_wal_chunks_error",
            key.contains("wal_chunks"),
            |_| Err(Error::Other(
                "s3fs_put_wal_chunks_error failpoint".to_string(),
            ))
        );

        let mut retry_cnt = 0;
        let start_time_with_retry = Instant::now();
        let data_len = data.len();
        let file_type = get_file_type_from_key(&key);
        loop {
            let start_time = Instant::now();
            let mut req = self.new_request("PUT", &key);
            req.add_header("Content-Length", &format!("{}", data.len()));
            if let Some(tagging) = tagging {
                req.add_header("x-amz-tagging", &tagging.to_url_encoded());
            }
            if let Some(storage_class) = storage_class.as_ref() {
                if self.is_on_aws()
                    || self.is_on_aliyun()
                    || self.is_on_tencentcloud()
                    || self.is_on_volcengine()
                {
                    self.provider.add_provider_header(
                        &mut req,
                        "x-amz-storage-class",
                        storage_class,
                    );
                } else {
                    debug!(
                        "{} ignore storage_class {} which is not supported by {}",
                        key, storage_class, self.hostname
                    );
                }
            }
            if let Some(checksum) = checksum {
                if self.is_on_aws() {
                    // `x-amz-checksum-crc32` value is base64 encoded in big-endian order.
                    let checksum = encode_checksum_crc32(checksum);
                    req.add_header(S3_CHECKSUM_CRC32_HEADER, &checksum);
                } else {
                    debug!(
                        "{} ignore x-amz-checksum-crc32 which is not supported by {}",
                        key, self.hostname
                    );
                }
            }
            let data = data.clone();
            let stream = futures::stream::once(async move { Ok(data) });
            req.set_payload_stream(rusoto_core::ByteStream::new(stream));
            let data_mb = data_len as u32 / (1024 * 1024);
            let timeout = self.opts.dispatch_timeout.0 + self.opts.write_timeout_per_mb.0 * data_mb;
            let result = self
                .dispatch_with_timeout(req, timeout, PutObjectError::from_response)
                .await;
            if result.is_ok() {
                info!(
                    "create file {}, size {}, takes {:?}, retry {}",
                    &file_name,
                    data_len,
                    start_time.saturating_elapsed(),
                    retry_cnt
                );
                KVENGINE_DFS_THROUGHPUT
                    .s3
                    .put
                    .get(MetricsFileType::from(file_type))
                    .inc_by(data_len as u64);
                KVENGINE_DFS_LATENCY
                    .put
                    .get(MetricsFileType::from(file_type))
                    .observe(start_time.saturating_elapsed().as_secs_f64());
                KVENGINE_DFS_LATENCY_WITH_RETRY
                    .put
                    .get(MetricsFileType::from(file_type))
                    .observe(start_time_with_retry.saturating_elapsed().as_secs_f64());
                KVENGINE_DFS_REQUEST_COUNTER
                    .put
                    .get(MetricsFileType::from(file_type))
                    .inc();

                return Ok(());
            }
            let err = result.unwrap_err();
            if self.is_err_retryable(&err) {
                if retry_cnt < self.opts.max_retry_count {
                    KVENGINE_DFS_RETRY_COUNTER
                        .put
                        .get(MetricsFileType::from(file_type))
                        .inc();
                    retry_cnt += 1;
                    let retry_sleep =
                        retry_sleep_duration(retry_cnt, self.opts.retry_sleep_interval.0);
                    tokio::time::sleep(retry_sleep).await;
                    warn!("retry create file {}, error {:?}", &file_name, &err);
                    continue;
                } else {
                    error!(
                        "create file {}, takes {:?}, reach max retry count {}",
                        &file_name,
                        start_time_with_retry.saturating_elapsed(),
                        self.opts.max_retry_count
                    );
                }
            }
            return Err(err.into());
        }
    }

    // Ref: https://docs.aws.amazon.com/AmazonS3/latest/API/API_DeleteObject.html
    pub async fn delete_object(&self, key: String, file_name: String) -> crate::dfs::Result<()> {
        if self.read_only {
            return Err(Error::ReadOnly);
        }

        let mut retry_cnt = 0;
        let file_type = get_file_type_from_key(&key);
        let start_time_with_retry = Instant::now_coarse();
        loop {
            let start_time = Instant::now_coarse();
            let req = self.new_request("DELETE", &key);
            let result = self.dispatch(req, DeleteObjectError::from_response).await;
            if result.is_ok() {
                info!(
                    "delete file {}, takes {:?}, retry {}",
                    &file_name,
                    start_time.saturating_elapsed(),
                    retry_cnt
                );

                let duration = start_time.saturating_elapsed();
                KVENGINE_DFS_LATENCY
                    .delete
                    .get(MetricsFileType::from(file_type))
                    .observe(duration.as_secs_f64());
                let duration_with_retry = start_time_with_retry.saturating_elapsed();
                KVENGINE_DFS_LATENCY_WITH_RETRY
                    .delete
                    .get(MetricsFileType::from(file_type))
                    .observe(duration_with_retry.as_secs_f64());
                KVENGINE_DFS_REQUEST_COUNTER
                    .delete
                    .get(MetricsFileType::from(file_type))
                    .inc();
                return Ok(());
            }
            let err = result.unwrap_err();
            if self.is_err_retryable(&err) {
                if retry_cnt < self.opts.max_retry_count {
                    KVENGINE_DFS_RETRY_COUNTER
                        .delete
                        .get(MetricsFileType::from(file_type))
                        .inc();
                    retry_cnt += 1;
                    let retry_sleep =
                        retry_sleep_duration(retry_cnt, self.opts.retry_sleep_interval.0);
                    tokio::time::sleep(retry_sleep).await;
                    warn!(
                        "retry delete file {}, error {:?}, retry cnt {}, retry after {:?}",
                        &file_name, &err, retry_cnt, retry_sleep
                    );
                    continue;
                } else {
                    error!(
                        "delete file {}, takes {:?}, reach max retry count {}",
                        &file_name,
                        start_time_with_retry.saturating_elapsed(),
                        self.opts.max_retry_count
                    );
                }
            }
            return Err(err.into());
        }
    }

    /// Copy object from `source_file_id` to `target_file_id`.
    ///
    /// `source_file_id` and `target_file_id` can be the same. And it's the only
    /// way to update the object creation timestamp, which is used for
    /// expiration rules of lifecycle.
    ///
    /// `target_tags`: replace tags if some, otherwise copy from source.
    ///
    /// `target_storage_class`: copy to another storage class if some. Note that
    /// only AWS S3 / OSS / COS support storage class.
    async fn copy_object(
        &self,
        source_key: &str,
        target_key: &str,
        target_tagging: Option<&Tagging>,
        target_storage_class: Option<&str>,
    ) -> Result<(), dfs::Error> {
        let full_source_key = format!("/{}/{}", self.bucket, source_key);
        self.raw_copy_object(
            &full_source_key,
            target_key,
            target_tagging,
            target_storage_class,
        )
        .await
    }

    /// Copy object from `source_key` to `target_key`.
    ///
    /// "raw" means the source key will be directly put to `x-amz-copy-source`.
    ///
    /// So this can be used to copy objects across buckets. For now copying a
    /// packed backup uses this.
    pub async fn raw_copy_object(
        &self,
        source_key: &str,
        target_key: &str,
        target_tagging: Option<&Tagging>,
        target_storage_class: Option<&str>,
    ) -> Result<(), dfs::Error> {
        self.raw_copy_object_impl(source_key, target_key, target_tagging, target_storage_class)
            .await
    }

    async fn raw_copy_object_impl(
        &self,
        source_key: &str,
        target_key: &str,
        target_tagging: Option<&Tagging>,
        target_storage_class: Option<&str>,
    ) -> Result<(), dfs::Error> {
        if self.read_only {
            return Err(Error::ReadOnly);
        }

        let metrics_file_type = MetricsFileType::from(get_file_type_from_key(target_key));
        let start_time_with_retry = Instant::now_coarse();
        let mut retry_cnt = 0;
        loop {
            let start_time = Instant::now_coarse();
            let mut req = self.new_request("PUT", target_key);
            req.add_header("x-amz-copy-source", source_key);
            req.add_header("x-amz-metadata-directive", "REPLACE");
            if let Some(target_tagging) = target_tagging {
                req.add_header("x-amz-tagging", &target_tagging.to_url_encoded());
                req.add_header("x-amz-tagging-directive", "REPLACE");
            }
            if let Some(target_storage_class) = target_storage_class.as_ref() {
                if self.is_on_aws()
                    || self.is_on_aliyun()
                    || self.is_on_tencentcloud()
                    || self.is_on_volcengine()
                {
                    self.provider.add_provider_header(
                        &mut req,
                        "x-amz-storage-class",
                        target_storage_class,
                    );
                } else {
                    debug!(
                        "{} ignore target_storage_class {} which is not supported by {}",
                        target_key, target_storage_class, self.hostname
                    );
                }
            }
            // Dispatch the request; on failure decide whether to retry.
            // S3 CopyObject can return HTTP 200 with an error in the XML body
            // (InternalError or SlowDown). We verify the body contains the
            // <CopyObjectResult> success marker; anything else is an error.
            // Ref: https://repost.aws/knowledge-center/s3-resolve-200-internalerror
            let body_err = match self.dispatch(req, CopyObjectError::from_response).await {
                Err(err) => {
                    // Non-retryable dispatch errors (DNS, validation, parse)
                    // fail immediately instead of burning through all retries.
                    if !self.is_err_retryable(&err) {
                        let err_msg = format!(
                            "failed to copy file {} from {}, non-retryable err {:?}",
                            target_key, source_key, err,
                        );
                        error!("{}", err_msg);
                        return Err(dfs::Error::S3(err_msg));
                    }
                    Some(format!(
                        "failed to copy file {} from {}, err {:?}",
                        target_key, source_key, err,
                    ))
                }
                Ok(mut resp) => match self
                    .read_body(&mut resp, false)
                    .await
                    .map(|b| String::from_utf8_lossy(&b).into_owned())
                {
                    Ok(body_str) if body_str.contains("<CopyObjectResult") => None,
                    Ok(body_str) => Some(format!(
                        "copy object {} from {} got 200 without CopyObjectResult: {}",
                        target_key, source_key, body_str,
                    )),
                    Err(err) => Some(format!(
                        "failed to read copy response body for {} from {}: {:?}",
                        target_key, source_key, err,
                    )),
                },
            };

            // Retry on transient errors with exponential backoff.
            if let Some(err_msg) = body_err {
                if retry_cnt < self.opts.max_retry_count {
                    KVENGINE_DFS_RETRY_COUNTER.copy.get(metrics_file_type).inc();
                    retry_cnt += 1;
                    let retry_sleep =
                        retry_sleep_duration(retry_cnt, self.opts.retry_sleep_interval.0);
                    warn!(
                        "retry copy file {}, retry count {}, retry after {:?}, err: {}",
                        target_key, retry_cnt, retry_sleep, err_msg,
                    );
                    tokio::time::sleep(retry_sleep).await;
                    continue;
                }
                let err_msg = format!(
                    "failed to copy file {} from {}, reach max retry count {}, err: {}",
                    target_key, source_key, self.opts.max_retry_count, err_msg,
                );
                error!("{}", err_msg);
                return Err(dfs::Error::S3(err_msg));
            }

            let duration = start_time.saturating_elapsed();
            KVENGINE_DFS_LATENCY
                .copy
                .get(metrics_file_type)
                .observe(duration.as_secs_f64());
            let duration_with_retry = start_time_with_retry.saturating_elapsed();
            KVENGINE_DFS_LATENCY_WITH_RETRY
                .copy
                .get(metrics_file_type)
                .observe(duration_with_retry.as_secs_f64());
            KVENGINE_DFS_REQUEST_COUNTER
                .copy
                .get(metrics_file_type)
                .inc();

            info!(
                "copy object {} from {}, takes {:?}, retry {}",
                target_key, source_key, duration_with_retry, retry_cnt
            );
            if self.verifier.should_verify() {
                let size = self
                    .object_size(target_key.to_string(), target_key.to_string())
                    .await
                    .unwrap_or_default();
                if size == 0 {
                    return Err(dfs::Error::S3(format!(
                        "failed to copy object from {} to {}",
                        source_key, target_key
                    )));
                }
            }
            return Ok(());
        }
    }

    pub async fn retain_file(&self, file_key: &str) -> Result<(), dfs::Error> {
        if self.is_removed(file_key).await? {
            let empty_tagging = Tagging::default();
            self.copy_object(
                file_key,
                file_key,
                Some(&empty_tagging),
                Some(self.storage_class_str(StorageClass::Default)),
            )
            .await?;
        }
        Ok(())
    }

    /// Choose proper storage class for removed files.
    fn choose_storage_class_for_removed_files(&self, file_len: Option<u64>) -> &'static str {
        // If removed_files_to_ia is disabled, keep the current storage class (use
        // default).
        if !self.removed_files_to_ia.load(Ordering::Relaxed) {
            return STORAGE_CLASS_DEFAULT;
        }
        // STORAGE_CLASS_STANDARD_IA is more cost efficient than STORAGE_CLASS_STANDARD
        // for NOT small files.
        if file_len.is_some() && file_len.unwrap() > SMALL_FILE_THRESHOLD_BYTES {
            self.storage_class_str(StorageClass::InfrequentAccess)
        } else {
            self.storage_class_str(StorageClass::Default)
        }
    }

    pub async fn begin_upload(
        &self,
        key: &str,
        storage_class: Option<&str>,
        checksum_algorithm: Option<&str>,
    ) -> Result<String, dfs::Error> {
        if self.read_only {
            return Err(Error::ReadOnly);
        }
        let full_key = format!("{}/{}", self.prefix, key);
        let mut retry_cnt = 0;
        let metrics_file_type = MetricsFileType::from(get_file_type_from_key(&full_key));
        let start_time_with_retry = Instant::now();
        loop {
            let start_time = Instant::now();
            let mut req = self.new_request("POST", &full_key);
            req.params.put_key("uploads");
            if let Some(storage_class) = storage_class {
                if self.is_on_aws()
                    || self.is_on_aliyun()
                    || self.is_on_tencentcloud()
                    || self.is_on_volcengine()
                {
                    self.provider.add_provider_header(
                        &mut req,
                        "x-amz-storage-class",
                        storage_class,
                    );
                } else {
                    debug!(
                        "{} ignore storage_class {} which is not supported by {}",
                        key, storage_class, self.hostname
                    );
                }
            }
            if let Some(checksum_algorithm) = checksum_algorithm {
                if self.is_on_aws() {
                    req.add_header(S3_CHECKSUM_ALGORITHM_HEADER, checksum_algorithm);
                } else {
                    debug!(
                        "{} ignore x-amz-checksum-algorithm which is not supported by {}",
                        key, self.hostname
                    );
                }
            }
            match self.dispatch_begin_upload_request(req).await {
                Ok(upload_id) => {
                    KVENGINE_DFS_LATENCY
                        .begin_multipart_upload
                        .get(metrics_file_type)
                        .observe(start_time.saturating_elapsed().as_secs_f64());
                    KVENGINE_DFS_LATENCY_WITH_RETRY
                        .begin_multipart_upload
                        .get(metrics_file_type)
                        .observe(start_time_with_retry.saturating_elapsed().as_secs_f64());
                    KVENGINE_DFS_REQUEST_COUNTER
                        .begin_multipart_upload
                        .get(metrics_file_type)
                        .inc();
                    return Ok(upload_id);
                }
                Err(err) => {
                    if self.is_multipart_upload_err_retryable(&err)
                        && retry_cnt < self.opts.max_retry_count
                    {
                        KVENGINE_DFS_RETRY_COUNTER
                            .begin_multipart_upload
                            .get(metrics_file_type)
                            .inc();
                        retry_cnt += 1;
                        let retry_sleep =
                            retry_sleep_duration(retry_cnt, self.opts.retry_sleep_interval.0);
                        warn!(
                            "retry begin multipart upload file {}, retry count {}, retry after {:?}, err {:?}",
                            key, retry_cnt, retry_sleep, err,
                        );
                        tokio::time::sleep(retry_sleep).await;
                        continue;
                    }
                    return Err(Error::S3(format!(
                        "Begin multipart upload for file {} failed {:?}",
                        key, err
                    )));
                }
            };
        }
    }

    async fn dispatch_begin_upload_request(
        &self,
        req: SignedRequest,
    ) -> Result<String, RusotoError<CreateMultipartUploadError>> {
        let mut resp = self
            .dispatch(req, CreateMultipartUploadError::from_response)
            .await?;
        let body_res = self.read_body(&mut resp, false).await?;
        let body_str = body_res
            .to_str()
            .map_err(|e| RusotoError::ParseError(format!("body to_str {}", e)))?;
        let res: InitiateMultipartUploadResult = quick_xml::de::from_str(body_str)
            .map_err(|e| RusotoError::ParseError(format!("parse body {}", e)))?;
        Ok(res.upload_id)
    }

    pub async fn upload_part(
        &self,
        key: &str,
        upload_id: &str,
        part_number: u64,
        data: Bytes,
        checksum: Option<u32>,
    ) -> Result<String, dfs::Error> {
        if self.read_only {
            return Err(Error::ReadOnly);
        }
        let full_key = format!("{}/{}", self.prefix, key);
        let mut retry_cnt = 0;
        let data_len = data.len();
        let metrics_file_type = MetricsFileType::from(get_file_type_from_key(&full_key));
        let start_time_with_retry = Instant::now();
        loop {
            let start_time = Instant::now();
            let mut req = self.new_request("PUT", &full_key);
            req.add_param("uploadId", upload_id);
            req.add_param("partNumber", &part_number.to_string());
            req.add_header("Content-Length", &format!("{}", data.len()));
            if let Some(checksum) = checksum {
                if self.is_on_aws() {
                    let checksum = encode_checksum_crc32(checksum);
                    req.add_header(S3_CHECKSUM_CRC32_HEADER, &checksum);
                } else {
                    debug!(
                        "{} ignore x-amz-checksum-crc32 for part {} which is not supported by {}",
                        key, part_number, self.hostname
                    );
                }
            }
            let data = data.clone();
            let stream = futures::stream::once(async move { Ok(data) });
            req.set_payload_stream(rusoto_core::ByteStream::new(stream));
            match self.dispatch_upload_part_request(req).await {
                Ok(e_tag) => {
                    KVENGINE_DFS_THROUGHPUT
                        .s3
                        .upload_part
                        .get(metrics_file_type)
                        .inc_by(data_len as u64);
                    KVENGINE_DFS_LATENCY
                        .upload_part
                        .get(metrics_file_type)
                        .observe(start_time.saturating_elapsed().as_secs_f64());
                    KVENGINE_DFS_LATENCY_WITH_RETRY
                        .upload_part
                        .get(metrics_file_type)
                        .observe(start_time_with_retry.saturating_elapsed().as_secs_f64());
                    KVENGINE_DFS_REQUEST_COUNTER
                        .upload_part
                        .get(metrics_file_type)
                        .inc();
                    return Ok(e_tag);
                }
                Err(err) => {
                    if self.is_multipart_upload_err_retryable(&err)
                        && retry_cnt < self.opts.max_retry_count
                    {
                        KVENGINE_DFS_RETRY_COUNTER
                            .upload_part
                            .get(metrics_file_type)
                            .inc();
                        retry_cnt += 1;
                        let retry_sleep =
                            retry_sleep_duration(retry_cnt, self.opts.retry_sleep_interval.0);
                        warn!(
                            "retry upload part {} for file {}, retry count {}, retry after {:?}, err {:?}",
                            part_number, key, retry_cnt, retry_sleep, err,
                        );
                        tokio::time::sleep(retry_sleep).await;
                        continue;
                    }
                    return Err(Error::S3(format!(
                        "Upload part {} for file {} failed {:?}",
                        part_number, key, err
                    )));
                }
            };
        }
    }

    async fn dispatch_upload_part_request(
        &self,
        req: SignedRequest,
    ) -> Result<String, RusotoError<UploadPartError>> {
        let resp = self.dispatch(req, UploadPartError::from_response).await?;
        if let Some(e_tag) = resp.headers.get("ETag") {
            return Ok(e_tag.to_string());
        }
        Err(RusotoError::ParseError(
            "Upload part missing ETag in response".to_string(),
        ))
    }

    pub async fn complete_multipart_upload(
        &self,
        key: &str,
        upload_id: &str,
        parts: Vec<MultipartUploadPart>,
    ) -> Result<(), dfs::Error> {
        if self.read_only {
            return Err(Error::ReadOnly);
        }
        let full_key = format!("{}/{}", self.prefix, key);
        let mut retry_cnt = 0;
        let metrics_file_type = MetricsFileType::from(get_file_type_from_key(&full_key));
        let complete_multipart_upload = CompleteMultipartUpload {
            parts: parts
                .into_iter()
                .map(|part| CompletedPart {
                    part_number: part.part_number,
                    e_tag: part.e_tag,
                    checksum_crc32: part.checksum_crc32.map(encode_checksum_crc32),
                })
                .collect(),
        };
        let body = complete_multipart_upload.to_xml();
        let start_time_with_retry = Instant::now();
        loop {
            let start_time = Instant::now();
            let mut req = self.new_request("POST", &full_key);
            req.add_param("uploadId", upload_id);
            req.add_header("Content-Length", &format!("{}", body.len()));
            req.set_payload(Some(body.clone().into_bytes()));
            match self.dispatch_complete_multipart_upload_request(req).await {
                Ok(_) => {
                    KVENGINE_DFS_LATENCY
                        .complete_multipart_upload
                        .get(metrics_file_type)
                        .observe(start_time.saturating_elapsed().as_secs_f64());
                    KVENGINE_DFS_LATENCY_WITH_RETRY
                        .complete_multipart_upload
                        .get(metrics_file_type)
                        .observe(start_time_with_retry.saturating_elapsed().as_secs_f64());
                    KVENGINE_DFS_REQUEST_COUNTER
                        .complete_multipart_upload
                        .get(metrics_file_type)
                        .inc();
                    return Ok(());
                }
                Err(err) => {
                    if self.is_complete_multipart_upload_err_retryable(&err)
                        && retry_cnt < self.opts.max_retry_count
                    {
                        KVENGINE_DFS_RETRY_COUNTER
                            .complete_multipart_upload
                            .get(metrics_file_type)
                            .inc();
                        retry_cnt += 1;
                        let retry_sleep =
                            retry_sleep_duration(retry_cnt, self.opts.retry_sleep_interval.0);
                        warn!(
                            "retry complete multipart upload for file {}, retry count {}, retry after {:?}, err {:?}",
                            key, retry_cnt, retry_sleep, err,
                        );
                        tokio::time::sleep(retry_sleep).await;
                        continue;
                    }
                    return Err(Error::S3(format!(
                        "Complete multipart upload for file {} failed {:?}",
                        key, err
                    )));
                }
            };
        }
    }

    async fn dispatch_complete_multipart_upload_request(
        &self,
        req: SignedRequest,
    ) -> Result<(), CompleteMultipartUploadResponseError> {
        let mut resp = self
            .dispatch(req, CompleteMultipartUploadError::from_response)
            .await
            .map_err(CompleteMultipartUploadResponseError::Request)?;
        let body_res = self.read_body(&mut resp, false).await?;
        let body_str = body_res
            .to_str()
            .map_err(|e| CompleteMultipartUploadResponseError::UnexpectedBody(format!(
                "complete multipart upload got 200 without CompleteMultipartUploadResult, invalid UTF-8 body: {}",
                e
            )))?;
        quick_xml::de::from_str::<CompleteMultipartUploadResult>(body_str)
            .map(|_| ())
            .map_err(|err| CompleteMultipartUploadResponseError::UnexpectedBody(format!(
                "complete multipart upload got 200 without CompleteMultipartUploadResult: {}, body: {}",
                err, body_str
            )))
    }

    fn is_complete_multipart_upload_err_retryable(
        &self,
        err: &CompleteMultipartUploadResponseError,
    ) -> bool {
        match err {
            CompleteMultipartUploadResponseError::Request(err) => {
                self.is_multipart_upload_err_retryable(err)
            }
            CompleteMultipartUploadResponseError::UnexpectedBody(_) => true,
        }
    }

    pub async fn abort_multipart_upload(
        &self,
        key: &str,
        upload_id: &str,
    ) -> Result<(), dfs::Error> {
        if self.read_only {
            return Err(Error::ReadOnly);
        }
        let full_key = format!("{}/{}", self.prefix, key);
        let mut retry_cnt = 0;
        let metrics_file_type = MetricsFileType::from(get_file_type_from_key(&full_key));
        let start_time_with_retry = Instant::now();
        loop {
            let start_time = Instant::now();
            let mut req = self.new_request("DELETE", &full_key);
            req.add_param("uploadId", upload_id);
            match self
                .dispatch(req, AbortMultipartUploadError::from_response)
                .await
            {
                Ok(_) => {
                    KVENGINE_DFS_LATENCY
                        .abort_multipart_upload
                        .get(metrics_file_type)
                        .observe(start_time.saturating_elapsed().as_secs_f64());
                    KVENGINE_DFS_LATENCY_WITH_RETRY
                        .abort_multipart_upload
                        .get(metrics_file_type)
                        .observe(start_time_with_retry.saturating_elapsed().as_secs_f64());
                    KVENGINE_DFS_REQUEST_COUNTER
                        .abort_multipart_upload
                        .get(metrics_file_type)
                        .inc();
                    return Ok(());
                }
                Err(err) => {
                    if self.is_multipart_upload_err_retryable(&err)
                        && retry_cnt < self.opts.max_retry_count
                    {
                        KVENGINE_DFS_RETRY_COUNTER
                            .abort_multipart_upload
                            .get(metrics_file_type)
                            .inc();
                        retry_cnt += 1;
                        let retry_sleep =
                            retry_sleep_duration(retry_cnt, self.opts.retry_sleep_interval.0);
                        warn!(
                            "retry abort multipart upload for file {}, retry count {}, retry after {:?}, err {:?}",
                            key, retry_cnt, retry_sleep, err,
                        );
                        tokio::time::sleep(retry_sleep).await;
                        continue;
                    }
                    return Err(Error::S3(format!(
                        "Abort multipart upload for file {} failed {:?}",
                        key, err
                    )));
                }
            };
        }
    }
}

enum CompleteMultipartUploadResponseError {
    Request(RusotoError<CompleteMultipartUploadError>),
    UnexpectedBody(String),
}

impl Debug for CompleteMultipartUploadResponseError {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Request(err) => f.debug_tuple("Request").field(err).finish(),
            Self::UnexpectedBody(msg) => f.debug_tuple("UnexpectedBody").field(msg).finish(),
        }
    }
}

impl From<HttpDispatchError> for CompleteMultipartUploadResponseError {
    fn from(err: HttpDispatchError) -> Self {
        Self::Request(RusotoError::HttpDispatch(err))
    }
}

impl Drop for S3FsCore {
    fn drop(&mut self) {
        if let Some(runtime) = self.runtime.take() {
            std::thread::spawn(|| drop(runtime));
        }
    }
}

impl ObjectStorage for S3Fs {
    fn put_objects(&self, objects: Vec<(String, Bytes)>) -> Result<(), String> {
        let put_object = |key: String, data: Bytes| {
            let full_key = format!("{}/{}", self.prefix, key);
            let fs = self.clone();
            let checksum = if self.is_on_aws() {
                Some(crc32fast::hash(&data))
            } else {
                None
            };
            let storage_class_default = self.storage_class_str(StorageClass::Default);
            async move {
                fs.put_object_with_options(
                    full_key,
                    data,
                    key.clone(),
                    None,
                    Some(storage_class_default),
                    checksum,
                )
                .await
                .map_err(|err| format!("put {} failed {:?}", &key, err))
            }
        };

        let runtime = self.get_runtime();
        let len = objects.len();

        if len == 1 {
            let (key, data) = objects.into_iter().next().unwrap();
            return runtime.block_on(put_object(key, data));
        }

        let mut handles = Vec::with_capacity(len);
        for (key, data) in objects {
            handles.push(runtime.spawn(put_object(key, data)));
        }

        let errs: Vec<_> = runtime
            .block_on(futures::future::join_all(handles))
            .into_iter()
            .filter_map(|r| r.unwrap().err())
            .collect();
        if !errs.is_empty() {
            return Err(format!("{:?}", errs));
        }
        Ok(())
    }

    fn get_objects(
        &self,
        keys: Vec<(String, GetObjectOptions)>,
        cache: Option<&ObjectCacheWithHook>,
    ) -> Result<Vec<(String, Bytes)>, String> {
        let get_object =
            |key: String, opts: GetObjectOptions, cache: Option<&ObjectCacheWithHook>| {
                let full_key = format!("{}/{}", self.prefix, key);
                let fs = self.clone();
                let cache = cache.cloned();
                async move {
                    match fs
                        .get_object_with_cache(full_key, key.clone(), opts, cache.as_ref())
                        .await
                    {
                        Ok(data) => Ok((key, data)),
                        Err(err) => Err(format!("get {} failed {:?}", &key, err)),
                    }
                }
            };

        let runtime = self.get_runtime();
        if keys.len() == 1 {
            let mut keys = keys;
            let (key, opts) = keys.pop().unwrap();
            let res = runtime.block_on(get_object(key, opts, cache))?;
            return Ok(vec![res]);
        }

        let mut join_set = tokio::task::JoinSet::new();
        for (key, opts) in keys {
            join_set.spawn_on(get_object(key, opts, cache), runtime.handle());
        }

        runtime.block_on(async move {
            let mut objects: Vec<(String, Bytes)> = vec![];
            while let Some(res) = join_set.join_next().await {
                let res = box_join_err!(res).map_err(|e: BoxError| e.to_string())??;
                objects.push(res);
            }
            Ok(objects)
        })
    }

    fn list_objects(
        &self,
        start_after: &str,
        prefix: Option<&str>,
        max_keys: Option<u32>,
    ) -> Result<(Vec<ListObjectContent>, Option<String>), String> {
        let runtime = self.get_runtime();
        runtime
            .block_on(self.list(start_after, prefix, max_keys))
            .map(|v| (v.0, v.2))
            .map_err(|err| format!("list failed {:?}", err))
    }
}

#[async_trait]
impl Dfs for S3Fs {
    async fn exists(&self, file_id: u64, opts: Options) -> crate::dfs::Result<bool> {
        let file_key = self.file_key(file_id, opts.file_type);
        let file_name = format!("{}.{}", file_id, opts.file_type.suffix());
        self.exist(file_key, file_name).await
    }

    async fn read_file(&self, file_id: u64, opts: Options) -> crate::dfs::Result<Bytes> {
        let filename = if let Some(end_off) = opts.end_off {
            format!("{}-{}-{}.seg", file_id, opts.start_off, end_off)
        } else {
            format!("{}.{}", file_id, opts.file_type.suffix())
        };
        self.get_object(
            self.file_key(file_id, opts.file_type),
            filename,
            GetObjectOptions {
                start_off: Some(opts.start_off),
                end_off: opts.end_off,
            },
        )
        .await
    }

    async fn create(&self, file_id: u64, data: Bytes, opts: Options) -> crate::dfs::Result<()> {
        // Calculate checksum for the file. This can be used to ensure the data
        // integrity when saving to dfs (s3 only) and verify the data integrity when
        // getting objects from dfs.
        let checksum = if self.is_on_aws() {
            Some(crc32fast::hash(&data))
        } else {
            None
        };
        self.put_object_with_options(
            self.file_key(file_id, opts.file_type),
            data,
            format!("{}.{}", file_id, opts.file_type.suffix()),
            None,
            Some(self.storage_class_str(StorageClass::Default)),
            checksum,
        )
        .await
    }

    /// Logically remove the file on S3 by tagging with "deleted=true".
    /// And the file would be permanently removed after `gc_lifetime`. See
    /// `DfsGc`.
    async fn remove(&self, file_id: u64, file_len: Option<u64>, opts: Options) {
        // Only AWS / OSS / COS / TOS supports storage class.
        let target_storage_class = if self.is_on_aws()
            || self.is_on_aliyun()
            || self.is_on_tencentcloud()
            || self.is_on_volcengine()
        {
            let new_storage_class = self.choose_storage_class_for_removed_files(file_len);
            (new_storage_class != self.storage_class_str(StorageClass::Default))
                .then_some(new_storage_class)
        } else {
            None
        };
        let target_tagging = Tagging::new_single_deleted();
        let file_key = self.file_key(file_id, opts.file_type);
        let _ = self
            .copy_object(
                &file_key,
                &file_key,
                Some(&target_tagging),
                target_storage_class,
            )
            .await;
    }

    /// Permanently remove the file on S3.
    /// This method should be used by `DfsGc` ONLY to meet GC rules.
    async fn permanently_remove(&self, file_id: u64, opts: Options) -> crate::dfs::Result<()> {
        if self.is_on_aws()
            || self.is_on_aliyun()
            || self.is_on_tencentcloud()
            || self.is_on_volcengine()
        {
            return Err(Error::Other(format!(
                "{} permanently_remove is forbidden on AWS / alibaba cloud / tencent cloud / volcengine",
                file_id
            )));
        }

        self.delete_object(self.file_key(file_id, opts.file_type), file_id.to_string())
            .await
    }

    async fn list(
        &self,
        start_after: &str,
        prefix: Option<&str>,
        max_keys: Option<u32>,
    ) -> crate::dfs::Result<(Vec<ListObjectContent>, bool, Option<String>)> {
        self.core.list(start_after, prefix, max_keys).await
    }

    async fn get_object(
        &self,
        key: String,
        file_name: String,
        opts: GetObjectOptions,
    ) -> crate::dfs::Result<Bytes> {
        self.core.get_object(key, file_name, opts).await
    }

    async fn get_object_to_path(
        &self,
        key: String,
        file_name: String,
        opts: GetObjectOptions,
        path: &Path,
    ) -> crate::dfs::Result<u64> {
        let path = path.to_path_buf();
        let build_writer = move || -> std::io::Result<_> {
            let file = std::fs::OpenOptions::new()
                .create(true)
                .write(true)
                .read(true)
                .truncate(true)
                .open(&path)?;
            Ok(std::io::BufWriter::new(file))
        };
        let (mut writer, len) = self
            .get_object_to_writer(key, file_name, opts, build_writer)
            .await?;
        writer.flush()?;
        Ok(len)
    }

    async fn put_object(
        &self,
        key: String,
        data: Bytes,
        file_name: String,
    ) -> crate::dfs::Result<()> {
        self.core.put_object(key, data, file_name).await
    }

    async fn put_object_with_storage_class(
        &self,
        key: String,
        data: Bytes,
        file_name: String,
        storage_class: StorageClass,
    ) -> crate::dfs::Result<()> {
        let storage_class = Some(self.storage_class_str(storage_class));
        self.put_object_with_options(key, data, file_name, None, storage_class, None)
            .await
    }

    async fn exist(&self, key: String, file_name: String) -> crate::dfs::Result<bool> {
        self.core.exist(key, file_name).await
    }

    async fn retain_file(&self, file_key: &str) -> crate::dfs::Result<()> {
        self.core.retain_file(file_key).await
    }

    async fn delete_object(&self, key: String, file_name: String) -> crate::dfs::Result<()> {
        self.core.delete_object(key, file_name).await
    }

    async fn object_size(&self, key: String, file_name: String) -> crate::dfs::Result<u64> {
        self.core.object_size(key, file_name).await
    }

    async fn list_folders(
        &self,
        prefix: &str,
        delimiter: Option<&str>,
    ) -> crate::dfs::Result<Vec<String>> {
        self.core.list_folders(prefix, delimiter).await
    }

    async fn is_removed(&self, file_key: &str) -> crate::dfs::Result<bool> {
        self.core.is_removed(file_key).await
    }

    fn list_objects(
        &self,
        start_after: &str,
        prefix: Option<&str>,
        max_keys: Option<u32>,
    ) -> std::result::Result<(Vec<ListObjectContent>, Option<String>), String> {
        ObjectStorage::list_objects(self, start_after, prefix, max_keys)
    }

    fn put_objects(&self, objects: Vec<(String, Bytes)>) -> std::result::Result<(), String> {
        ObjectStorage::put_objects(self, objects)
    }

    async fn begin_upload(&self, key: &str) -> std::result::Result<String, String> {
        let storage_class_default = self.storage_class_str(StorageClass::Default);
        let checksum_algorithm = if self.is_on_aws() {
            Some(S3_CHECKSUM_ALGORITHM_CRC32)
        } else {
            None
        };
        self.core
            .begin_upload(key, Some(storage_class_default), checksum_algorithm)
            .await
            .map_err(|err| format!("begin upload failed {:?}", err))
    }

    async fn upload_part(
        &self,
        key: &str,
        upload_id: &str,
        part_number: u64,
        data: Bytes,
    ) -> std::result::Result<MultipartUploadPart, String> {
        let checksum_crc32 = if self.is_on_aws() {
            Some(crc32fast::hash(&data))
        } else {
            None
        };
        let e_tag = self
            .core
            .upload_part(key, upload_id, part_number, data, checksum_crc32)
            .await
            .map_err(|err| {
                format!(
                    "upload part key:{} num: {} failed {:?}",
                    key, part_number, err
                )
            })?;
        Ok(MultipartUploadPart {
            part_number,
            e_tag,
            checksum_crc32,
        })
    }

    async fn complete_multipart_upload(
        &self,
        key: &str,
        upload_id: &str,
        parts: Vec<MultipartUploadPart>,
    ) -> std::result::Result<(), String> {
        self.core
            .complete_multipart_upload(key, upload_id, parts)
            .await
            .map_err(|err| format!("complete upload key:{} failed {:?}", key, err))
    }

    async fn abort_multipart_upload(
        &self,
        key: &str,
        upload_id: &str,
    ) -> std::result::Result<(), String> {
        self.core
            .abort_multipart_upload(key, upload_id)
            .await
            .map_err(|err| format!("abort upload key:{} failed {:?}", key, err))
    }

    async fn raw_copy_object(
        &self,
        source_key: &str,
        target_key: &str,
        target_tagging: Option<&Tagging>,
        target_storage_class: Option<&str>,
    ) -> crate::dfs::Result<()> {
        self.raw_copy_object_impl(source_key, target_key, target_tagging, target_storage_class)
            .await
    }

    fn get_runtime(&self) -> &Runtime {
        self.runtime.as_ref().unwrap()
    }

    fn get_prefix(&self) -> String {
        self.prefix.clone()
    }

    fn get_storage_name(&self) -> String {
        self.get_bucket()
    }

    async fn size(&self, file_id: u64, opts: Options) -> crate::dfs::Result<u64> {
        // Use HEAD to check and get size from Content-Length where possible
        let key = self.file_key(file_id, opts.file_type);
        let name = format!("{}.{}", file_id, opts.file_type.suffix());
        let mut retry_cnt = 0;
        let file_type = get_file_type_from_key(&key);
        let start_time_with_retry = Instant::now();
        loop {
            let start_time = Instant::now();
            let req = self.new_request("HEAD", &key);
            let result = self.dispatch(req, HeadObjectError::from_response).await;
            match result {
                Ok(resp) => {
                    // Try to parse Content-Length
                    let len = match resp
                        .headers
                        .get("Content-Length")
                        .and_then(|v| v.parse::<u64>().ok())
                    {
                        Some(len) => len,
                        None => {
                            error!(
                                "missing or invalid Content-Length header for file {}",
                                &name
                            );
                            return Err(dfs::Error::Other(format!(
                                "missing or invalid Content-Length header for file {}",
                                &name
                            )));
                        }
                    };
                    let duration = start_time.saturating_elapsed();
                    KVENGINE_DFS_LATENCY
                        .head
                        .get(MetricsFileType::from(file_type))
                        .observe(duration.as_secs_f64());
                    let duration_with_retry = start_time_with_retry.saturating_elapsed();
                    KVENGINE_DFS_LATENCY_WITH_RETRY
                        .head
                        .get(MetricsFileType::from(file_type))
                        .observe(duration_with_retry.as_secs_f64());
                    KVENGINE_DFS_REQUEST_COUNTER
                        .head
                        .get(MetricsFileType::from(file_type))
                        .inc();
                    return Ok(len);
                }
                Err(err) => {
                    if let RusotoError::Service(HeadObjectError::NoSuchKey(err_msg)) = &err {
                        error!("file {} not exist, err msg {}", &name, err_msg);
                        return Err(dfs::Error::NoSuchKey(format!("file {} not exist,", &name)));
                    }
                    if self.is_err_not_found(&err) {
                        error!("file {} not exist, err {}", &name, err.to_string());
                        return Err(dfs::Error::NoSuchKey(format!("file {} not exist,", &name)));
                    }
                    if self.is_err_retryable(&err)
                        && self.sleep_for_retry(&mut retry_cnt, &name).await
                    {
                        KVENGINE_DFS_RETRY_COUNTER
                            .head
                            .get(MetricsFileType::from(file_type))
                            .inc();
                        warn!("retry head file {}, error {:?}", &name, &err);
                        continue;
                    }
                    return Err(err.into());
                }
            }
        }
    }
}

#[derive(Default, Debug, Serialize, Deserialize)]
#[serde(default)]
#[serde(rename_all = "PascalCase")]
pub struct ListObjects {
    pub common_prefixes: Vec<CommonPrefix>,
    pub contents: Vec<ListObjectContent>,
    pub is_truncated: bool,
}

#[derive(Clone, Default, Debug, Serialize, Deserialize)]
#[serde(default)]
#[serde(rename_all = "PascalCase")]
pub struct CommonPrefix {
    pub prefix: String,
}

#[derive(Clone, Default, Debug, Serialize, Deserialize)]
#[serde(default)]
#[serde(rename_all = "PascalCase")]
pub struct Tagging {
    tag_set: TagSet,
}

impl Tagging {
    fn new_single_deleted() -> Tagging {
        Self {
            tag_set: TagSet {
                tag: vec![Tag::new_deleted()],
            },
        }
    }

    pub fn add_tag(&mut self, key: String, value: String) {
        self.tag_set.tag.push(Tag { key, value });
    }

    fn has_deleted_tag(&self) -> bool {
        self.tag_set.tag.iter().any(|tag| tag.is_deleted())
    }

    pub fn to_xml(&self) -> String {
        quick_xml::se::to_string(&self).unwrap()
    }

    pub fn to_url_encoded(&self) -> String {
        let mut se = url::form_urlencoded::Serializer::new(String::new());
        for tag in &self.tag_set.tag {
            se.append_pair(&tag.key, &tag.value);
        }
        se.finish()
    }

    pub fn from_url_encoded(query: &str) -> Tagging {
        let mut tagging = Self::default();
        for (key, value) in url::form_urlencoded::parse(query.as_bytes()) {
            tagging.tag_set.tag.push(Tag {
                key: key.into_owned(),
                value: value.into_owned(),
            });
        }
        tagging
    }
}

#[derive(Clone, Default, Debug, Serialize, Deserialize)]
#[serde(default)]
#[serde(rename_all = "PascalCase")]
struct TagSet {
    tag: Vec<Tag>,
}

#[derive(Clone, Default, Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(default)]
#[serde(rename_all = "PascalCase")]
struct Tag {
    #[serde(rename = "$unflatten=Key")]
    key: String,
    #[serde(rename = "$unflatten=Value")]
    value: String,
}

impl Tag {
    fn new_deleted() -> Self {
        Self {
            key: "deleted".to_string(),
            value: "true".to_string(),
        }
    }

    fn is_deleted(&self) -> bool {
        self.key == "deleted" && self.value == "true"
    }
}

#[derive(Clone, Default, Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(default)]
#[serde(rename_all = "PascalCase")]
struct InitiateMultipartUploadResult {
    bucket: String,
    key: String,
    upload_id: String,
}

#[derive(Clone, Debug, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "PascalCase")]
struct CompleteMultipartUploadResult {
    location: String,
    bucket: String,
    key: String,
    #[serde(rename = "ETag")]
    e_tag: String,
}

#[derive(Clone, Default, Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(default)]
#[serde(rename_all = "PascalCase")]
struct CompleteMultipartUpload {
    #[serde(rename = "Part")]
    parts: Vec<CompletedPart>,
}

impl CompleteMultipartUpload {
    fn to_xml(&self) -> String {
        quick_xml::se::to_string(self).unwrap()
    }
}

#[derive(Clone, Default, Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(default)]
#[serde(rename_all = "PascalCase")]
struct CompletedPart {
    #[serde(rename = "$unflatten=PartNumber")]
    part_number: u64,
    #[serde(rename = "$unflatten=ETag")]
    e_tag: String,
    #[serde(
        rename = "$unflatten=ChecksumCRC32",
        skip_serializing_if = "Option::is_none"
    )]
    checksum_crc32: Option<String>,
}

struct Response {
    resp: HttpResponse,
}

impl Deref for Response {
    type Target = HttpResponse;

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

impl DerefMut for Response {
    fn deref_mut(&mut self) -> &mut Self::Target {
        &mut self.resp
    }
}

impl Debug for Response {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.resp.status)
    }
}

pub struct OperationVerifier {
    counter: AtomicUsize,
    limit: usize,
}

impl OperationVerifier {
    pub fn new(limit: usize) -> Self {
        Self {
            counter: AtomicUsize::new(0),
            limit,
        }
    }

    pub fn should_verify(&self) -> bool {
        if self.counter.load(Ordering::Relaxed) > self.limit {
            return false;
        }
        let count = self.counter.fetch_add(1, Ordering::Relaxed);
        count <= self.limit
    }
}

#[cfg(any(test, feature = "testexport"))]
pub mod test_util {
    use std::sync::{Arc, Mutex};

    use rusoto_core::signature::SignedRequest;
    use rusoto_mock::{MockCredentialsProvider, MockRequestDispatcher};

    use crate::dfs::{S3Fs, s3::ConnOptions};

    struct State {
        created: bool,
        file_data: Vec<u8>,
    }

    #[derive(Clone)]
    struct S3TestDispatcher {
        state: Arc<Mutex<State>>,
    }

    impl rusoto_core::request::DispatchSignedRequest for S3TestDispatcher {
        fn dispatch(
            &self,
            request: SignedRequest,
            _timeout: Option<std::time::Duration>,
        ) -> rusoto_core::request::DispatchSignedRequestFuture {
            let method = request.method.clone();
            let state = self.state.clone();
            Box::pin(async move {
                match method.as_str() {
                    "HEAD" => {
                        let (created, file_len) = {
                            let state = state.lock().unwrap();
                            (state.created, state.file_data.len())
                        };
                        if created {
                            MockRequestDispatcher::with_status(200)
                                .with_header("Content-Length", &file_len.to_string())
                                .dispatch(request, _timeout)
                                .await
                        } else {
                            MockRequestDispatcher::with_status(404)
                                .dispatch(request, _timeout)
                                .await
                        }
                    }
                    "PUT" => {
                        {
                            let mut state = state.lock().unwrap();
                            state.created = true;
                        }
                        let is_copy = request.headers.contains_key("x-amz-copy-source");
                        let mock = MockRequestDispatcher::with_status(200);
                        if is_copy {
                            mock.with_body(
                                "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\
                                 <CopyObjectResult>\
                                 <ETag>\"mock-etag\"</ETag>\
                                 <LastModified>2025-01-01T00:00:00.000Z</LastModified>\
                                 </CopyObjectResult>",
                            )
                            .dispatch(request, _timeout)
                            .await
                        } else {
                            mock.dispatch(request, _timeout).await
                        }
                    }
                    "GET" => {
                        let (created, file_data) = {
                            let state = state.lock().unwrap();
                            (state.created, state.file_data.clone())
                        };
                        if created {
                            MockRequestDispatcher::with_status(200)
                                .with_body(std::str::from_utf8(&file_data).unwrap())
                                .dispatch(request, _timeout)
                                .await
                        } else {
                            MockRequestDispatcher::with_status(404)
                                .dispatch(request, _timeout)
                                .await
                        }
                    }
                    _ => {
                        MockRequestDispatcher::with_status(200)
                            .dispatch(request, _timeout)
                            .await
                    }
                }
            })
        }
    }

    /// Creates a test S3Fs where CopyObject (PUT with x-amz-copy-source)
    /// returns HTTP 200 with the given XML body. This simulates the S3
    /// "200-with-Error" behavior for CopyObject operations.
    pub fn new_test_s3fs_copy_error(error_body: &str) -> S3Fs {
        let error_body = error_body.to_string();
        let dispatcher = CopyErrorDispatcher {
            error_body: Arc::new(error_body),
        };
        let s3c = rusoto_core::Client::new_with(MockCredentialsProvider, dispatcher);
        // Use a small retry count to keep the test fast (~3s instead of ~511s).
        let opts = ConnOptions {
            max_retry_count: 2,
            ..Default::default()
        };
        S3Fs::new_for_test_with_opts(s3c, "shard-db".into(), "prefix".into(), opts)
    }

    /// Creates a test S3Fs where CompleteMultipartUpload returns HTTP 200 with
    /// the given XML body.
    pub fn new_test_s3fs_complete_multipart_error(error_body: &str) -> S3Fs {
        let dispatcher = CompleteMultipartErrorDispatcher {
            error_body: Arc::new(error_body.to_owned()),
        };
        let s3c = rusoto_core::Client::new_with(MockCredentialsProvider, dispatcher);
        let opts = ConnOptions {
            max_retry_count: 0,
            ..Default::default()
        };
        S3Fs::new_for_test_with_opts(s3c, "shard-db".into(), "prefix".into(), opts)
    }

    #[derive(Clone)]
    struct CopyErrorDispatcher {
        error_body: Arc<String>,
    }

    impl rusoto_core::request::DispatchSignedRequest for CopyErrorDispatcher {
        fn dispatch(
            &self,
            request: SignedRequest,
            _timeout: Option<std::time::Duration>,
        ) -> rusoto_core::request::DispatchSignedRequestFuture {
            let is_copy = request.headers.contains_key("x-amz-copy-source");
            let error_body = self.error_body.clone();
            Box::pin(async move {
                if is_copy {
                    // Simulate S3 CopyObject 200-with-Error
                    MockRequestDispatcher::with_status(200)
                        .with_body(&error_body)
                        .dispatch(request, _timeout)
                        .await
                } else {
                    MockRequestDispatcher::with_status(200)
                        .dispatch(request, _timeout)
                        .await
                }
            })
        }
    }

    #[derive(Clone)]
    struct CompleteMultipartErrorDispatcher {
        error_body: Arc<String>,
    }

    impl rusoto_core::request::DispatchSignedRequest for CompleteMultipartErrorDispatcher {
        fn dispatch(
            &self,
            request: SignedRequest,
            _timeout: Option<std::time::Duration>,
        ) -> rusoto_core::request::DispatchSignedRequestFuture {
            let error_body = self.error_body.clone();
            Box::pin(async move {
                MockRequestDispatcher::with_status(200)
                    .with_body(&error_body)
                    .dispatch(request, _timeout)
                    .await
            })
        }
    }

    pub fn new_test_s3fs(file_data: &[u8]) -> S3Fs {
        let state = Arc::new(Mutex::new(State {
            created: false,
            file_data: file_data.to_vec(),
        }));
        let dispatcher = S3TestDispatcher { state };
        let s3c = rusoto_core::Client::new_with(MockCredentialsProvider, dispatcher);
        S3Fs::new_for_test(s3c, "shard-db".into(), "prefix".into())
    }
}

#[cfg(test)]
mod tests {
    use std::{
        fs,
        io::Write,
        time::{SystemTime, UNIX_EPOCH},
    };

    use bytes::Buf;

    use super::*;
    use crate::{
        dfs::{FileType, test_util::new_test_s3fs},
        table::{
            file::{File, LocalFile},
            sstable::new_filename,
        },
    };

    #[test]
    fn test_s3() {
        ::test_util::init_log_for_test();

        let local_dir = tempfile::tempdir().unwrap();
        let file_data = "abcdefgh".to_string().into_bytes();
        let s3fs = new_test_s3fs(&file_data);
        let file_id = 321u64;
        let opts = Options::default();

        // Test exists before create
        let fs = s3fs.clone();
        let (tx_exists_before, rx_exists_before) = tikv_util::mpsc::bounded(1);
        let f_exists_before = async move {
            match fs.exists(file_id, opts).await {
                Ok(exists) => {
                    tx_exists_before.send(exists).unwrap();
                }
                Err(err) => {
                    println!("exists before create error {:?}", err);
                    tx_exists_before.send(true).unwrap(); // Fail test on error
                }
            }
        };
        s3fs.get_runtime().spawn(f_exists_before);
        assert!(!rx_exists_before.recv().unwrap());

        // Test size before create - should return error
        let fs = s3fs.clone();
        let (tx_size_before, rx_size_before) = tikv_util::mpsc::bounded(1);
        let f_size_before = async move {
            match fs.size(file_id, opts).await {
                Ok(_) => {
                    tx_size_before.send(false).unwrap(); // Should not succeed
                }
                Err(_) => {
                    tx_size_before.send(true).unwrap(); // Expected: error
                }
            }
        };
        s3fs.get_runtime().spawn(f_size_before);
        assert!(rx_size_before.recv().unwrap());

        let fs = s3fs.clone();
        let (tx, rx) = tikv_util::mpsc::bounded(1);
        let file_data_clone_create = file_data.clone();
        let f_create = async move {
            match fs
                .create(file_id, bytes::Bytes::from(file_data_clone_create), opts)
                .await
            {
                Ok(_) => {
                    tx.send(true).unwrap();
                    println!("create ok");
                }
                Err(err) => {
                    tx.send(false).unwrap();
                    println!("create error {:?}", err)
                }
            }
        };
        s3fs.get_runtime().spawn(f_create);
        assert!(rx.recv().unwrap());

        // Test exists after create
        let fs = s3fs.clone();
        let (tx_exists_after, rx_exists_after) = tikv_util::mpsc::bounded(1);
        let f_exists_after = async move {
            match fs.exists(file_id, opts).await {
                Ok(exists) => {
                    tx_exists_after.send(exists).unwrap();
                }
                Err(err) => {
                    println!("exists after create error {:?}", err);
                    tx_exists_after.send(false).unwrap(); // Fail test on error
                }
            }
        };
        s3fs.get_runtime().spawn(f_exists_after);
        assert!(rx_exists_after.recv().unwrap());

        // Test size after create
        let fs = s3fs.clone();
        let (tx_size_after, rx_size_after) = tikv_util::mpsc::bounded(1);
        let expected_size = file_data.len() as u64;
        let f_size_after = async move {
            match fs.size(file_id, opts).await {
                Ok(size) => {
                    tx_size_after.send((true, size)).unwrap();
                }
                Err(_err) => {
                    tx_size_after.send((false, 0)).unwrap();
                }
            }
        };
        s3fs.get_runtime().spawn(f_size_after);
        let (success, size) = rx_size_after.recv().unwrap();
        assert!(success);
        assert_eq!(size, expected_size);

        let fs = s3fs.clone();
        let (tx, rx) = tikv_util::mpsc::bounded(1);
        let local_file = new_filename(file_id, local_dir.path());
        let move_local_file = local_file.clone();
        let f = async move {
            let opts = Options::default();
            match fs.read_file(file_id, opts).await {
                Ok(data) => {
                    let mut file = std::fs::File::create(&move_local_file).unwrap();
                    file.write_all(data.chunk()).unwrap();
                    tx.send(true).unwrap();
                    println!("prefetch ok");
                }
                Err(err) => {
                    tx.send(false).unwrap();
                    println!("prefetch failed {:?}", err)
                }
            }
        };
        s3fs.get_runtime().spawn(f);
        assert!(rx.recv().unwrap());
        let data = std::fs::read(&local_file).unwrap();
        assert_eq!(&data, &file_data);
        let file = LocalFile::open(file_id, local_file.clone()).unwrap();
        assert_eq!(file.size(), 8u64);
        assert_eq!(file.id(), file_id);
        let data = file.read(0, 8).unwrap();
        assert_eq!(&data, &file_data);
        let fs = s3fs.clone();
        let (tx, rx) = tikv_util::mpsc::bounded(1);
        let f = async move {
            fs.remove(file_id, None, Options::default()).await;
            tx.send(true).unwrap();
        };
        s3fs.get_runtime().spawn(f);
        assert!(rx.recv().unwrap());
        let _ = fs::remove_file(local_file);
    }

    #[test]
    fn test_s3_schema_file() {
        ::test_util::init_log_for_test();

        let local_dir = tempfile::tempdir().unwrap();
        let file_data = "abcdefgh".to_string().into_bytes();
        let s3fs = new_test_s3fs(&file_data);
        let (tx, rx) = tikv_util::mpsc::bounded(1);

        let fs = s3fs.clone();
        let file_data2 = file_data.clone();
        let opts = Options::default().with_type(FileType::Schema);
        let f = async move {
            match fs.create(1234, bytes::Bytes::from(file_data2), opts).await {
                Ok(_) => {
                    tx.send(true).unwrap();
                    println!("create ok");
                }
                Err(err) => {
                    tx.send(false).unwrap();
                    println!("create error {:?}", err)
                }
            }
        };
        s3fs.get_runtime().spawn(f);
        assert!(rx.recv().unwrap());
        let fs = s3fs.clone();
        let (tx, rx) = tikv_util::mpsc::bounded(1);
        let local_file = local_dir.path().join(format!("{:016x}.schema", 1234));
        let move_local_file = local_file.clone();
        let f = async move {
            match fs.read_file(1234, opts).await {
                Ok(data) => {
                    let mut file = std::fs::File::create(&move_local_file).unwrap();
                    file.write_all(data.chunk()).unwrap();
                    tx.send(true).unwrap();
                    println!("prefetch ok");
                }
                Err(err) => {
                    tx.send(false).unwrap();
                    println!("prefetch failed {:?}", err)
                }
            }
        };
        s3fs.get_runtime().spawn(f);
        assert!(rx.recv().unwrap());
        let data = std::fs::read(&local_file).unwrap();
        assert_eq!(&data, &file_data);
        let file = LocalFile::open(1234, local_file.clone()).unwrap();
        assert_eq!(file.size(), 8u64);
        assert_eq!(file.id(), 1234u64);
        let data = file.read(0, 8).unwrap();
        assert_eq!(&data, &file_data);
        let fs = s3fs.clone();
        let (tx, rx) = tikv_util::mpsc::bounded(1);
        let f = async move {
            fs.remove(1234, None, opts).await;
            tx.send(true).unwrap();
        };
        s3fs.get_runtime().spawn(f);
        assert!(rx.recv().unwrap());
        let _ = fs::remove_file(local_file);
    }

    // Ensure read_body_to_writer returns Err when the S3 body is shorter
    // than Content-Length so truncated responses never silently succeed.
    #[test]
    fn test_s3_truncated_body_should_error() {
        use crate::dfs::config::ConnOptions;

        ::test_util::init_log_for_test();

        let actual_body = "short";
        let claimed_length = "100"; // Content-Length says 100, but body is only 5 bytes

        // Build a mock that returns truncated GET response.
        let s3c = rusoto_core::Client::new_with(
            rusoto_mock::MockCredentialsProvider,
            rusoto_mock::MultipleMockRequestDispatcher::new(vec![
                rusoto_mock::MockRequestDispatcher::with_status(200) // GET: truncated
                    .with_header(http::header::CONTENT_LENGTH.as_str(), claimed_length)
                    .with_body(actual_body),
            ]),
        );
        // Use max_retry_count = 0 to avoid retry sleeps in test.
        let mut opts = ConnOptions::default();
        opts.max_retry_count = 0;
        let core = S3FsCore::new_with_s3_client(
            s3c,
            "".to_string(),
            "local".to_string(),
            "shard-db".to_string(),
            "prefix".to_string(),
            opts,
            false,
        );
        let s3fs = S3Fs {
            core: Arc::new(core),
        };

        // Read directly via get_object — should fail because body is truncated.
        let (tx, rx) = tikv_util::mpsc::bounded(1);
        let fs = s3fs.clone();
        s3fs.get_runtime().spawn(async move {
            let result = fs
                .get_object(
                    "prefix/test".to_string(),
                    "test.sst".to_string(),
                    GetObjectOptions::default(),
                )
                .await;
            tx.send(result).unwrap();
        });
        let result = rx.recv().unwrap();
        assert!(
            result.is_err(),
            "expected error for truncated S3 body, but got Ok with {} bytes",
            result.unwrap().len()
        );
    }

    #[test]
    fn test_tagging() {
        let tagging_deleted = Tagging::new_single_deleted();

        assert_eq!(
            tagging_deleted.to_xml(),
            "<Tagging><TagSet><Tag><Key>deleted</Key><Value>true</Value></Tag></TagSet></Tagging>"
        );
        assert_eq!(tagging_deleted.to_url_encoded(), "deleted=true");
        assert_eq!(
            tagging_deleted.tag_set.tag,
            Tagging::from_url_encoded("deleted=true").tag_set.tag
        );
    }

    #[test]
    fn test_complete_multipart_upload_to_xml() {
        let upload = CompleteMultipartUpload {
            parts: vec![
                CompletedPart {
                    part_number: 1,
                    e_tag: "\"etag-1\"".to_string(),
                    checksum_crc32: None,
                },
                CompletedPart {
                    part_number: 2,
                    e_tag: "\"etag-&-2\"".to_string(),
                    checksum_crc32: None,
                },
            ],
        };

        assert_eq!(
            upload.to_xml(),
            "<CompleteMultipartUpload><Part><PartNumber>1</PartNumber><ETag>&quot;etag-1&quot;</ETag></Part><Part><PartNumber>2</PartNumber><ETag>&quot;etag-&amp;-2&quot;</ETag></Part></CompleteMultipartUpload>"
        );

        let upload = CompleteMultipartUpload {
            parts: vec![CompletedPart {
                part_number: 1,
                e_tag: "\"etag-1\"".to_string(),
                checksum_crc32: Some(encode_checksum_crc32(0x01020304)),
            }],
        };

        assert_eq!(
            upload.to_xml(),
            "<CompleteMultipartUpload><Part><PartNumber>1</PartNumber><ETag>&quot;etag-1&quot;</ETag><ChecksumCRC32>AQIDBA==</ChecksumCRC32></Part></CompleteMultipartUpload>"
        );
    }

    fn make_unknown_error<T>(status: StatusCode) -> RusotoError<T> {
        RusotoError::Unknown(BufferedHttpResponse {
            status,
            body: bytes::Bytes::new(),
            headers: Default::default(),
        })
    }

    #[test]
    fn test_is_err_retryable_too_many_requests() {
        let s3fs = new_test_s3fs(b"test");

        // HTTP 429 (Too Many Requests) should be retryable.
        let err = make_unknown_error::<rusoto_s3::GetObjectError>(StatusCode::TOO_MANY_REQUESTS);
        assert!(s3fs.is_err_retryable(&err));

        // HTTP 500 (Internal Server Error) should be retryable.
        let err =
            make_unknown_error::<rusoto_s3::GetObjectError>(StatusCode::INTERNAL_SERVER_ERROR);
        assert!(s3fs.is_err_retryable(&err));

        // HTTP 503 (Service Unavailable) should be retryable.
        let err = make_unknown_error::<rusoto_s3::GetObjectError>(StatusCode::SERVICE_UNAVAILABLE);
        assert!(s3fs.is_err_retryable(&err));

        // HTTP 404 (Not Found) should NOT be retryable.
        let err = make_unknown_error::<rusoto_s3::GetObjectError>(StatusCode::NOT_FOUND);
        assert!(!s3fs.is_err_retryable(&err));

        // HTTP 400 (Bad Request) should NOT be retryable.
        let err = make_unknown_error::<rusoto_s3::GetObjectError>(StatusCode::BAD_REQUEST);
        assert!(!s3fs.is_err_retryable(&err));
    }

    #[test]
    fn test_is_multipart_upload_err_retryable() {
        let s3fs = new_test_s3fs(b"test");

        let err = make_unknown_error::<UploadPartError>(StatusCode::INTERNAL_SERVER_ERROR);
        assert!(s3fs.is_multipart_upload_err_retryable(&err));

        let err = make_unknown_error::<UploadPartError>(StatusCode::TOO_MANY_REQUESTS);
        assert!(s3fs.is_multipart_upload_err_retryable(&err));

        let err = make_unknown_error::<UploadPartError>(StatusCode::BAD_REQUEST);
        assert!(!s3fs.is_multipart_upload_err_retryable(&err));

        let err = RusotoError::<CreateMultipartUploadError>::ParseError("bad xml".to_string());
        assert!(!s3fs.is_multipart_upload_err_retryable(&err));

        let err = RusotoError::Service(AbortMultipartUploadError::NoSuchUpload(
            "missing upload".to_string(),
        ));
        assert!(!s3fs.is_multipart_upload_err_retryable(&err));
    }

    #[test]
    fn test_copy_object_200_with_error() {
        ::test_util::init_log_for_test();

        let error_xml = "\
            <?xml version=\"1.0\" encoding=\"UTF-8\"?>\
            <Error>\
                <Code>InternalError</Code>\
                <Message>We encountered an internal error. Please try again.</Message>\
            </Error>";
        let s3fs = crate::dfs::test_util::new_test_s3fs_copy_error(error_xml);

        // max_retry_count=2, so total sleep is 2^1*500ms + 2^2*500ms = 3s.
        let result = s3fs.get_runtime().block_on(async {
            s3fs.raw_copy_object("src-bucket/src-key", "dst-key", None, None)
                .await
        });
        assert!(result.is_err(), "should detect 200-with-Error");
        let err_msg = format!("{:?}", result.unwrap_err());
        assert!(
            err_msg.contains("without CopyObjectResult"),
            "error message should mention missing CopyObjectResult, got: {}",
            err_msg,
        );
    }

    #[test]
    fn test_copy_object_200_with_empty_body() {
        ::test_util::init_log_for_test();

        let s3fs = crate::dfs::test_util::new_test_s3fs_copy_error("");
        let result = s3fs.get_runtime().block_on(async {
            s3fs.raw_copy_object("src-bucket/src-key", "dst-key", None, None)
                .await
        });
        assert!(result.is_err(), "empty body should be treated as error");
        let err_msg = format!("{:?}", result.unwrap_err());
        assert!(
            err_msg.contains("without CopyObjectResult"),
            "error message should mention missing CopyObjectResult, got: {}",
            err_msg,
        );
    }

    #[test]
    fn test_complete_multipart_upload_200_with_error() {
        ::test_util::init_log_for_test();

        let error_xml = "\
            <?xml version=\"1.0\" encoding=\"UTF-8\"?>\
            <Error>\
                <Code>InternalError</Code>\
                <Message>We encountered an internal error. Please try again.</Message>\
            </Error>";
        let s3fs = crate::dfs::test_util::new_test_s3fs_complete_multipart_error(error_xml);

        let result = s3fs.get_runtime().block_on(async {
            s3fs.core
                .complete_multipart_upload(
                    "multipart-key",
                    "upload-id",
                    vec![MultipartUploadPart {
                        part_number: 1,
                        e_tag: "etag-1".to_owned(),
                        checksum_crc32: None,
                    }],
                )
                .await
        });
        assert!(
            result.is_err(),
            "should detect CompleteMultipartUpload 200-with-Error"
        );
        let err_msg = format!("{:?}", result.unwrap_err());
        assert!(
            err_msg.contains("without CompleteMultipartUploadResult"),
            "error message should mention missing CompleteMultipartUploadResult, got: {}",
            err_msg,
        );
    }

    #[test]
    fn test_complete_multipart_upload_200_with_error_does_not_match_body_text() {
        ::test_util::init_log_for_test();

        let error_xml = "\
            <?xml version=\"1.0\" encoding=\"UTF-8\"?>\
            <Error>\
                <!-- <CompleteMultipartUploadResult> is mentioned in an error comment. -->\
                <Code>InternalError</Code>\
                <Message>We encountered an internal error. Please try again.</Message>\
            </Error>";
        let s3fs = crate::dfs::test_util::new_test_s3fs_complete_multipart_error(error_xml);

        let result = s3fs.get_runtime().block_on(async {
            s3fs.core
                .complete_multipart_upload(
                    "multipart-key",
                    "upload-id",
                    vec![MultipartUploadPart {
                        part_number: 1,
                        e_tag: "etag-1".to_owned(),
                        checksum_crc32: None,
                    }],
                )
                .await
        });
        assert!(
            result.is_err(),
            "should detect CompleteMultipartUpload 200-with-Error"
        );
        let err_msg = format!("{:?}", result.unwrap_err());
        assert!(
            err_msg.contains("without CompleteMultipartUploadResult"),
            "error message should mention missing CompleteMultipartUploadResult, got: {}",
            err_msg,
        );
    }

    #[test]
    #[ignore = "requires DFS_S3_* environment and a real S3-compatible bucket"]
    fn test_s3_multipart_upload_with_env_config() {
        ::test_util::init_log_for_test();

        let mut config = Config::default();
        config.override_from_env();
        if config.s3_bucket.is_empty() {
            return;
        }
        config.conn_options.max_retry_count = 0;
        if config.prefix.is_empty() {
            config.prefix = "multipart-env-test".to_string();
        }

        let now_millis = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_millis();
        let key = format!("multipart-env-test/{now_millis}/object.bin");
        let abort_key = format!("multipart-env-test/{now_millis}/aborted.bin");
        let s3fs = S3Fs::new_from_config(config);
        let runtime = s3fs.get_runtime();

        let part1 = Bytes::from(vec![b'a'; 5 * 1024 * 1024]);
        let part2 = Bytes::from_static(b"tail");
        let upload_id = runtime.block_on(Dfs::begin_upload(&s3fs, &key)).unwrap();
        let uploaded_part1 = runtime
            .block_on(Dfs::upload_part(&s3fs, &key, &upload_id, 1, part1.clone()))
            .unwrap();
        let uploaded_part2 = runtime
            .block_on(Dfs::upload_part(&s3fs, &key, &upload_id, 2, part2.clone()))
            .unwrap();

        runtime
            .block_on(Dfs::complete_multipart_upload(
                &s3fs,
                &key,
                &upload_id,
                vec![uploaded_part1, uploaded_part2],
            ))
            .unwrap();

        let data = ObjectStorage::get_objects(
            &s3fs,
            vec![(key.clone(), GetObjectOptions::default())],
            None,
        )
        .unwrap()
        .pop()
        .unwrap()
        .1;
        assert_eq!(data.len(), part1.len() + part2.len());
        assert_eq!(&data[..part1.len()], part1.as_ref());
        assert_eq!(&data[part1.len()..], part2.as_ref());

        let full_key = format!("{}/{}", s3fs.get_prefix(), key);
        runtime
            .block_on(s3fs.delete_object(full_key, key.clone()))
            .unwrap();

        let abort_upload_id = runtime
            .block_on(Dfs::begin_upload(&s3fs, &abort_key))
            .unwrap();
        let abort_part = runtime
            .block_on(Dfs::upload_part(
                &s3fs,
                &abort_key,
                &abort_upload_id,
                1,
                part1,
            ))
            .unwrap();
        assert!(!abort_part.e_tag.is_empty());
        runtime
            .block_on(Dfs::abort_multipart_upload(
                &s3fs,
                &abort_key,
                &abort_upload_id,
            ))
            .unwrap();
    }

    #[test]
    fn test_cloud_provider_from_endpoint() {
        assert_eq!(
            CloudProvider::from_endpoint("https://s3.us-west-2.amazonaws.com"),
            CloudProvider::Aws
        );
        assert_eq!(
            CloudProvider::from_endpoint("https://oss-cn-beijing-internal.aliyuncs.com"),
            CloudProvider::Aliyun
        );
        assert_eq!(
            CloudProvider::from_endpoint("https://cos.ap-beijing.myqcloud.com"),
            CloudProvider::Tencentcloud
        );
        // Volcengine public and internal TOS endpoints.
        assert_eq!(
            CloudProvider::from_endpoint("https://tos-s3-cn-beijing.volces.com"),
            CloudProvider::Volcengine
        );
        assert_eq!(
            CloudProvider::from_endpoint("https://tos-s3-cn-beijing.ivolces.com"),
            CloudProvider::Volcengine
        );
        assert_eq!(
            CloudProvider::from_endpoint("http://127.0.0.1:9000"),
            CloudProvider::Unknown
        );
    }

    #[test]
    fn test_volcengine_storage_class_is_always_standard() {
        let provider = CloudProvider::Volcengine;
        for storage_class in [
            StorageClass::Standard,
            StorageClass::InfrequentAccess,
            StorageClass::GlacierInstantRetrieval,
            StorageClass::IntelligentTiering,
            StorageClass::Default,
        ] {
            assert_eq!(
                provider.storage_class_str(storage_class),
                STORAGE_CLASS_STANDARD
            );
        }
    }

    #[test]
    fn test_volcengine_add_provider_header_normalizes_storage_class() {
        let provider = CloudProvider::Volcengine;
        let region = rusoto_core::Region::Custom {
            name: "cn-beijing".to_owned(),
            endpoint: "https://tos-s3-cn-beijing.volces.com".to_owned(),
        };
        let mut req = SignedRequest::new("PUT", "s3", &region, "/bucket/key");
        // Even if the caller asks for a non-standard class, Volcengine requests
        // must carry x-amz-storage-class: STANDARD.
        provider.add_provider_header(&mut req, "x-amz-storage-class", STORAGE_CLASS_STANDARD_IA);
        let values = req.headers().get("x-amz-storage-class").unwrap();
        assert_eq!(values.len(), 1);
        assert_eq!(values[0], STORAGE_CLASS_STANDARD.as_bytes());
    }
}