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

use std::{collections::HashMap, fs, path::Path, time::Duration};

use collections::HashSet;
use futures::executor::block_on;
use kvengine::{
    ShardStats, new_schema_filename, new_tmp_filename, new_vector_index_filename,
    table::{get_local_dir, sstable::new_filename},
};
use kvproto::raft_cmdpb::{RaftCmdRequest, RaftRequestHeader};
use pd_client::PdClient;
use rand::Rng;
use rfstore::store::rlog::*;
use test_cloud_server::{
    MajorCompactionTarget, ServerCluster, client::RequestOptions, keyspace::CreateKeyspaceOptions,
    must_wait,
};
use tikv_util::{
    config::{ReadableDuration, ReadableSize},
    info,
    time::Instant,
};
use uuid::Uuid;

use crate::{alloc_node_id, generate_keyspace_key};

#[test]
fn test_local_file_gc() {
    let node_id = alloc_node_id();
    let mut cluster = ServerCluster::new(vec![node_id], |_, cfg| {
        cfg.raft_store.local_file_gc_timeout = ReadableDuration::secs(1);
        cfg.raft_store.local_file_gc_tick_interval = ReadableDuration::millis(300);
    });
    let kv = cluster.get_kvengine(node_id);
    let mut client = cluster.new_client();
    let mut new_file_paths = vec![];
    let mut new_tmp_file_paths = vec![];
    let mut new_schema_file_paths = vec![];
    let mut new_vector_index_file_paths = vec![];
    let mut new_txn_file_paths = vec![];
    let mut new_txn_tmp_file_paths = vec![];
    for local_dir in &kv.opts.local_dirs {
        info!("local_dir {}", local_dir.to_string_lossy());
        let new_file_id = client.get_ts().into_inner();
        let new_file_path = new_filename(new_file_id, local_dir.as_path());
        fs::write(&new_file_path, "abc").unwrap();
        new_file_paths.push(new_file_path);
        let new_tmp_file_path = local_dir.join(new_tmp_filename(new_file_id, 1));
        fs::write(&new_tmp_file_path, "def").unwrap();
        new_tmp_file_paths.push(new_tmp_file_path);
        let new_file_id = client.get_ts().into_inner();
        let new_schema_file_path = local_dir.join(new_schema_filename(new_file_id));
        fs::write(&new_schema_file_path, "jkl").unwrap();
        new_schema_file_paths.push(new_schema_file_path);
        let new_file_id = client.get_ts().into_inner();
        let new_vector_index_file_path = local_dir.join(new_vector_index_filename(new_file_id));
        fs::write(&new_vector_index_file_path, "mno").unwrap();
        new_vector_index_file_paths.push(new_vector_index_file_path);

        let new_file_id = client.get_ts().into_inner();
        // Txn chunk files are sharded by file id across local dirs. Use the same
        // mapping as runtime code so GC removes the exact file we inject.
        let txn_local_dir = get_local_dir(&kv.opts.local_dirs, new_file_id);
        let txn_dir = txn_local_dir.join("txn");
        fs::create_dir_all(&txn_dir).unwrap();
        let new_txn_file_path = txn_dir.join(format!("{:016x}.txn", new_file_id));
        fs::write(&new_txn_file_path, "pqr").unwrap();
        new_txn_file_paths.push(new_txn_file_path);
        // Create txn tmp file
        let new_txn_tmp_file_path = txn_dir.join(format!("{:016x}.1.tmp", new_file_id));
        fs::write(&new_txn_tmp_file_path, "txn_tmp").unwrap();
        new_txn_tmp_file_paths.push(new_txn_tmp_file_path);
    }
    // Create importer file
    let config = cluster.get_node_config(node_id);
    let import_dir = Path::new(&config.storage.data_dir).join("import");
    fs::create_dir_all(&import_dir).unwrap();
    let uuid = Uuid::new_v4();
    let new_importer_file_path = import_dir.join(format!("{}_{}_{}_{}_default.sst", uuid, 1, 1, 1));
    fs::write(&new_importer_file_path, "importer_sst_data").unwrap();

    cluster.wait_pd_region_count(1);
    client.put_kv(0..1000, gen_key, gen_val);
    client.split(&gen_key(500));
    let shard_ids = kv.get_all_shard_id_vers();
    assert_eq!(shard_ids.len(), 2);
    let mut all_files = HashSet::default();
    for _ in 0..10 {
        for shard_id in &shard_ids {
            let snap = kv.get_snap_access(shard_id.id).unwrap();
            all_files.extend(snap.get_all_sst_files());
        }
        if !all_files.is_empty() {
            break;
        }
        std::thread::sleep(Duration::from_secs(1));
    }
    assert!(!all_files.is_empty());
    for &file_id in &all_files {
        let local_dir = get_local_dir(&kv.opts.local_dirs, file_id);
        let file_path = new_filename(file_id, local_dir.as_path());
        assert!(file_path.exists());
    }
    // Collect existing files.
    let exist_files = || -> Option<Vec<String>> {
        let mut files = vec![];
        for ff in [
            &new_file_paths,
            &new_tmp_file_paths,
            &new_schema_file_paths,
            &new_vector_index_file_paths,
            &new_txn_file_paths,
            &new_txn_tmp_file_paths,
        ] {
            files.extend(
                ff.iter()
                    .filter(|path| path.exists())
                    .map(|path| path.to_string_lossy().to_string()),
            );
        }
        if new_importer_file_path.exists() {
            files.push(new_importer_file_path.to_string_lossy().to_string());
        }
        if files.is_empty() { None } else { Some(files) }
    };
    for _ in 0..30 {
        if exist_files().is_none() {
            break;
        }
        std::thread::sleep(Duration::from_secs(1));
    }
    assert_eq!(None, exist_files());
    cluster.stop();
}

fn gen_key(i: usize) -> Vec<u8> {
    format!("xkey_{:08}", i).into_bytes()
}

fn gen_val(i: usize) -> Vec<u8> {
    format!("val_{:08}", i).into_bytes()
}

fn gen_val2(i: usize) -> Vec<u8> {
    format!("val2_{:08}", i).into_bytes()
}

#[test]
fn test_raft_log_gc() {
    let mut node_ids = (0..3).map(|_| alloc_node_id()).collect::<Vec<_>>();
    let mut cluster = ServerCluster::new(node_ids.clone(), |_, cfg| {
        cfg.raft_store.raft_base_tick_interval = ReadableDuration::millis(100);
        cfg.raft_store.raft_store_max_leader_lease = ReadableDuration::millis(50);
        cfg.raft_store.raft_log_gc_tick_interval = ReadableDuration::millis(100);
        cfg.raft_store.pd_heartbeat_tick_interval = ReadableDuration::millis(100);
        cfg.raft_store.hibernate_regions = false;
        cfg.raft_store.max_peer_down_duration = ReadableDuration::secs(2);
        cfg.rocksdb.writecf.write_buffer_size = ReadableSize::kb(16);
    });
    cluster.wait_region_replicated(&[], 3);
    let mut client = cluster.new_client();

    let region_id = client.get_region_id(&[]);
    let mut peer_ids = HashMap::new();
    for &node_id in &node_ids {
        let store_id = cluster.get_store_id(node_id);
        let peer_id = client.get_peer_id(&[], store_id);
        peer_ids.insert(node_id, peer_id);
    }
    let before_truncated_idxes = node_ids
        .iter()
        .map(|id| {
            let &peer_id = peer_ids.get(id).unwrap();
            cluster
                .get_rfengine(*id)
                .get_peer_stats(peer_id)
                .truncated_idx
        })
        .collect::<Vec<_>>();

    // Trigger switching and flushing memtable.
    for i in 0..50 {
        client.put_kv(i * 20..(i + 1) * 20, gen_key, gen_val);
    }
    let shard_stats = get_shard_stats(&cluster, region_id);
    assert!(
        shard_stats
            .iter()
            .all(|stat| stat.mem_table_count + stat.l0_table_count > 1 && stat.mem_table_size > 0)
    );

    // Wait for raftlog truncation.
    let wait_truncated = |before_truncated_idxes: Vec<_>| -> Vec<_> {
        let mut curr_truncated_idxes = vec![];
        for i in 0..10 {
            curr_truncated_idxes = node_ids
                .iter()
                .map(|id| {
                    let &peer_id = peer_ids.get(id).unwrap();
                    cluster
                        .get_rfengine(*id)
                        .get_peer_stats(peer_id)
                        .truncated_idx
                })
                .collect::<Vec<_>>();
            if curr_truncated_idxes
                .iter()
                .zip(before_truncated_idxes.iter())
                .all(|(curr, before)| curr > before)
            {
                break;
            }
            if i == 9 {
                panic!("waiting for raftlog truncation timeouts");
            }
            std::thread::sleep(Duration::from_secs(1));
        }
        curr_truncated_idxes
    };
    let curr_truncated_idxes = wait_truncated(before_truncated_idxes);

    let wait_for_memtable_flush = |cluster: &ServerCluster, node_ids: &[u16], timeout: Duration| {
        let mut curr_shard_stats;
        let mut flushed;
        let start = Instant::now_coarse();
        loop {
            curr_shard_stats = node_ids
                .iter()
                .map(|id| cluster.get_kvengine(*id).get_shard_stat(region_id))
                .collect::<Vec<_>>();
            flushed = curr_shard_stats
                .iter()
                .all(|curr| curr.mem_table_size == 0 && curr.mem_table_count == 1);
            if flushed {
                break;
            }
            if start.saturating_elapsed() >= timeout {
                break;
            }
            std::thread::sleep(Duration::from_millis(200));
        }
        (flushed, curr_shard_stats)
    };

    // Trigger switching and flushing memtable.
    let flush_memtable = |cluster: &ServerCluster, node_ids: &[u16]| {
        let mut client = cluster.new_client();
        let region_id = client.get_region_id(&[]);
        let ctx = client.new_rpc_ctx(region_id, b"").unwrap();
        ctx.get_peer().get_store_id();
        let mut req = RaftCmdRequest::default();
        let mut header = RaftRequestHeader::default();
        header.set_region_id(ctx.get_region_id());
        header.set_peer(ctx.get_peer().clone());
        header.set_region_epoch(ctx.get_region_epoch().clone());
        header.set_term(6);
        req.set_header(header);
        let mut custom_builder = CustomBuilder::new();
        custom_builder.set_switch_mem_table(1);
        req.set_custom_request(custom_builder.build());
        cluster.send_raft_command(req);

        let (flushed, curr_shard_stats) =
            wait_for_memtable_flush(cluster, node_ids, Duration::from_secs(6));
        assert!(flushed, "shard_stats: {:?}", curr_shard_stats);
        curr_shard_stats
    };

    let (flushed, mut curr_shard_stats) =
        wait_for_memtable_flush(&cluster, &node_ids, Duration::from_secs(1));
    if !flushed {
        // Invoke `flush_memtable` only when memtable has not flushed.
        // Otherwise, `switch_mem_table` on empty memtable will advance
        // `Shard.sequence` only, but `ShardMeta.data_sequence` is unchanged.
        // In this situation, raft log truncate index will be less than `Shard.sequence`
        // (and `shard_stat.write_sequence`).
        curr_shard_stats = flush_memtable(&cluster, &node_ids);
    }
    let curr_truncated_idxes = wait_truncated(curr_truncated_idxes);
    std::thread::sleep(Duration::from_secs(1));
    // Flushing memtable will propose a ChangeSet request which won't write data to
    // memtable, so it can't trigger memtable flush. We should be able to
    // truncate these logs.
    assert!(
        curr_truncated_idxes
            .iter()
            .zip(curr_shard_stats.iter())
            .any(|(truncated_idx, shard_stat)| *truncated_idx >= shard_stat.write_sequence),
        "truncated_stat: {:?} shard_stats: {:?}",
        curr_truncated_idxes,
        curr_shard_stats
    );

    // Stop one node, and leader shouldn't truncate logs immediately.
    cluster.stop_node(node_ids[2]);
    node_ids.pop();
    let mut client = cluster.new_client();
    for i in 1000..1010 {
        client.put_kv(i..i + 1, gen_key, gen_val);
    }
    flush_memtable(&cluster, &node_ids);
    std::thread::sleep(Duration::from_millis(200));
    // Leader's truncated index doesn't change immediately.
    let region = client.get_region_by_key(&[]);
    let leader_peer_id = region.peers[region.leader_idx].id;
    let &leader_peer_node = peer_ids
        .iter()
        .find(|(_, &peer_id)| peer_id == leader_peer_id)
        .unwrap()
        .0;
    let leader_truncate_idx = cluster
        .get_rfengine(leader_peer_node)
        .get_peer_stats(leader_peer_id)
        .truncated_idx;
    assert_eq!(leader_truncate_idx, curr_truncated_idxes[0]);
    // Wait for marking down peer.
    must_wait(
        || {
            let leader_truncate_idx = cluster
                .get_rfengine(leader_peer_node)
                .get_peer_stats(leader_peer_id)
                .truncated_idx;
            leader_truncate_idx > curr_truncated_idxes[0]
        },
        10,
        || {
            let leader_truncate_idx = cluster
                .get_rfengine(leader_peer_node)
                .get_peer_stats(leader_peer_id)
                .truncated_idx;
            format!("{:?} {:?}", leader_truncate_idx, curr_truncated_idxes[0])
        },
    );
    cluster.stop();
}

#[test]
fn test_raft_log_gc_no_kv_logs() {
    let node_ids = (0..3).map(|_| alloc_node_id()).collect::<Vec<_>>();
    let mut cluster = ServerCluster::new(node_ids.clone(), |_, conf| {
        conf.raft_store.raft_log_gc_no_kv_count = 1;
    });
    cluster.wait_region_replicated(&[], 3);
    let mut client = cluster.new_client();
    client.put_kv(0..100, gen_key, gen_val);
    let region = client.get_region_by_key(&[]);
    let pd_client = cluster.get_pd_client();
    pd_client.disable_default_operator();
    let remove_peer = region.peers[2].clone();
    pd_client.remove_peer(region.id, remove_peer.clone());
    must_wait(
        || pd_client.get_region(&[]).unwrap().peers.len() == 2,
        10,
        || "add peer failed".to_string(),
    );
    let mut new_peer = remove_peer.clone();
    new_peer.id = pd_client.alloc_id().unwrap();
    pd_client.add_peer(region.id, new_peer.clone());
    must_wait(
        || pd_client.get_region(&[]).unwrap().peers.len() == 3,
        10,
        || "remove peer failed".to_string(),
    );
    let scheduler = cluster.new_scheduler();
    let mut counter = 0;
    let transfer_count = rand::thread_rng().gen_range(0usize..5);
    while counter < transfer_count {
        if scheduler.transfer_random_leader() {
            counter += 1;
        }
    }
    let no_kv_count = cluster
        .get_node_config(node_ids[0])
        .raft_store
        .raft_log_gc_no_kv_count;
    must_wait(
        || {
            let rfengine = cluster.get_rfengine(node_ids[0]);
            let region_peers = rfengine.get_region_peer_map();
            let peer_id = region_peers.get(&region.id).cloned().unwrap();
            let truncated_idx = rfengine.get_truncated_index(peer_id).unwrap();
            let last_idx = rfengine.get_last_index(peer_id).unwrap_or(truncated_idx);
            info!(
                "truncated_idx: {:?} last_idx: {:?}",
                truncated_idx, last_idx
            );
            truncated_idx + no_kv_count >= last_idx
        },
        10,
        || "truncate failed".to_string(),
    );
    cluster.stop();
}

#[test]
fn test_raft_log_gc_size_limit() {
    let node_ids = (0..3).map(|_| alloc_node_id()).collect::<Vec<_>>();
    let mut cluster = ServerCluster::new(node_ids.clone(), |_, cfg| {
        cfg.raft_store.raft_base_tick_interval = ReadableDuration::millis(100);
        cfg.raft_store.raft_store_max_leader_lease = ReadableDuration::millis(50);
        cfg.raft_store.raft_log_gc_tick_interval = ReadableDuration::millis(100);
        cfg.raft_store.pd_heartbeat_tick_interval = ReadableDuration::millis(100);
        cfg.rocksdb.writecf.write_buffer_size = ReadableSize::kb(16);
        cfg.raft_store.raft_log_gc_size_limit = Some(ReadableSize::kb(1));
    });
    cluster.wait_region_replicated(&[], 3);
    cluster.stop_node(node_ids[2]);

    let mut client = cluster.new_client();
    let store_id = cluster.get_store_id(node_ids[0]);
    let peer_id = client.get_peer_id(&[], store_id);
    std::thread::sleep(Duration::from_millis(300));
    let prev_truncated_idx = cluster
        .get_rfengine(node_ids[0])
        .get_peer_stats(peer_id)
        .truncated_idx;
    for i in 0..50 {
        client.put_kv(i * 20..(i + 1) * 20, gen_key, gen_val);
    }
    must_wait(
        || {
            let curr_truncated_idx = cluster
                .get_rfengine(node_ids[0])
                .get_peer_stats(peer_id)
                .truncated_idx;
            curr_truncated_idx > prev_truncated_idx && curr_truncated_idx > 40
        },
        3,
        || "raft log size limit doesn't take effect".to_string(),
    );
    cluster.stop();
}

fn get_shard_stats(cluster: &ServerCluster, region_id: u64) -> Vec<ShardStats> {
    let nodes = cluster.get_nodes();
    nodes
        .iter()
        .map(|id| cluster.get_kvengine(*id).get_shard_stat(region_id))
        .collect::<Vec<_>>()
}

#[test]
fn test_tombstone_check_tick_triggers_tombstone_l0_compaction() {
    let node_id = alloc_node_id();
    let mut cluster = ServerCluster::new(vec![node_id], |_, conf| {
        // Trigger the leader-only file-GC tick fast enough for the test. Must
        // stay >= raft_base_tick_interval (50ms in tests).
        conf.raft_store.file_gc_compaction_check_tick_interval = ReadableDuration::millis(100);
        // Disable the per-node trigger rate limit. The candidate descends one
        // level per trigger (L0 move-down -> L1 move-down -> L2 -> drop at L3),
        // so the test needs several consecutive triggers; the default 5s limit
        // would stretch that past the test window.
        conf.kvengine.file_gc_compaction_min_interval = ReadableDuration::secs(0);
    });
    cluster.wait_pd_region_count(1);
    // Route the data into a registered keyspace so the GC safe point can be set
    // through PD below. A direct engine-cache injection would be wiped by the
    // store's periodic gc-states refresh: the default region's keyspace is
    // unregistered in the test PD, so the refresh writes back a zero safe point.
    let keyspace_id = block_on(
        cluster.create_keyspace(&CreateKeyspaceOptions::default(), Duration::from_secs(10)),
    );
    let pd_client = cluster.get_pd_client();
    let gen_key = generate_keyspace_key(keyspace_id);
    let mut client = cluster.new_client();
    client.split_keyspace(keyspace_id);
    // Put then delete the same keys within one mem-table generation, then
    // flush once: the single L0 holds both versions (~50% WRITE_CF
    // tombstones), and a single L0 table never trips the size-driven L0
    // trigger, so only the tombstone tick can compact it.
    client.put_kv(0..100, &gen_key, gen_val);
    client.del_kv(0..100, &gen_key);
    let region_id = client.get_region_id(&gen_key(0));
    cluster.flush_memtable(region_id).unwrap();
    assert!(cluster.wait_for_memtable_flushed(region_id, Duration::from_secs(10)));
    let kv = cluster.get_kvengine(node_id);
    must_wait(
        || kv.get_shard_stat(region_id).l0_table_count > 0,
        10,
        || "flushed L0 table not visible".to_string(),
    );

    let l0_file_gc_triggers = || {
        kvengine::metrics::ENGINE_COMPACTION_TRIGGER_COUNTER
            .with_label_values(&["l0_file_gc"])
            .get()
    };
    let triggers_before = l0_file_gc_triggers();

    // Advance the GC safe point past the delete versions so the tombstone
    // gates pass. Set it through PD on the registered keyspace: the store's
    // periodic gc-states refresh re-fetches from PD, so a PD-backed safe point
    // survives the whole test instead of being wiped mid-run.
    let gc_safe_point = client.get_ts().into_inner();
    block_on(pd_client.advance_txn_safe_point(keyspace_id, gc_safe_point.into())).unwrap();
    block_on(pd_client.advance_gc_safe_point(keyspace_id, gc_safe_point.into())).unwrap();
    block_on(cluster.update_gc_states_immediately()).unwrap();

    // The tombstone tick finds the tombstone L0 candidate and triggers a
    // tombstone-driven L0 compaction.
    must_wait(
        || l0_file_gc_triggers() > triggers_before,
        30,
        || "tombstone tick did not trigger an L0 file-GC compaction".to_string(),
    );
    // Prove the tombstones are dropped, not merely relocated. The candidate
    // descends one level per trigger; at L2 move-down is no longer allowed, so
    // the L2->L3 merge runs with no deeper overlap and drops every obsolete
    // tombstone plus the puts they cover. With the safe point past all
    // versions the whole WRITE_CF is reclaimed: no L0 table, every Ln level
    // empty, and zero tombstones left. A move-down alone would leave the file
    // and its ~100 tombstones parked at some level, so only the dropping
    // rewrite can reach this state.
    must_wait(
        || {
            let stat = kv.get_shard_stat(region_id);
            stat.l0_table_count == 0 && stat.write_cf_level_n_is_empty() && stat.tombs == 0
        },
        60,
        || {
            let stat = kv.get_shard_stat(region_id);
            format!(
                "tombstones not dropped: l0_table_count={}, write_cf_n_empty={}, tombs={}",
                stat.l0_table_count,
                stat.write_cf_level_n_is_empty(),
                stat.tombs,
            )
        },
    );

    // No resurrection: every deleted key must read as not-found after the drop.
    // A keep/drop bug that dropped the tombstone while keeping the put would
    // surface the old value here.
    for i in [0usize, 50, 99] {
        let (val, _) = client
            .get_key_version_opt(
                &gen_key(i),
                u64::MAX,
                Instant::now(),
                &RequestOptions::default(),
            )
            .unwrap();
        assert!(
            val.is_none(),
            "deleted key {} resurrected after file-GC: {:?}",
            i,
            val
        );
    }
    // file-GC must compact at the advanced safe point, never beyond it (else it
    // could drop data newer than the safe point).
    assert!(
        kv.get_max_used_gc_safe_point(region_id) <= gc_safe_point.into(),
        "file-GC used a safe point beyond the advanced one"
    );

    cluster.stop();
}

// File-GC must reclaim deleted keys without dropping live ones. Half the keys
// are deleted and half stay live in one L0; after the tick reclaims the
// tombstones, the live half must still read back with its value and the deleted
// half must read as not-found. Guards against a reclaim that loses live data.
#[test]
fn test_file_gc_keeps_live_keys_while_dropping_deleted() {
    let node_id = alloc_node_id();
    let mut cluster = ServerCluster::new(vec![node_id], |_, conf| {
        conf.raft_store.file_gc_compaction_check_tick_interval = ReadableDuration::millis(100);
        conf.kvengine.file_gc_compaction_min_interval = ReadableDuration::secs(0);
        // Trigger on any non-trivial ratio so the test does not depend on the
        // exact deleted fraction.
        conf.kvengine.file_gc_compaction_ratio = 0.1;
    });
    cluster.wait_pd_region_count(1);
    let mut client = cluster.new_client();
    // Put 0..100, then delete only 0..50 in the same generation and flush once:
    // a single L0 whose WRITE_CF tombstones cover just the deleted half.
    client.put_kv(0..100, gen_key, gen_val);
    client.del_kv(0..50, gen_key);
    let region_id = client.get_region_id(&gen_key(0));
    cluster.flush_memtable(region_id).unwrap();
    assert!(cluster.wait_for_memtable_flushed(region_id, Duration::from_secs(10)));
    let kv = cluster.get_kvengine(node_id);
    must_wait(
        || kv.get_shard_stat(region_id).l0_table_count > 0,
        10,
        || "flushed L0 table not visible".to_string(),
    );

    // Advance the safe point past the deletes so the tombstones are reclaimable.
    let keyspace_id = kv.get_shard(region_id).unwrap().range.keyspace_id;
    inject_keyspace_gc_safe_point(&kv, keyspace_id, client.get_ts().into_inner());

    // The tick reclaims every tombstone (the deleted half only).
    must_wait(
        || kv.get_shard_stat(region_id).tombs == 0,
        60,
        || {
            format!(
                "tombstones not reclaimed: tombs={}",
                kv.get_shard_stat(region_id).tombs
            )
        },
    );

    // The deleted half reads as not-found; the live half keeps its value.
    for i in 0..50 {
        let (val, _) = client
            .get_key_version_opt(
                &gen_key(i),
                u64::MAX,
                Instant::now(),
                &RequestOptions::default(),
            )
            .unwrap();
        assert!(
            val.is_none(),
            "deleted key {} must be gone after file-GC: {:?}",
            i,
            val
        );
    }
    for i in 50..100 {
        let (val, _) = client
            .get_key_version_opt(
                &gen_key(i),
                u64::MAX,
                Instant::now(),
                &RequestOptions::default(),
            )
            .unwrap();
        assert_eq!(val, Some(gen_val(i)), "live key {} must survive file-GC", i);
    }
    cluster.stop();
}

// File-GC must not resurrect a deleted key whose older PUT lives in a deeper
// level. A major compaction seats the PUTs at the last level, then the deletes
// land in a fresh L0; as the tombstones descend they must keep shadowing the
// deep PUTs (keep_latest_obsolete_tombstone) until the final merge drops both,
// so the keys never reappear.
#[test]
fn test_file_gc_does_not_resurrect_across_levels() {
    let node_id = alloc_node_id();
    let mut cluster = ServerCluster::new(vec![node_id], |_, conf| {
        conf.raft_store.file_gc_compaction_check_tick_interval = ReadableDuration::millis(100);
        conf.kvengine.file_gc_compaction_min_interval = ReadableDuration::secs(0);
        conf.kvengine.file_gc_compaction_ratio = 0.1;
    });
    cluster.wait_pd_region_count(1);
    let mut client = cluster.new_client();
    // Seat the PUTs at the last level with a major compaction.
    client.put_kv(0..100, gen_key, gen_val);
    let region_id = client.get_region_id(&gen_key(0));
    cluster.flush_memtable(region_id).unwrap();
    assert!(cluster.wait_for_memtable_flushed(region_id, Duration::from_secs(10)));
    cluster.request_major_compaction(
        MajorCompactionTarget::Region(region_id),
        false,
        Duration::from_secs(20),
    );
    let kv = cluster.get_kvengine(node_id);
    must_wait(
        || kv.get_shard_stat(region_id).is_major_compacted(),
        60,
        || "major compaction did not seat the puts at the last level".to_string(),
    );

    // Delete every key into a fresh L0 above the deep puts.
    client.del_kv(0..100, gen_key);
    cluster.flush_memtable(region_id).unwrap();
    assert!(cluster.wait_for_memtable_flushed(region_id, Duration::from_secs(10)));

    // Advance the safe point past the deletes so file-GC engages the tombstones
    // and drives them down through the deep puts.
    let keyspace_id = kv.get_shard(region_id).unwrap().range.keyspace_id;
    inject_keyspace_gc_safe_point(&kv, keyspace_id, client.get_ts().into_inner());
    must_wait(
        || kv.get_shard_stat(region_id).tombs == 0,
        60,
        || {
            format!(
                "tombstones not reclaimed across levels: tombs={}",
                kv.get_shard_stat(region_id).tombs
            )
        },
    );

    // No resurrection: every deleted key must read as not-found. A keep/drop bug
    // that dropped a tombstone while the deep put remained would surface the old
    // value here.
    for i in [0usize, 50, 99] {
        let (val, _) = client
            .get_key_version_opt(
                &gen_key(i),
                u64::MAX,
                Instant::now(),
                &RequestOptions::default(),
            )
            .unwrap();
        assert!(
            val.is_none(),
            "deleted key {} resurrected after cross-level file-GC: {:?}",
            i,
            val
        );
    }
    cluster.stop();
}

#[test]
fn test_file_gc_tick_triggers_old_version_compaction() {
    let node_id = alloc_node_id();
    let mut cluster = ServerCluster::new(vec![node_id], |_, conf| {
        // Trigger the leader-only file-GC tick fast enough for the test. Must
        // stay >= raft_base_tick_interval (50ms in tests).
        conf.raft_store.file_gc_compaction_check_tick_interval = ReadableDuration::millis(100);
        // Disable the trigger rate limit so the tick triggers promptly.
        conf.kvengine.file_gc_compaction_min_interval = ReadableDuration::secs(0);
        // Trigger on any non-trivial reclaimable ratio so the test does not
        // depend on the exact file layout.
        conf.kvengine.file_gc_compaction_ratio = 0.1;
    });
    cluster.wait_pd_region_count(1);
    // Route the data into a registered keyspace so the GC safe point can be set
    // through PD below. A direct engine-cache injection would be wiped by the
    // store's periodic gc-states refresh: the default region's keyspace is
    // unregistered in the test PD, so the refresh writes back a zero safe point.
    let keyspace_id = block_on(
        cluster.create_keyspace(&CreateKeyspaceOptions::default(), Duration::from_secs(10)),
    );
    let pd_client = cluster.get_pd_client();
    let gen_key = generate_keyspace_key(keyspace_id);
    let mut client = cluster.new_client();
    client.split_keyspace(keyspace_id);
    // Write each key twice in two transactions without a flush between, then
    // flush once. Both MVCC versions land in the same mem-table generation, so
    // the single L0 co-locates them and the older versions become old-version
    // data. The key count is small so both writes fit one mem-table (a larger
    // batch would spill into two L0s, leaving one version per file and no
    // old-version data). A single L0 never trips the size-driven trigger, so
    // only the old-version tick can compact it.
    client.put_kv(0..10, &gen_key, gen_val);
    client.put_kv(0..10, &gen_key, gen_val2);
    let region_id = client.get_region_id(&gen_key(0));
    cluster.flush_memtable(region_id).unwrap();
    assert!(cluster.wait_for_memtable_flushed(region_id, Duration::from_secs(10)));
    let kv = cluster.get_kvengine(node_id);
    must_wait(
        || kv.get_shard_stat(region_id).old_data_size > 0,
        10,
        || {
            let s = kv.get_shard_stat(region_id);
            format!(
                "flushed L0 has no old-version data: l0={} old_data_size={} entries={}",
                s.l0_table_count, s.old_data_size, s.entries
            )
        },
    );

    let l0_file_gc_triggers = || {
        kvengine::metrics::ENGINE_COMPACTION_TRIGGER_COUNTER
            .with_label_values(&["l0_file_gc"])
            .get()
    };
    let triggers_before = l0_file_gc_triggers();

    // Advance the GC safe point past both versions so the old-version gates
    // pass. Set it through PD on the registered keyspace: the store's periodic
    // gc-states refresh re-fetches from PD, so a PD-backed safe point survives
    // the whole test instead of being wiped mid-run.
    let gc_safe_point = client.get_ts().into_inner();
    block_on(pd_client.advance_txn_safe_point(keyspace_id, gc_safe_point.into())).unwrap();
    block_on(pd_client.advance_gc_safe_point(keyspace_id, gc_safe_point.into())).unwrap();
    block_on(cluster.update_gc_states_immediately()).unwrap();

    // The old-version tick finds the L0 candidate and triggers an old-version
    // driven L0 compaction.
    must_wait(
        || l0_file_gc_triggers() > triggers_before,
        30,
        || "old-version tick did not trigger an L0 file-GC compaction".to_string(),
    );
    // Prove the old versions are dropped, not merely relocated. The candidate
    // descends one level per trigger; the L2->L3 merge has no deeper overlap,
    // so it drops every superseded version below the safe point while keeping
    // the latest. The shard keeps its live keys (entries > 0) but holds no
    // old-version data and no L0 table afterwards. A move-down alone would
    // leave the old versions parked at some level, so only the dropping rewrite
    // can reach this state.
    must_wait(
        || {
            let stat = kv.get_shard_stat(region_id);
            stat.l0_table_count == 0
                && stat.old_data_size == 0
                && stat.old_entries == 0
                && stat.entries > 0
        },
        60,
        || {
            let stat = kv.get_shard_stat(region_id);
            format!(
                "old versions not dropped: l0_table_count={}, old_data_size={}, old_entries={}, entries={}",
                stat.l0_table_count, stat.old_data_size, stat.old_entries, stat.entries,
            )
        },
    );
    // The latest value survives the old-version drop — a value read-back, not
    // just an entries>0 count. Dropping the latest version would surface v1 or
    // not-found here.
    for i in 0..10usize {
        let (val, _) = client
            .get_key_version_opt(
                &gen_key(i),
                u64::MAX,
                Instant::now(),
                &RequestOptions::default(),
            )
            .unwrap();
        assert_eq!(
            val,
            Some(gen_val2(i)),
            "old-version file-GC dropped the latest value of key {}",
            i
        );
    }
    cluster.stop();
}

/// Inject a single-keyspace GC safe point straight into the engine cache, for
/// tests whose region keyspace is not registered in the test PD.
fn inject_keyspace_gc_safe_point(kv: &kvengine::Engine, keyspace_id: u32, safe_point: u64) {
    let now = Instant::now();
    let mut gc_states = txn_types::ClusterGcStates::new(HashMap::default(), now);
    gc_states.keyspace_gc_states.insert(
        keyspace_id,
        txn_types::GcState::new(
            keyspace_id,
            true,
            safe_point.into(),
            safe_point.into(),
            vec![],
            now,
        ),
    );
    kv.update_cluster_gc_states_cache(gc_states);
}

#[test]
fn test_file_gc_tick_in_place_drops_last_level_tombstones() {
    let node_id = alloc_node_id();
    let mut cluster = ServerCluster::new(vec![node_id], |_, conf| {
        conf.raft_store.file_gc_compaction_check_tick_interval = ReadableDuration::millis(100);
        conf.kvengine.file_gc_compaction_min_interval = ReadableDuration::secs(0);
        // The last-level in-place rewrite is off by default for rolling-upgrade
        // safety; enable it so this test exercises the L3 drop path.
        conf.kvengine.file_gc_compaction_allow_l3 = true;
    });
    cluster.wait_pd_region_count(1);
    let mut client = cluster.new_client();
    // A safe point captured before the writes: injecting it keeps the major
    // compaction below the deletes so it preserves the tombstones.
    let safe_point_before_writes = client.get_ts().into_inner();
    // Put then delete the same keys, flush, then major-compact. With the safe
    // point still behind the deletes, the major keeps the tombstones and seats
    // them in one last-level (L3) file — the layout the in-place rewrite path
    // needs (a tombstone candidate already at the last level, no lower level).
    client.put_kv(0..100, gen_key, gen_val);
    client.del_kv(0..100, gen_key);
    let region_id = client.get_region_id(&gen_key(0));
    let kv = cluster.get_kvengine(node_id);
    let keyspace_id = kv.get_shard(region_id).unwrap().range.keyspace_id;
    inject_keyspace_gc_safe_point(&kv, keyspace_id, safe_point_before_writes);
    cluster.flush_memtable(region_id).unwrap();
    assert!(cluster.wait_for_memtable_flushed(region_id, Duration::from_secs(10)));
    cluster.request_major_compaction(
        MajorCompactionTarget::Region(region_id),
        false,
        Duration::from_secs(20),
    );
    must_wait(
        || {
            let stat = kv.get_shard_stat(region_id);
            stat.is_major_compacted() && stat.tombs > 0
        },
        60,
        || {
            let stat = kv.get_shard_stat(region_id);
            format!(
                "major did not seat tombstones at the last level: major_compacted={}, tombs={}",
                stat.is_major_compacted(),
                stat.tombs
            )
        },
    );

    // C1: the refresh that ran on the major-compaction apply must have
    // populated the per-shard wakeup hint from the last-level tombstone file,
    // so the tick can find it once the safe point advances.
    assert_ne!(
        kv.get_shard(region_id)
            .unwrap()
            .file_gc_candidate_min_max_ts(),
        u64::MAX,
        "major-compacted tombstone file must populate the file-GC wakeup hint"
    );

    // The data is now one L3 file full of tombstones. Advance the safe point so
    // the tick picks it at the last level and rewrites it in place.
    let in_place_triggers = || {
        kvengine::metrics::ENGINE_COMPACTION_TRIGGER_COUNTER
            .with_label_values(&["l1_plus_file_gc_l3"])
            .get()
    };
    let triggers_before = in_place_triggers();
    inject_keyspace_gc_safe_point(&kv, keyspace_id, client.get_ts().into_inner());

    // The last-level in-place rewrite fires and drops every obsolete tombstone.
    must_wait(
        || in_place_triggers() > triggers_before,
        30,
        || "tick did not fire a last-level in-place tombstone compaction".to_string(),
    );
    must_wait(
        || kv.get_shard_stat(region_id).tombs == 0,
        60,
        || {
            format!(
                "in-place rewrite did not drop tombstones: tombs={}",
                kv.get_shard_stat(region_id).tombs
            )
        },
    );
    cluster.stop();
}

// A replica that is a follower when a file-GC knob changes online must still
// recompute its wakeup hint and carry a fresh one into a later leadership. The
// hint refresh runs on the check tick before the leader-only early return;
// regression for the lost-refresh bug where it sat behind that return, so a
// follower's stale hint kept hiding candidates the new knob made eligible.
#[test]
fn test_file_gc_hint_refreshes_on_follower_after_online_config_change() {
    let node_ids = (0..3).map(|_| alloc_node_id()).collect::<Vec<_>>();
    let mut cluster = ServerCluster::new(node_ids.clone(), |_, conf| {
        conf.raft_store.file_gc_compaction_check_tick_interval = ReadableDuration::millis(100);
        conf.kvengine.file_gc_compaction_min_interval = ReadableDuration::secs(0);
        // Gate the last-level path off so the L3 tombstone file is excluded from
        // the wakeup hint; enabling it online is the change under test.
        conf.kvengine.file_gc_compaction_allow_l3 = false;
    });
    cluster.wait_pd_region_count(1);
    let mut client = cluster.new_client();

    // Seat an all-tombstone file at the last level on every replica: a safe
    // point behind the deletes keeps the major compaction from dropping them.
    let safe_point_before_writes = client.get_ts().into_inner();
    client.put_kv(0..100, gen_key, gen_val);
    client.del_kv(0..100, gen_key);
    let region_id = client.get_region_id(&gen_key(0));
    cluster.wait_region_replicated(&gen_key(0), 3);
    let keyspace_id = cluster
        .get_kvengine(node_ids[0])
        .get_shard(region_id)
        .unwrap()
        .range
        .keyspace_id;
    for &nid in &node_ids {
        inject_keyspace_gc_safe_point(
            &cluster.get_kvengine(nid),
            keyspace_id,
            safe_point_before_writes,
        );
    }
    cluster.flush_memtable(region_id).unwrap();
    assert!(cluster.wait_for_memtable_flushed(region_id, Duration::from_secs(10)));
    cluster.request_major_compaction(
        MajorCompactionTarget::Region(region_id),
        false,
        Duration::from_secs(20),
    );

    // Pick a replica that is a follower for this region.
    let follower_peer = cluster.get_other_peer(region_id).unwrap();
    let follower_node = cluster
        .get_server_node_id(follower_peer.get_store_id())
        .unwrap();
    let follower_kv = cluster.get_kvengine(follower_node);

    // Precondition: once the follower applies the major compaction, the gated
    // last-level file leaves its hint at MAX.
    must_wait(
        || {
            let stat = follower_kv.get_shard_stat(region_id);
            stat.is_major_compacted()
                && stat.tombs > 0
                && follower_kv
                    .get_shard(region_id)
                    .is_some_and(|s| s.file_gc_candidate_min_max_ts() == u64::MAX)
        },
        20,
        || {
            let stat = follower_kv.get_shard_stat(region_id);
            format!(
                "follower did not apply the major compaction under allow_l3=false: \
                 major_compacted={}, tombs={}",
                stat.is_major_compacted(),
                stat.tombs
            )
        },
    );

    // Apply the online knob change on every node: enable L3 and stamp the
    // config-change time the tick keys its staggered refresh on.
    let changed_at = tikv_util::time::timespec_to_ns(tikv_util::time::monotonic_raw_now());
    for &nid in &node_ids {
        let kv = cluster.get_kvengine(nid);
        kv.opts.set_file_gc_compaction_allow_l3(true);
        kv.opts.set_file_gc_compaction_config_changed_at(changed_at);
    }

    // The fix: the follower recomputes its hint on its own check tick, so the
    // now-eligible last-level file drops the hint below MAX while this replica
    // is still a follower.
    must_wait(
        || {
            follower_kv
                .get_shard(region_id)
                .is_some_and(|s| s.file_gc_candidate_min_max_ts() != u64::MAX)
        },
        20,
        || "follower did not refresh its file-GC hint after the online config change".to_string(),
    );

    // The fresh hint survives the follower becoming leader.
    cluster.transfer_leader(region_id, follower_peer.clone());
    assert!(cluster.wait_leader(region_id, follower_peer.clone(), Duration::from_secs(10)));
    assert_ne!(
        follower_kv
            .get_shard(region_id)
            .unwrap()
            .file_gc_candidate_min_max_ts(),
        u64::MAX,
        "the hint refreshed while a follower must persist into leadership"
    );
    cluster.stop();
}