use std::{
    collections::HashMap,
    sync::{
        Arc, Mutex,
        atomic::{AtomicU64, Ordering},
    },
    time::{Duration, SystemTime},
};

use anyhow::Result;
use log::{info, warn};
use scopeguard::defer;
use tici::{
    HandleRange, RangeBound,
    config::{MetaServiceConfig, make_default_test_meta_config, make_default_test_worker_config},
    meta::reader::reader_pool::warmup::{WarmupTestAction, WarmupTestController},
    server_util::print_cpu,
    worker::mocks::midpoint_bytes,
};
use tici_common::index::IndexKey;
use tici_grpc::ErrorCode;
use tici_system::readable::{ReadableDuration, ReadableSize};
use tici_test_utils::test_utils::{
    reader_handle::{ReaderNodeHandle, make_default_reader_config},
    test_suite::E2ETestSuite,
    *,
};
use tokio::sync::RwLock;

use crate::sequential::Workload;

#[test]
fn test_append() -> Result<()> {
    tici::status_server::profile::set_prof_active(true).unwrap();
    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(4)
        .enable_all()
        .build()?;
    fail::cfg("worker_service_mock_meta_service_url", "return").unwrap();
    defer!({
        fail::remove("worker_service_mock_meta_service_url");
    });

    let mut cfg = make_default_test_meta_config(true);
    cfg.shard.auto_compaction = true;
    cfg.shard.enable_split = true;
    cfg.shard.enable_merge = true;
    cfg.shard.max_size = ReadableSize::gb(4);
    cfg.shard.split.split_threshold = 0.75;

    let test_suite = rt.block_on(async {
        Arc::new(RwLock::new(
            E2ETestSuite::new_with_mock_cfg(cfg, make_default_test_worker_config(false))
                .await
                .unwrap(),
        ))
    });
    let workload = Workload::new(
        IndexKey::new_nullspace(1, 1001),
        50 * 1024 * 1024,
        10 * 1024,
    );
    let workload_clone = workload.clone();
    rt.block_on(async {
        info!("Boot workload {:?}", &workload);
        test_suite
            .write()
            .await
            .add_mocked_reader_node(
                "127.0.0.1:8995".to_string(),
                "127.0.0.1:8995".to_string(),
                Some(Duration::from_millis(500)),
            )
            .await;

        let _ = workload.boot(test_suite.clone()).await;
        must_finally(async || {
            test_suite
                .read()
                .await
                .check_get_local_cache_info(&workload.index_key)
                .await
                .is_ok()
        })
        .await;

        info!("Run workload {:?}", &workload);
        let addr = test_suite.read().await.get_one_worker_node_addr();

        let mut append_handles = vec![];
        for para in 0..workload.n_parallel {
            let suite = test_suite.clone();
            let workload = workload.clone();
            append_handles.push(tokio::spawn(async move {
                let runc = if para == 0 {
                    workload.n_default / workload.n_parallel
                        + workload.n_default % workload.n_parallel
                } else {
                    workload.n_default / workload.n_parallel
                };
                for i in 0..runc {
                    tokio::time::sleep(std::time::Duration::from_millis(1)).await;
                    workload.append(suite.clone(), para, i).await;
                }
            }));
        }

        for handle in append_handles.drain(..) {
            handle.await.unwrap();
        }

        workload_clone
            .stop_and_print_result(&mut *test_suite.write().await, addr, None)
            .await;
    });

    workload_clone.print_heap_size();

    rt.block_on(async {
        {
            let check = tici_test_utils::test_utils::must_finally_ex({
                let test_suite = test_suite.clone();
                let index_key = workload.index_key.clone();
                move |_| {
                    let test_suite = test_suite.clone();
                    let index_key = index_key.clone();
                    async move {
                        tici_test_utils::test_utils::check_reader_writer_manifest_consistent(
                            test_suite, index_key,
                        )
                        .await
                    }
                }
            })
            .await;
            if let either::Either::Right(info) = check {
                panic!("{info}");
            }
        }
        test_suite.write().await.stop_meta().await.unwrap();
    });
    Ok(())
}

#[test]
fn test_skewed_append() -> Result<()> {
    print_cpu();
    tici::status_server::profile::set_prof_active(true).unwrap();
    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(4)
        .enable_all()
        .build()?;
    fail::cfg("worker_service_mock_meta_service_url", "return").unwrap();
    defer!({
        fail::remove("worker_service_mock_meta_service_url");
    });

    let mut cfg = make_default_test_meta_config(true);
    cfg.shard.auto_compaction = true;
    cfg.shard.enable_split = true;
    cfg.shard.enable_merge = true;
    cfg.shard.max_size = ReadableSize::mb(128);
    cfg.shard.split.split_threshold = 0.75;
    cfg.shard.writer_scheduling_strategy = "balanced-count".to_string();

    let test_suite = rt.block_on(async {
        Arc::new(RwLock::new(
            E2ETestSuite::new_with_mock_cfg(cfg, make_default_test_worker_config(false))
                .await
                .unwrap(),
        ))
    });
    let workload = Workload::new(
        IndexKey::new_nullspace(1, 1001),
        50 * 1024 * 1024,
        10 * 1024,
    );
    let workload_clone = workload.clone();

    let op_counter = Arc::new(AtomicU64::new(0));
    let done_counter = Arc::new(AtomicU64::new(0));
    let shard_hits: Arc<Mutex<HashMap<u64, u64>>> = Arc::new(Mutex::new(HashMap::new()));
    let retry_stats: Arc<Mutex<HashMap<i32, u64>>> = Arc::new(Mutex::new(HashMap::new()));
    let secondary_worker_addr = "127.0.0.1:8512".to_string();

    rt.block_on(async {
        info!("boot skewed workload {:?}", &workload);
        test_suite
            .write()
            .await
            .add_mocked_worker_node(Some(secondary_worker_addr.clone()))
            .await;
        let _ = workload.boot(test_suite.clone()).await;
        test_suite.write().await.start = SystemTime::now();

        info!("run skewed workload {:?}", &workload);
        let primary_worker_addr = test_suite.read().await.get_worker_node_template_addr();
        let addr = primary_worker_addr.clone();

        let mut append_handles = vec![];
        for para in 0..workload.n_parallel {
            let suite = test_suite.clone();
            let workload = workload.clone();
            let op_counter = op_counter.clone();
            let done_counter = done_counter.clone();
            let shard_hits = shard_hits.clone();
            let retry_stats = retry_stats.clone();
            append_handles.push(tokio::spawn(async move {
                let runc = if para == 0 {
                    workload.n_default / workload.n_parallel
                        + workload.n_default % workload.n_parallel
                } else {
                    workload.n_default / workload.n_parallel
                };
                for _ in 0..runc {
                    tokio::time::sleep(std::time::Duration::from_millis(1)).await;
                    let op = op_counter.fetch_add(1, Ordering::Relaxed);
                    let shard_id = workload
                        .append_skewed(suite.clone(), para, op, retry_stats.clone())
                        .await;
                    {
                        let mut guard = shard_hits.lock().unwrap();
                        *guard.entry(shard_id).or_insert(0) += 1;
                    }

                    let appended = done_counter.fetch_add(1, Ordering::Relaxed) + 1;
                    if appended % 100 == 0 {
                        let (shards_cnt, p50, p90, p95, p99, p100) =
                            collect_skewed_append_shard_size_stats(
                                suite.clone(),
                                &workload.index_key,
                            )
                            .await;
                        warn!(
                            "skewed append progress: appended={appended}, shards={shards_cnt}, shard_size_mib(p50={:.3},p90={:.3},p95={:.3},p99={:.3},p100={:.3})",
                            ReadableSize(p50).as_mb_f64(),
                            ReadableSize(p90).as_mb_f64(),
                            ReadableSize(p95).as_mb_f64(),
                            ReadableSize(p99).as_mb_f64(),
                            ReadableSize(p100).as_mb_f64(),
                        );
                    }
                }
            }));
        }

        for handle in append_handles.drain(..) {
            handle.await.unwrap();
        }

        let mut hits = shard_hits
            .lock()
            .unwrap()
            .clone()
            .into_iter()
            .collect::<Vec<_>>();
        hits.sort_by(|a, b| b.1.cmp(&a.1));
        let total_ops = hits.iter().map(|(_, v)| *v).sum::<u64>();
        info!(
            "skewed append shard_hits(top10)={:?}, total_ops={}",
            hits.iter().take(10).collect::<Vec<_>>(),
            total_ops
        );
        let retry_stats_snapshot = {
            let guard = retry_stats.lock().unwrap_or_else(|e| e.into_inner());
            guard.clone()
        };
        if !retry_stats_snapshot.is_empty() {
            let mut stats = retry_stats_snapshot.into_iter().collect::<Vec<_>>();
            stats.sort_by(|a, b| b.1.cmp(&a.1));
            let formatted = stats
                .into_iter()
                .map(|(status, count)| {
                    format!("{}({})={}", error_code_name(status), status, count)
                })
                .collect::<Vec<_>>();
            info!("skewed append retry stats: {:?}", formatted);
        }

        let (shards_cnt, _p50, _p90, _p95, _p99, p100) = collect_skewed_append_shard_size_stats(
            test_suite.clone(),
            &workload.index_key,
        ).await;
        // TODO(P1) Solve this imbalance.
        assert!(
            p100 > 10,
            "expected p100 (max shard size) > 10, got {} for {} shards",
            p100,
            shards_cnt
        );
        let shards_cnt = test_suite
            .read()
            .await
            .collect_shards_from_index(&workload.index_key)
            .await
            .len() as u64;
        info!("skewed append shards_cnt={shards_cnt}");

        must_finally(async || {
            let suite = test_suite.read().await;
            let meta = suite.get_meta_service();
            let primary_worker = match meta
                .worker_node_mgr
                .get_worker_node(&primary_worker_addr)
                .await
            {
                Some(worker) => worker,
                None => return false,
            };
            let secondary_worker = match meta
                .worker_node_mgr
                .get_worker_node(&secondary_worker_addr)
                .await
            {
                Some(worker) => worker,
                None => return false,
            };
            let primary_count = primary_worker.shards_len().await;
            let secondary_count = secondary_worker.shards_len().await;
            let diff = if primary_count > secondary_count {
                primary_count - secondary_count
            } else {
                secondary_count - primary_count
            };
            info!("shard count diff: {}-{}={}", primary_count, secondary_count, diff);
            diff <= 3
        })
        .await;

        workload_clone
            .stop_and_print_result(&mut *test_suite.write().await, addr, None)
            .await;

        Ok::<(), anyhow::Error>(())
    })?;

    rt.block_on(async { test_suite.write().await.stop_meta().await })?;
    Ok(())
}

async fn collect_skewed_append_shard_size_stats(
    suite: Arc<RwLock<E2ETestSuite>>,
    index_key: &IndexKey,
) -> (usize, u64, u64, u64, u64, u64) {
    let suite = suite.read().await;
    let mut sizes = suite
        .collect_shards_from_index(index_key)
        .await
        .into_iter()
        .map(|s| s.get_total_estimated_bytes())
        .collect::<Vec<_>>();
    sizes.sort_unstable();
    if sizes.is_empty() {
        (0usize, 0u64, 0u64, 0u64, 0u64, 0u64)
    } else {
        (
            sizes.len(),
            percentile_nearest_rank(&sizes, 50),
            percentile_nearest_rank(&sizes, 90),
            percentile_nearest_rank(&sizes, 95),
            percentile_nearest_rank(&sizes, 99),
            percentile_nearest_rank(&sizes, 100),
        )
    }
}

fn percentile_nearest_rank(sorted: &[u64], p: u64) -> u64 {
    debug_assert!(!sorted.is_empty());
    let n = sorted.len() as u64;
    let k = (p * n).div_ceil(100);
    let k = k.clamp(1, n);
    sorted[(k - 1) as usize]
}

fn error_code_name(code: i32) -> &'static str {
    match code {
        x if x == ErrorCode::Success as i32 => "SUCCESS",
        x if x == ErrorCode::UnknownError as i32 => "UNKNOWN_ERROR",
        x if x == ErrorCode::InvalidArguments as i32 => "INVALID_ARGUMENTS",
        x if x == ErrorCode::TryAgain as i32 => "TRY_AGAIN",
        x if x == ErrorCode::Deprecated as i32 => "DEPRECATED",
        x if x == ErrorCode::ShardNotFound as i32 => "SHARD_NOT_FOUND",
        x if x == ErrorCode::IndexNotFound as i32 => "INDEX_NOT_FOUND",
        x if x == ErrorCode::WorkerNotFound as i32 => "WORKER_NOT_FOUND",
        x if x == ErrorCode::ShardNotScheduled as i32 => "SHARD_NOT_SCHEDULED",
        x if x == ErrorCode::CompactionNotFound as i32 => "COMPACTION_NOT_FOUND",
        x if x == ErrorCode::FragRangeMismatch as i32 => "FRAG_RANGE_MISMATCH",
        x if x == ErrorCode::InvalidOp as i32 => "INVALID_OP",
        x if x == ErrorCode::Throttled as i32 => "THROTTLED",
        _ => "UNKNOWN_STATUS",
    }
}

#[test]
fn test_many_shards() -> Result<()> {
    print_cpu();
    tici::status_server::profile::set_prof_active(true).unwrap();
    let workload = Workload::new(
        IndexKey::new_nullspace(1, 1001),
        50 * 1024 * 1024,
        10 * 1024,
    );
    let mut cfg = make_default_test_meta_config(true);
    cfg.shard.auto_compaction = true;
    cfg.shard.enable_split = false;
    cfg.shard.enable_merge = false;
    cfg.shard.max_size = ReadableSize::gb(4);
    cfg.shard.split.split_threshold = 0.75;
    cfg.shard.shard_scheduler_runner_cnt = workload.n_runner;
    if workload.update_delay_ms > 0 {
        log::info!("mock delay");
        fail::cfg(
            "tidb_backend_mock_delayed_update",
            &format!("return({})", workload.update_delay_ms),
        )
        .unwrap();
    }
    test_many_shards_impl(workload, cfg, None, true)
}

fn test_many_shards_impl(
    workload: Workload,
    cfg: MetaServiceConfig,
    output: Option<&mut std::fs::File>,
    print_heap_size: bool,
) -> Result<()> {
    let worker_thread_cnt = std::env::var("W")
        .map(|v: String| v.parse().unwrap())
        .unwrap_or(4usize);
    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(worker_thread_cnt)
        .enable_all()
        .build()?;
    fail::cfg("worker_service_mock_meta_service_url", "return").unwrap();
    defer!({
        fail::remove("worker_service_mock_meta_service_url");
    });

    let test_suite = rt.block_on(async {
        Arc::new(RwLock::new(
            E2ETestSuite::new_with_mock_cfg(cfg, make_default_test_worker_config(false))
                .await
                .unwrap(),
        ))
    });
    let workload_clone = workload.clone();

    rt.block_on(async {
        info!("Boot workload {:?}", &workload);

        let _ = workload.boot(test_suite.clone()).await;
        let addr = test_suite.read().await.get_one_worker_node_addr();

        info!("Run workload {:?}", &workload);
        if workload.n_presplit > 1 {
            let mut v = vec![];
            for split in 0..workload.n_presplit {
                v.push(HandleRange::new(
                    format!("A{:06}", split).into(),
                    format!("A{:06}", split + 1).into(),
                ));
            }
            let original_shard = test_suite.read().await.get_first_shard().await;
            let _ = test_suite
                .read()
                .await
                .must_manual_split_shard(
                    &IndexKey::new_nullspace(1, 1001),
                    original_shard.range.sortable_start(),
                    v,
                    0,
                )
                .await
                .unwrap();
        }

        // Presplit cost time.
        test_suite.write().await.start = SystemTime::now();

        let mut append_handles = vec![];
        for para in 0..workload.n_parallel {
            let suite = test_suite.clone();
            let workload = workload.clone();
            append_handles.push(tokio::spawn(async move {
                let runc = if para == 0 {
                    workload.n_default / workload.n_parallel
                        + workload.n_default % workload.n_parallel
                } else {
                    workload.n_default / workload.n_parallel
                };
                for i in 0..runc {
                    tokio::time::sleep(std::time::Duration::from_millis(1)).await;
                    workload.append(suite.clone(), para, i).await;
                }
            }));
        }

        for handle in append_handles.drain(..) {
            handle.await.unwrap();
        }

        workload_clone
            .stop_and_print_result(&mut *test_suite.write().await, addr, output)
            .await;
    });

    if print_heap_size {
        workload_clone.print_heap_size();
    }

    rt.block_on(async {
        test_suite.write().await.stop_meta().await.unwrap();
    });
    Ok(())
}

#[test]
fn test_bulk_split() -> Result<()> {
    print_cpu();
    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(4)
        .enable_all()
        .build()?;
    fail::cfg("worker_service_mock_meta_service_url", "return").unwrap();
    defer!({
        fail::remove("worker_service_mock_meta_service_url");
    });
    let mut cfg = make_default_test_meta_config(true);
    cfg.shard.auto_compaction = true;
    cfg.shard.enable_split = false;
    cfg.shard.enable_merge = false;
    cfg.shard.max_size = ReadableSize::gb(4);
    cfg.shard.split.split_threshold = 0.75;
    cfg.shard.enable_patrol();
    cfg.shard.check_dead_shard_tick = 1;
    cfg.worker_node.worker_node_down_timeout = ReadableDuration::secs(2);
    cfg.worker_node.health_check_interval = ReadableDuration::secs(1);
    cfg.cache_scheduling.allocate_parallelism = 1000;

    let n_db_batch: usize = std::env::var("TEST_DB_BATCH")
        .map(|v| v.parse().unwrap())
        .unwrap_or(1usize);
    let test_suite = rt.block_on(async {
        Arc::new(RwLock::new(
            E2ETestSuite::new_with_mock_cfg(cfg, make_default_test_worker_config(true))
                .await
                .unwrap(),
        ))
    });
    rt.block_on(async {
        let warmup_controller = Arc::new(WarmupTestController::new());
        warmup_controller
            .set_fallback_action(Some(WarmupTestAction::SucceedAfter(Duration::from_millis(
                1000,
            ))))
            .await;
        test_suite
            .write()
            .await
            .get_meta_service()
            .get_reader_pool()
            .warmup_mgr()
            .set_test_controller(Some(warmup_controller))
            .await;
    });
    let workload = Workload::new(
        IndexKey::new_nullspace(1, 1001),
        50 * 1024 * 1024,
        10 * 1024,
    );

    rt.block_on(async {
        info!("Boot workload {:?}", &workload);

        let init_shard = workload.boot(test_suite.clone()).await;

        let searchlib_config = make_default_reader_config(
            &test_suite
                .read()
                .await
                .get_meta_service()
                .config
                .load_full(),
            "127.0.0.1:19410".to_string(),
            "127.0.0.1:19411".to_string(),
        );
        let mut reader_handle = ReaderNodeHandle::new(searchlib_config).await;

        info!("Run workload {:?}", &workload);
        let mut v = vec![];

        for split in 0..workload.n_presplit {
            let start_key = if split == 0 {
                init_shard.read().await.range.encoded_start()
            } else {
                let mut start_key = init_shard.read().await.range.encoded_start();
                let t: Vec<u8> = format!("A{:06}", split).into();
                start_key.extend_from_slice(t.as_slice());
                start_key
            };
            let end_key = if split == workload.n_presplit - 1 {
                init_shard.read().await.range.encoded_end()
            } else {
                let mut end_key = init_shard.read().await.range.encoded_start();
                let t: Vec<u8> = format!("A{:06}", split + 1).into();
                end_key.extend_from_slice(t.as_slice());
                end_key
            };
            v.push(HandleRange::new(start_key, end_key));
        }
        {
            let suite = test_suite.read().await;
            let original_shard = suite.get_first_shard().await;
            let start = SystemTime::now();
            // Note that if we call `debug_get_shard_manifest`, it can get a newer shard,
            // thus causing warmup failed. However, this case won't let the test
            // fail, because it can then return newer warmed shards.
            let _ = suite
                .must_manual_split_shard(
                    &workload.index_key,
                    original_shard.range.sortable_start(),
                    v,
                    n_db_batch,
                )
                .await
                .unwrap();

            must_finally(async || {
                test_suite
                    .read()
                    .await
                    .check_get_local_cache_info(&workload.index_key)
                    .await
                    .is_ok()
            })
            .await;

            log::info!(
                "split elapsed={:?}",
                SystemTime::now().duration_since(start)
            );

            // Mock failure
            suite
                .get_worker_node_handle("127.0.0.1:8510")
                .shard_write_manager
                .as_mock()
                .skip_heartbeat
                .store(true, std::sync::atomic::Ordering::Relaxed);
        }

        tokio::time::sleep(std::time::Duration::from_secs(2 * 2)).await;

        // Old worknode evicted
        assert_eq!(
            test_suite
                .read()
                .await
                .get_meta_service()
                .worker_node_mgr
                .get_worker_nodes_len()
                .await,
            0
        );

        tokio::time::sleep(std::time::Duration::from_secs(2)).await;

        test_suite
            .read()
            .await
            .get_worker_node_handle("127.0.0.1:8510")
            .shard_write_manager
            .as_mock()
            .skip_heartbeat
            .store(false, std::sync::atomic::Ordering::Relaxed);

        tokio::time::sleep(std::time::Duration::from_secs(2 * 2)).await;

        {
            let suite = test_suite.read().await;
            assert_eq!(
                suite.collect_shards_from_memory().await.len(),
                workload.n_presplit as usize
            );
            assert_eq!(
                suite
                    .get_meta_service()
                    .worker_node_mgr
                    .get_worker_nodes_len()
                    .await,
                1
            );
            assert_eq!(
                suite
                    .get_meta_service()
                    .worker_node_mgr
                    .get_worker_node("127.0.0.1:8510")
                    .await
                    .unwrap()
                    .shards_len()
                    .await,
                workload.n_presplit as usize
            );

            assert_eq!(
                suite
                    .get_worker_node_handle("127.0.0.1:8510")
                    .worker_node_service
                    .shard_count(),
                workload.n_presplit as usize
            );

            must_finally(async || {
                test_suite
                    .read()
                    .await
                    .check_get_local_cache_info(&workload.index_key)
                    .await
                    .is_ok()
            })
            .await;
        }

        reader_handle.stop().await;
        test_suite.write().await.stop_worker().await.unwrap();
        test_suite.write().await.stop_meta().await.unwrap();
    });
    Ok(())
}

#[test]
fn test_random_split() -> Result<()> {
    print_cpu();
    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(4)
        .enable_all()
        .build()?;
    fail::cfg("worker_service_mock_meta_service_url", "return").unwrap();
    defer!({
        fail::remove("worker_service_mock_meta_service_url");
    });
    let mut cfg = make_default_test_meta_config(true);
    cfg.shard.auto_compaction = true;
    cfg.shard.enable_split = false;
    cfg.shard.enable_merge = false;
    cfg.shard.max_size = ReadableSize::gb(4);
    cfg.shard.split.split_threshold = 0.75;

    let n_db_batch: usize = std::env::var("TEST_DB_BATCH")
        .map(|v| v.parse().unwrap())
        .unwrap_or(1usize);
    let test_suite = rt.block_on(async {
        Arc::new(RwLock::new(
            E2ETestSuite::new_with_mock_cfg(cfg, make_default_test_worker_config(true))
                .await
                .unwrap(),
        ))
    });
    let mut workload = Workload::new(
        IndexKey::new_nullspace(1, 1001),
        50 * 1024 * 1024,
        10 * 1024,
    );
    workload.n_presplit = std::cmp::max(workload.n_presplit, 10);

    rt.block_on(async {
        info!("Boot workload {:?}", &workload);

        let _ = workload.boot(test_suite.clone()).await;

        info!("Run workload {:?}", &workload);
        let start = SystemTime::now();

        let searchlib_config = make_default_reader_config(
            &test_suite
                .read()
                .await
                .get_meta_service()
                .config
                .load_full(),
            "127.0.0.1:19410".to_string(),
            "127.0.0.1:19411".to_string(),
        );
        let mut reader_handle = ReaderNodeHandle::new(searchlib_config).await;

        info!("Prepare split test {:?}", &workload);
        for _ in 0..workload.n_presplit {
            let shard = test_suite.read().await.get_random_shard().await;
            let mid = midpoint_bytes(
                shard.range.encoded_start().as_slice(),
                shard.range.encoded_end().as_slice(),
            );
            let new_ranges = vec![
                HandleRange::new_from_sorted(
                    shard.range.sortable_start(),
                    RangeBound::make_end(mid.clone()),
                ),
                HandleRange::new_from_sorted(
                    RangeBound::make_start(mid.clone()),
                    shard.range.sortable_end(),
                ),
            ];
            let _ = test_suite
                .read()
                .await
                .must_manual_split_shard(
                    &workload.index_key,
                    shard.range.sortable_start(),
                    new_ranges,
                    n_db_batch,
                )
                .await
                .unwrap();

            must_finally(async || {
                test_suite
                    .read()
                    .await
                    .check_get_local_cache_info(&workload.index_key)
                    .await
                    .is_ok()
            })
            .await;

            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
        }

        reader_handle.stop().await;

        log::info!(
            "split elapsed={:?}",
            SystemTime::now().duration_since(start)
        );

        test_suite.write().await.stop_worker().await.unwrap();
        test_suite.write().await.stop_meta().await.unwrap();
    });
    Ok(())
}