// Copyright 2026 TiKV Project Authors. Licensed under Apache-2.0.

//! Unit tests for the file-GC-triggered compaction triggers
//! (`find_file_gc_candidate_level` / `find_file_gc_candidate_l0`), the L0
//! input picker (`pick_candidate_l0_input_indices`), and dispatch through
//! `Engine::compact`.

use std::{
    collections::{HashMap, HashSet},
    sync::{Arc, atomic::Ordering},
};

use bytes::Bytes;
use cloud_encryption::EncryptionState;
use online_config::{ConfigManager, OnlineConfig};
use tikv_util::{config::VersionTrack, time::Instant};
use txn_types::{ClusterGcStates, GcState};

use crate::{
    EXTRA_CF, Engine, LOCK_CF, ObsoleteFileBuckets, Shard, WRITE_CF,
    compaction::{
        CompactionPriority, FileGcCompactionTriggerLimiter, file_gc_priority_score, file_gc_ratio,
        l0_file_gc_ratio,
    },
    config::Config,
    config_manager::KvEngineConfigManager,
    context::IaCtx,
    shard::{ShardCf, ShardCfBuilder, ShardDataBuilder, ShardRange},
    table,
    table::{
        BIT_DELETE, ChecksumType, SnapVersion,
        columnar::ColumnarLevels,
        file::InMemFile,
        sstable::{BlockCache, L0Builder, L0Table, SsTable},
    },
    test_engine::new_test_engine_opt_with_custom_options,
    tests::{DEF_BLOCK_SIZE, TABLE_KEY_PREFIX, TestEngine},
};

fn new_engine_with_ratio(ratio: f64) -> TestEngine {
    ::test_util::init_log_for_test();
    let (engine, _) =
        new_test_engine_opt_with_custom_options(true, DEF_BLOCK_SIZE, TABLE_KEY_PREFIX, |opts| {
            opts.set_file_gc_compaction_ratio(ratio);
            // Exercise the full file-GC behavior including the bottommost-level (L3)
            // in-place rewrite. Production defaults this off for rolling-upgrade
            // safety; the gate itself is covered by the dedicated L3-gate tests.
            opts.set_file_gc_compaction_allow_l3(true);
        });
    engine
}

/// Build a single-version SST mixing PUT and DELETE records: `tomb_every`
/// of 1 = all tombstones, 2 = 50%, 0 = none. Every entry is stamped with
/// `max_ts`.
fn build_sst_with_tombs(
    engine: &TestEngine,
    id: u64,
    begin: usize,
    end: usize,
    max_ts: u64,
    tomb_every: usize,
) -> SsTable {
    let block_size = engine.opts.table_builder_options.block_size;
    let comp_tp = engine.opts.table_builder_options.compression_tps[0];
    let comp_lvl = engine.opts.table_builder_options.compression_lvl;

    let mut builder = table::sstable::builder::Builder::new(
        id,
        block_size,
        comp_tp,
        comp_lvl,
        ChecksumType::Crc32,
        None,
    );
    for i in begin..end {
        let key = engine.key_builder.i_to_inner_key(i);
        let is_tomb = tomb_every > 0 && i % tomb_every == 0;
        let val = if is_tomb {
            table::Value::new_with_meta_version(BIT_DELETE, max_ts, 0, &[])
        } else {
            let payload = key.as_ref().repeat(2);
            table::Value::new_with_meta_version(0, max_ts, 0, &payload)
        };
        builder.add(key.as_ref(), &val, None).unwrap();
    }
    let mut data_buf = Vec::new();
    builder.finish(0, &mut data_buf).unwrap();
    let data = Bytes::from(data_buf);
    let opts = crate::dfs::Options::default();
    let fs = engine.fs.as_ref();
    let runtime = fs.get_runtime();
    runtime.block_on(fs.create(id, data.clone(), opts)).unwrap();
    let file = InMemFile::new(id, data);
    SsTable::new(Arc::new(file), BlockCache::None, EncryptionState::Disabled).unwrap()
}

/// L0 counterpart of `build_sst_with_tombs`; entries go to WRITE_CF.
fn build_l0_with_tombs(
    engine: &TestEngine,
    id: u64,
    begin: usize,
    end: usize,
    max_ts: u64,
    tomb_every: usize,
) -> L0Table {
    let block_size = engine.opts.table_builder_options.block_size;
    let mut builder = L0Builder::new(id, block_size, max_ts.into(), ChecksumType::Crc32, None);
    for i in begin..end {
        let key = engine.key_builder.i_to_inner_key(i);
        let is_tomb = tomb_every > 0 && i % tomb_every == 0;
        let val = if is_tomb {
            table::Value::new_with_meta_version(BIT_DELETE, max_ts, 0, &[])
        } else {
            let payload = key.as_ref().repeat(2);
            table::Value::new_with_meta_version(0, max_ts, 0, &payload)
        };
        builder.add(WRITE_CF, key.as_ref(), &val, None).unwrap();
    }
    let (_, data) = builder.finish().unwrap();
    let opts = crate::dfs::Options::default();
    let fs = engine.fs.as_ref();
    let runtime = fs.get_runtime();
    runtime.block_on(fs.create(id, data.clone(), opts)).unwrap();
    let file = Arc::new(InMemFile::new(id, data));
    L0Table::new(file, BlockCache::None, false, EncryptionState::Disabled)
        .unwrap()
        .unwrap()
}

/// Build a write-cf-only split L0 (the flush-split shape): a plain SST
/// stamped with a snap version so its footer carries the split-L0 magic,
/// which is what makes `try_move_down_l0` consider it.
fn build_split_l0_with_tombs(
    engine: &TestEngine,
    id: u64,
    begin: usize,
    end: usize,
    max_ts: u64,
    tomb_every: usize,
) -> L0Table {
    let block_size = engine.opts.table_builder_options.block_size;
    let comp_tp = engine.opts.table_builder_options.compression_tps[0];
    let comp_lvl = engine.opts.table_builder_options.compression_lvl;
    let mut builder = table::sstable::builder::Builder::new(
        id,
        block_size,
        comp_tp,
        comp_lvl,
        ChecksumType::Crc32,
        None,
    );
    builder.set_snap_version(max_ts.into());
    for i in begin..end {
        let key = engine.key_builder.i_to_inner_key(i);
        let is_tomb = tomb_every > 0 && i % tomb_every == 0;
        let val = if is_tomb {
            table::Value::new_with_meta_version(BIT_DELETE, max_ts, 0, &[])
        } else {
            let payload = key.as_ref().repeat(2);
            table::Value::new_with_meta_version(0, max_ts, 0, &payload)
        };
        builder.add(key.as_ref(), &val, None).unwrap();
    }
    let mut data_buf = Vec::new();
    builder.finish(0, &mut data_buf).unwrap();
    let data = Bytes::from(data_buf);
    let opts = crate::dfs::Options::default();
    let fs = engine.fs.as_ref();
    let runtime = fs.get_runtime();
    runtime.block_on(fs.create(id, data.clone(), opts)).unwrap();
    let file = Arc::new(InMemFile::new(id, data));
    L0Table::new(file, BlockCache::None, true, EncryptionState::Disabled)
        .unwrap()
        .unwrap()
}

/// Six large disjoint lower-level SSTs whose total size dwarfs a small
/// candidate, to trip the overlap-bloat gate.
fn build_huge_lower_ssts(engine: &TestEngine, base_id: u64) -> Vec<SsTable> {
    (0..6u64)
        .map(|k| {
            let start = k as usize * 10000;
            build_sst_with_tombs(engine, base_id + k, start, start + 5000, 40, 0)
        })
        .collect()
}

/// Build an SST where every entry has two MVCC versions so `old_data_size() >
/// 0`. The newer version (max_ts) is written first; the older version
/// (max_ts - 1) follows as the old-section entry.
fn build_sst_with_old_versions(
    engine: &TestEngine,
    id: u64,
    begin: usize,
    end: usize,
    max_ts: u64,
) -> SsTable {
    let block_size = engine.opts.table_builder_options.block_size;
    let comp_tp = engine.opts.table_builder_options.compression_tps[0];
    let comp_lvl = engine.opts.table_builder_options.compression_lvl;

    let mut builder = table::sstable::builder::Builder::new(
        id,
        block_size,
        comp_tp,
        comp_lvl,
        ChecksumType::Crc32,
        None,
    );
    for i in begin..end {
        let key = engine.key_builder.i_to_inner_key(i);
        let payload = key.as_ref().repeat(2);
        // Newer version first (goes to the main block section).
        let newer = table::Value::new_with_meta_version(0, max_ts, 0, &payload);
        builder.add(key.as_ref(), &newer, None).unwrap();
        // Older version second (same inner key → routes to old-data section).
        let older = table::Value::new_with_meta_version(0, max_ts - 1, 0, &payload);
        builder.add(key.as_ref(), &older, None).unwrap();
    }
    let mut data_buf = Vec::new();
    builder.finish(0, &mut data_buf).unwrap();
    let data = Bytes::from(data_buf);
    let opts = crate::dfs::Options::default();
    let fs = engine.fs.as_ref();
    let runtime = fs.get_runtime();
    runtime.block_on(fs.create(id, data.clone(), opts)).unwrap();
    let file = InMemFile::new(id, data);
    SsTable::new(Arc::new(file), BlockCache::None, EncryptionState::Disabled).unwrap()
}

/// L0 counterpart of `build_sst_with_old_versions`.
fn build_l0_with_old_versions(
    engine: &TestEngine,
    id: u64,
    begin: usize,
    end: usize,
    max_ts: u64,
) -> L0Table {
    let block_size = engine.opts.table_builder_options.block_size;
    let mut builder = L0Builder::new(id, block_size, max_ts.into(), ChecksumType::Crc32, None);
    for i in begin..end {
        let key = engine.key_builder.i_to_inner_key(i);
        let payload = key.as_ref().repeat(2);
        let newer = table::Value::new_with_meta_version(0, max_ts, 0, &payload);
        builder.add(WRITE_CF, key.as_ref(), &newer, None).unwrap();
        let older = table::Value::new_with_meta_version(0, max_ts - 1, 0, &payload);
        builder.add(WRITE_CF, key.as_ref(), &older, None).unwrap();
    }
    let (_, data) = builder.finish().unwrap();
    let opts = crate::dfs::Options::default();
    let fs = engine.fs.as_ref();
    let runtime = fs.get_runtime();
    runtime.block_on(fs.create(id, data.clone(), opts)).unwrap();
    let file = Arc::new(InMemFile::new(id, data));
    L0Table::new(file, BlockCache::None, false, EncryptionState::Disabled)
        .unwrap()
        .unwrap()
}

/// Install the given layout into shard 1: L0 tables (older-first, like the
/// engine stores them) plus a `(level -> ssts)` WRITE_CF layout.
fn install_layout(
    engine: &TestEngine,
    l0_tables: Vec<L0Table>,
    write_cf: Vec<(usize, Vec<SsTable>)>,
) -> Arc<Shard> {
    let shard = engine.get_shard(1).unwrap();
    let mut write_cf_builder = ShardCfBuilder::new(WRITE_CF);
    for (level, ssts) in write_cf {
        for sst in ssts {
            write_cf_builder.add_table(sst, level);
        }
    }
    let mut builder = ShardDataBuilder::new(shard.get_data());
    builder.set_l0_tbls(l0_tables);
    builder.set_cfs([
        write_cf_builder.build(),
        ShardCf::new(LOCK_CF),
        ShardCf::new(EXTRA_CF),
    ]);
    builder.set_persisted_version(SnapVersion::new(100, 100));
    shard.set_data(builder.build());
    shard.refresh_estimated_size_and_entries();
    // `Engine::compact` bails unless the shard is active and initially
    // flushed; synthetic layouts must satisfy both.
    shard.initial_flushed.store(true, Ordering::Release);
    shard
}

/// Mark the given L0 tables as unconverted in shard 1's data.
fn mark_unconverted_l0s(engine: &TestEngine, l0s: Vec<L0Table>) {
    let shard = engine.get_shard(1).unwrap();
    let mut col_levels = ColumnarLevels::new();
    col_levels.unconverted_l0s = l0s;
    let mut builder = ShardDataBuilder::new(shard.get_data());
    builder.set_columnar_levels(col_levels);
    shard.set_data(builder.build());
}

/// Run `Engine::pick_candidate_l0_input_indices` against shard 1's current
/// data.
fn pick_l0_overlap_set(
    engine: &TestEngine,
    tombstone_candidate_id: u64,
    unconverted: &[u64],
) -> Option<Vec<usize>> {
    let shard = engine.get_shard(1).unwrap();
    let data = shard.get_data();
    let unconverted: HashSet<u64> = unconverted.iter().cloned().collect();
    Engine::pick_candidate_l0_input_indices(
        &data,
        tombstone_candidate_id,
        &unconverted,
        shard.tag(),
    )
}

/// Make `Engine::get_gc_safe_point` return `safe_point` for shard 1's
/// keyspace.
fn set_gc_safe_point(engine: &TestEngine, safe_point: u64) {
    let shard = engine.get_shard(1).unwrap();
    let keyspace_id = shard.range.keyspace_id;
    let now = Instant::now();
    let mut cluster_gc_states = ClusterGcStates::new(HashMap::default(), now);
    cluster_gc_states.keyspace_gc_states.insert(
        keyspace_id,
        GcState::new(
            keyspace_id,
            true,
            safe_point.into(),
            safe_point.into(),
            vec![],
            now,
        ),
    );
    engine.update_cluster_gc_states_cache(cluster_gc_states);
}

fn run_compact(
    engine: &TestEngine,
    shard: &Arc<Shard>,
) -> Option<crate::Result<kvenginepb::ChangeSet>> {
    let runtime = engine.fs.as_ref().get_runtime();
    runtime.block_on(engine.compact(shard.clone()))
}

// ---------------------------------------------------------------------------
// L1Plus trigger: find_file_gc_candidate_level (Tombstone driver)
// Order: hits, then gate-by-gate skips, then scan-scope rules.
// ---------------------------------------------------------------------------

#[test]
fn test_tombstone_candidate_level_basic() {
    let engine = new_engine_with_ratio(0.3);
    // Candidate at L2 with ~50% tombstones, max_ts below safe point.
    let tombstone_candidate = build_sst_with_tombs(&engine, 101, 0, 200, 50, 2);
    let lower = build_sst_with_tombs(&engine, 102, 0, 20, 40, 0);
    install_layout(
        &engine,
        vec![],
        vec![(2, vec![tombstone_candidate]), (3, vec![lower])],
    );
    let (level, tombs_ratio) = engine
        .get_shard(1)
        .unwrap()
        .find_file_gc_candidate_level(200)
        .expect("expected file-level tombstone hit");
    assert_eq!(level, 2);
    assert!(
        tombs_ratio >= 0.3,
        "tombs_ratio {} should pass threshold",
        tombs_ratio
    );
}

#[test]
fn test_tombstone_candidate_level_hits_when_lower_empty() {
    let engine = new_engine_with_ratio(0.3);
    // A tombstone-heavy L2 SST with no L3 overlap must still hit; the picker
    // side skips move-down so the merge actually drops tombstones.
    let tombstone_candidate = build_sst_with_tombs(&engine, 180, 0, 200, 50, 2);
    install_layout(&engine, vec![], vec![(2, vec![tombstone_candidate])]);
    let (level, tombs_ratio) = engine
        .get_shard(1)
        .unwrap()
        .find_file_gc_candidate_level(200)
        .expect("expected file-level tombstone hit on empty-lower");
    assert_eq!(level, 2);
    assert!(tombs_ratio >= 0.3);
}

#[test]
fn test_tombstone_candidate_level_skips_when_max_ts_too_new() {
    let engine = new_engine_with_ratio(0.3);
    // max_ts=200 > gc_safe_point=100.
    let tombstone_candidate = build_sst_with_tombs(&engine, 110, 0, 200, 200, 2);
    let lower = build_sst_with_tombs(&engine, 111, 0, 20, 40, 0);
    install_layout(
        &engine,
        vec![],
        vec![(2, vec![tombstone_candidate]), (3, vec![lower])],
    );
    assert!(
        engine
            .get_shard(1)
            .unwrap()
            .find_file_gc_candidate_level(100)
            .is_none()
    );
}

#[test]
fn test_tombstone_candidate_level_skips_when_ratio_below_threshold() {
    // Demand 40% — candidate has 25% tombstones (tomb_every=4).
    let engine = new_engine_with_ratio(0.4);
    let tombstone_candidate = build_sst_with_tombs(&engine, 120, 0, 400, 50, 4);
    let lower = build_sst_with_tombs(&engine, 121, 0, 20, 40, 0);
    install_layout(
        &engine,
        vec![],
        vec![(2, vec![tombstone_candidate]), (3, vec![lower])],
    );
    assert!(
        engine
            .get_shard(1)
            .unwrap()
            .find_file_gc_candidate_level(100)
            .is_none()
    );
}

#[test]
fn test_tombstone_candidate_level_overlap_bloat_skip() {
    let engine = new_engine_with_ratio(0.3);
    // Tiny candidate vs huge overlapping lower SSTs exceeds the overlap ratio
    // bound.
    let tombstone_candidate = build_sst_with_tombs(&engine, 150, 0, 60, 50, 2);
    let lower = build_huge_lower_ssts(&engine, 151);
    install_layout(
        &engine,
        vec![],
        vec![(2, vec![tombstone_candidate]), (3, lower)],
    );
    assert!(
        engine
            .get_shard(1)
            .unwrap()
            .find_file_gc_candidate_level(100)
            .is_none()
    );
}

#[test]
fn test_tombstone_candidate_level_prefers_deepest_level() {
    let engine = new_engine_with_ratio(0.3);
    // Candidates at L1 (100%), L2 (100%), and L3 (50%): deepest-first
    // scanning picks the bottommost level even with the lowest ratio, so files
    // closest to the final dropping rewrite are cleared first.
    let tombstone_candidate_l1 = build_sst_with_tombs(&engine, 140, 0, 200, 50, 1);
    let tombstone_candidate_l2 = build_sst_with_tombs(&engine, 141, 500, 700, 50, 1);
    let tombstone_candidate_l3 = build_sst_with_tombs(&engine, 142, 1000, 1200, 50, 2);
    install_layout(
        &engine,
        vec![],
        vec![
            (1, vec![tombstone_candidate_l1]),
            (2, vec![tombstone_candidate_l2]),
            (3, vec![tombstone_candidate_l3]),
        ],
    );
    let (level, tombs_ratio) = engine
        .get_shard(1)
        .unwrap()
        .find_file_gc_candidate_level(100)
        .expect("expected hit");
    assert_eq!(level, 3, "deepest candidate level wins");
    assert!(
        (0.3..0.9).contains(&tombs_ratio),
        "the L3 candidate's ratio, not L1/L2's"
    );
}

#[test]
fn test_tombstone_candidate_level_finds_l1_candidate() {
    let engine = new_engine_with_ratio(0.3);
    // A candidate at L1 with clean L2/L3: L1 is scanned too (the lv2plus
    // counter is zero here, narrowing the scan to exactly L1) so the
    // move-down path can drain it.
    let tombstone_candidate_l1 = build_sst_with_tombs(&engine, 144, 0, 200, 50, 1);
    let lower_l2 = build_sst_with_tombs(&engine, 145, 0, 20, 40, 0);
    let lower_l3 = build_sst_with_tombs(&engine, 146, 0, 20, 40, 0);
    install_layout(
        &engine,
        vec![],
        vec![
            (1, vec![tombstone_candidate_l1]),
            (2, vec![lower_l2]),
            (3, vec![lower_l3]),
        ],
    );
    let (level, tombs_ratio) = engine
        .get_shard(1)
        .unwrap()
        .find_file_gc_candidate_level(100)
        .expect("expected L1 hit");
    assert_eq!(level, 1);
    assert!(tombs_ratio > 0.9);
}

#[test]
fn test_tombstone_candidate_level_hits_last_level() {
    let engine = new_engine_with_ratio(0.3);
    let shard = engine.get_shard(1).unwrap();
    let bottommost_level = shard
        .get_data()
        .get_cf(WRITE_CF)
        .levels
        .last()
        .unwrap()
        .level;
    let tombstone_candidate = build_sst_with_tombs(&engine, 130, 0, 200, 50, 2);
    install_layout(
        &engine,
        vec![],
        vec![(bottommost_level, vec![tombstone_candidate])],
    );
    let (level, tombs_ratio) = engine
        .get_shard(1)
        .unwrap()
        .find_file_gc_candidate_level(100)
        .expect("expected bottommost-level tombstone hit");
    assert_eq!(
        level, bottommost_level,
        "bottommost level rewrites in place"
    );
    assert!(tombs_ratio >= 0.3);
}

#[test]
fn test_tombstone_candidate_level_last_level_still_gated_by_max_ts() {
    let engine = new_engine_with_ratio(0.3);
    // The in-place rewrite drops tombstones outright, so the gc gate must
    // hold at the bottommost level too: max_ts=200 > gc_safe_point=100.
    let tombstone_candidate = build_sst_with_tombs(&engine, 137, 0, 200, 200, 2);
    install_layout(&engine, vec![], vec![(3, vec![tombstone_candidate])]);
    assert!(
        engine
            .get_shard(1)
            .unwrap()
            .find_file_gc_candidate_level(100)
            .is_none()
    );
}

// ---------------------------------------------------------------------------
// L1Plus priority setter and tick yield gates
// ---------------------------------------------------------------------------

#[test]
fn test_try_set_tombstone_triggered_l1_plus_priority_yields_if_priority_set() {
    let engine = new_engine_with_ratio(0.3);
    let shard = engine.get_shard(1).unwrap();
    // First call wins.
    assert!(shard.try_set_file_gc_triggered_l1_plus_priority(2, 0.5));
    assert!(shard.has_compaction_priority());
    // Second call yields because priority is already set.
    assert!(!shard.try_set_file_gc_triggered_l1_plus_priority(2, 0.6));
}

// ---------------------------------------------------------------------------
// L0 trigger: find_file_gc_candidate_l0 (Tombstone driver) and priority setter
// ---------------------------------------------------------------------------

#[test]
fn test_tombstone_candidate_l0_basic() {
    let engine = new_engine_with_ratio(0.3);
    // 50% tombstones, max_ts <= safe_point.
    let l0 = build_l0_with_tombs(&engine, 201, 0, 200, 50, 2);
    install_layout(&engine, vec![l0], vec![]);
    let (file_id, ratio) = engine
        .get_shard(1)
        .unwrap()
        .find_file_gc_candidate_l0(100)
        .expect("expected L0 tombstone hit");
    assert_eq!(file_id, 201);
    assert!(ratio >= 0.3);
}

#[test]
fn test_tombstone_candidate_l0_picks_highest_ratio_across_files() {
    let engine = new_engine_with_ratio(0.3);
    // Expect the 100% tomb file (id=232) to win.
    let l0_a = build_l0_with_tombs(&engine, 230, 0, 200, 50, 3); // ~33%
    let l0_b = build_l0_with_tombs(&engine, 231, 500, 700, 50, 2); // 50%
    let l0_c = build_l0_with_tombs(&engine, 232, 1000, 1200, 50, 1); // 100%
    install_layout(&engine, vec![l0_a, l0_b, l0_c], vec![]);
    let (file_id, ratio) = engine
        .get_shard(1)
        .unwrap()
        .find_file_gc_candidate_l0(100)
        .expect("expected hit");
    assert_eq!(file_id, 232);
    assert!(ratio > 0.9);
}

#[test]
fn test_tombstone_candidate_l0_skips_when_max_ts_too_new() {
    let engine = new_engine_with_ratio(0.3);
    // max_ts=200 > gc_safe_point=100.
    let l0 = build_l0_with_tombs(&engine, 210, 0, 200, 200, 2);
    install_layout(&engine, vec![l0], vec![]);
    assert!(
        engine
            .get_shard(1)
            .unwrap()
            .find_file_gc_candidate_l0(100)
            .is_none()
    );
}

#[test]
fn test_tombstone_candidate_l0_skips_when_ratio_below_threshold() {
    let engine = new_engine_with_ratio(0.4);
    // 25% tombstones, below the 40% threshold.
    let l0 = build_l0_with_tombs(&engine, 220, 0, 400, 50, 4);
    install_layout(&engine, vec![l0], vec![]);
    assert!(
        engine
            .get_shard(1)
            .unwrap()
            .find_file_gc_candidate_l0(100)
            .is_none()
    );
}

#[test]
fn test_tombstone_candidate_l0_overlap_bloat_skip() {
    let engine = new_engine_with_ratio(0.3);
    // 100% tombstone L0 candidate sitting under a huge L1 file: its
    // tombstones are spread thin across the keyspace, so the early merge
    // buys little scan speedup and would pull all that L1 data in.
    let l0 = build_l0_with_tombs(&engine, 240, 0, 60, 50, 1);
    let l1 = build_huge_lower_ssts(&engine, 241);
    install_layout(&engine, vec![l0], vec![(1, l1)]);
    assert!(
        engine
            .get_shard(1)
            .unwrap()
            .find_file_gc_candidate_l0(100)
            .is_none()
    );
}

#[test]
fn test_tombstone_candidate_l0_hits_with_small_l1_overlap() {
    let engine = new_engine_with_ratio(0.3);
    // A small L1 overlap stays under the gate: the candidate still hits.
    let l0 = build_l0_with_tombs(&engine, 245, 0, 200, 50, 2);
    let l1 = build_sst_with_tombs(&engine, 246, 0, 20, 40, 0);
    install_layout(&engine, vec![l0], vec![(1, vec![l1])]);
    let (file_id, ratio) = engine
        .get_shard(1)
        .unwrap()
        .find_file_gc_candidate_l0(100)
        .expect("expected hit despite small L1 overlap");
    assert_eq!(file_id, 245);
    assert!(ratio >= 0.3);
}

#[test]
fn test_tombstone_candidate_l0_skips_unconverted_candidates() {
    let engine = new_engine_with_ratio(0.3);
    // The higher-ratio file (100%) is unconverted, so the converted 50% file
    // wins: an unconverted candidate would only make the picker bail every tick
    // until conversion finishes.
    let unconverted = build_l0_with_tombs(&engine, 250, 0, 200, 50, 1);
    let converted = build_l0_with_tombs(&engine, 251, 1000, 1200, 50, 2);
    install_layout(&engine, vec![unconverted.clone(), converted], vec![]);
    mark_unconverted_l0s(&engine, vec![unconverted]);
    let (file_id, ratio) = engine
        .get_shard(1)
        .unwrap()
        .find_file_gc_candidate_l0(100)
        .expect("converted candidate should hit");
    assert_eq!(file_id, 251);
    assert!((0.3..0.9).contains(&ratio), "the 50% converted file wins");
}

#[test]
fn test_try_set_tombstone_triggered_l0_priority_yields_if_priority_set() {
    let engine = new_engine_with_ratio(0.3);
    let shard = engine.get_shard(1).unwrap();
    // First call wins.
    assert!(shard.try_set_file_gc_triggered_l0_priority(42, 0.5));
    assert!(shard.has_compaction_priority());
    // Second call yields.
    assert!(!shard.try_set_file_gc_triggered_l0_priority(43, 0.6));
}

// ---------------------------------------------------------------------------
// L0 input picker: the candidate's overlap set, nothing more.
// Order: overlap set shapes, then race/skip cases, then L1-independence.
// ---------------------------------------------------------------------------

#[test]
fn test_l0_overlap_set_single_tombstone_candidate_no_neighbors() {
    let engine = new_engine_with_ratio(0.3);
    let l0 = build_l0_with_tombs(&engine, 300, 0, 200, 50, 2);
    install_layout(&engine, vec![l0], vec![]);
    let indices = pick_l0_overlap_set(&engine, 300, &[]).expect("expected overlap set");
    assert_eq!(indices, vec![0]);
}

#[test]
fn test_l0_overlap_set_chain_overlap() {
    let engine = new_engine_with_ratio(0.3);
    // A=[0,100), B=[50,150), C=[120,200); candidate=B pulls in A and C.
    let l0_a = build_l0_with_tombs(&engine, 310, 0, 100, 50, 2);
    let l0_b = build_l0_with_tombs(&engine, 311, 50, 150, 50, 2);
    let l0_c = build_l0_with_tombs(&engine, 312, 120, 200, 50, 2);
    install_layout(&engine, vec![l0_a, l0_b, l0_c], vec![]);
    let indices = pick_l0_overlap_set(&engine, 311, &[]).expect("expected overlap set");
    assert_eq!(indices, vec![0, 1, 2]);
}

#[test]
fn test_l0_overlap_set_excludes_non_overlapping() {
    let engine = new_engine_with_ratio(0.3);
    // Two disjoint L0; overlap set stays {candidate}.
    let l0_a = build_l0_with_tombs(&engine, 320, 0, 100, 50, 2);
    let l0_b = build_l0_with_tombs(&engine, 321, 5000, 5100, 50, 2);
    install_layout(&engine, vec![l0_a, l0_b], vec![]);
    let indices = pick_l0_overlap_set(&engine, 320, &[]).expect("expected overlap set");
    assert_eq!(indices, vec![0]);
}

#[test]
fn test_l0_overlap_set_returns_none_if_tombstone_candidate_missing() {
    let engine = new_engine_with_ratio(0.3);
    let l0 = build_l0_with_tombs(&engine, 330, 0, 200, 50, 2);
    install_layout(&engine, vec![l0], vec![]);
    let indices = pick_l0_overlap_set(&engine, 9999, &[]); // wrong id
    assert!(
        indices.is_none(),
        "tombstone candidate missing should return None"
    );
}

#[test]
fn test_l0_overlap_set_skips_unconverted_tombstone_candidate() {
    let engine = new_engine_with_ratio(0.3);
    let l0 = build_l0_with_tombs(&engine, 340, 0, 200, 50, 2);
    install_layout(&engine, vec![l0], vec![]);
    let indices = pick_l0_overlap_set(&engine, 340, &[340]);
    assert!(
        indices.is_none(),
        "unconverted tombstone candidate should return None to avoid recompacting before columnar finishes"
    );
}

#[test]
fn test_l0_overlap_set_defers_when_unconverted_neighbor_overlaps() {
    let engine = new_engine_with_ratio(0.3);
    // The overlapping neighbor (351) is unconverted. Leaving it out could
    // strand an older version of a overlap set key in L0 while the overlap set
    // moves a newer one to L1, so the pick must defer to the next tick
    // instead of compacting around it.
    let l0_a = build_l0_with_tombs(&engine, 350, 0, 100, 50, 2);
    let l0_b = build_l0_with_tombs(&engine, 351, 50, 150, 60, 2);
    install_layout(&engine, vec![l0_a, l0_b], vec![]);
    let indices = pick_l0_overlap_set(&engine, 350, &[351]);
    assert!(
        indices.is_none(),
        "overlapping unconverted neighbor must defer the pick"
    );
}

#[test]
fn test_l0_overlap_set_proceeds_when_unconverted_neighbor_disjoint() {
    let engine = new_engine_with_ratio(0.3);
    // A disjoint unconverted L0 holds no versions of any overlap set key, so
    // the candidate compacts alone.
    let l0_a = build_l0_with_tombs(&engine, 360, 0, 100, 50, 2);
    let l0_b = build_l0_with_tombs(&engine, 361, 5000, 5100, 60, 2);
    install_layout(&engine, vec![l0_a, l0_b], vec![]);
    let indices =
        pick_l0_overlap_set(&engine, 360, &[361]).expect("disjoint unconverted must not block");
    assert_eq!(indices, vec![0]);
}

#[test]
fn test_l0_overlap_set_ignores_l1_layout() {
    let engine = new_engine_with_ratio(0.3);
    // Candidate L0=[0,100); L1 files both overlapping ([0,100)) and adjacent
    // ([200,500)). The L0 input set is decided by L0 overlap only — L1
    // overlap joins the compaction later via `multi_cfs_l1_tbls`, never by
    // pulling extra L0s in.
    let tombstone_candidate_l0 = build_l0_with_tombs(&engine, 400, 0, 100, 50, 2);
    let l1a = build_sst_with_tombs(&engine, 402, 0, 100, 30, 0);
    let l1b = build_sst_with_tombs(&engine, 403, 200, 500, 30, 0);
    install_layout(
        &engine,
        vec![tombstone_candidate_l0],
        vec![(1, vec![l1a, l1b])],
    );
    let indices = pick_l0_overlap_set(&engine, 400, &[]).expect("expected overlap set");
    assert_eq!(indices, vec![0]);
}

#[test]
fn test_l0_overlap_set_excludes_disjoint_l0_despite_wide_l1() {
    let engine = new_engine_with_ratio(0.3);
    // Candidate L0=[0,100), far L0=[400,500), one L1=[0,600) covering both.
    // The far L0 shares no key with the candidate, so it stays out even
    // though the same L1 file overlaps both.
    let tombstone_candidate_l0 = build_l0_with_tombs(&engine, 430, 0, 100, 50, 2);
    let far_l0 = build_l0_with_tombs(&engine, 431, 400, 500, 50, 2);
    let l1_wide = build_sst_with_tombs(&engine, 432, 0, 600, 30, 0);
    install_layout(
        &engine,
        vec![tombstone_candidate_l0, far_l0],
        vec![(1, vec![l1_wide])],
    );
    let indices = pick_l0_overlap_set(&engine, 430, &[]).expect("expected overlap set");
    assert_eq!(indices, vec![0], "disjoint far L0 must stay out of the set");
}

// ---------------------------------------------------------------------------
// Dispatch: Engine::compact end to end with tombstone/size-driven priorities
// ---------------------------------------------------------------------------

#[test]
fn test_compact_dispatch_tombstone_l1_plus_rewrites_candidate() {
    let engine = new_engine_with_ratio(0.3);
    // 50% tombstone candidate at L2 with a small L3 overlap, gc-able.
    let tombstone_candidate = build_sst_with_tombs(&engine, 401, 0, 200, 50, 2);
    let lower = build_sst_with_tombs(&engine, 402, 0, 20, 40, 0);
    let shard = install_layout(
        &engine,
        vec![],
        vec![(2, vec![tombstone_candidate]), (3, vec![lower])],
    );
    set_gc_safe_point(&engine, 100);
    let (level, tombs_ratio) = shard
        .find_file_gc_candidate_level(100)
        .expect("expected tombstone candidate level");
    assert!(shard.try_set_file_gc_triggered_l1_plus_priority(level, tombs_ratio));
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    assert!(comp.get_top_deletes().contains(&401));
    assert!(comp.get_bottom_deletes().contains(&402));
    // Rewritten, not moved down: no output table reuses the candidate id.
    assert!(comp.get_table_creates().iter().all(|t| t.get_id() != 401));
}

#[test]
fn test_compact_dispatch_tombstone_l1_plus_returns_none_when_candidate_gone() {
    let engine = new_engine_with_ratio(0.3);
    // 25% tombstones — below the 0.3 threshold, so the pick-time re-check
    // finds no qualifying candidate even though the priority says tombstone-driven
    // (simulates the candidate being compacted away between tick and pick).
    let upper = build_sst_with_tombs(&engine, 411, 0, 400, 50, 4);
    let lower = build_sst_with_tombs(&engine, 412, 0, 20, 40, 0);
    let shard = install_layout(&engine, vec![], vec![(2, vec![upper]), (3, vec![lower])]);
    set_gc_safe_point(&engine, 100);
    assert!(shard.try_set_file_gc_triggered_l1_plus_priority(2, 0.5));
    assert!(run_compact(&engine, &shard).is_none());
}

#[test]
fn test_compact_dispatch_tombstone_l1_plus_lower_empty_skips_move_down() {
    let engine = new_engine_with_ratio(0.3);
    // All-tombstone candidate at L2 with empty L3: move-down must be skipped so
    // the rewrite drops every tombstone instead of sliding them to L3.
    let tombstone_candidate = build_sst_with_tombs(&engine, 421, 0, 200, 50, 1);
    let shard = install_layout(&engine, vec![], vec![(2, vec![tombstone_candidate])]);
    set_gc_safe_point(&engine, 100);
    assert!(shard.try_set_file_gc_triggered_l1_plus_priority(2, 1.0));
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    assert!(comp.get_top_deletes().contains(&421));
    // Everything was a gc-able tombstone, so the rewrite produces nothing.
    assert!(comp.get_table_creates().is_empty());
}

#[test]
fn test_compact_dispatch_tombstone_last_level_rewrites_in_place() {
    let engine = new_engine_with_ratio(0.3);
    // An all-tombstone, gc-able candidate already at the bottommost level. There is
    // no level below it, so the trigger rewrites it in place: the request is
    // encoded as a plain L1Plus at `bottommost_level - 1` with an empty upper slot
    // and the candidate in the bottom slot, so the output lands back at the
    // bottommost level and the rewrite drops every tombstone.
    let bottommost_level = engine
        .get_shard(1)
        .unwrap()
        .get_data()
        .get_cf(WRITE_CF)
        .levels
        .last()
        .unwrap()
        .level;
    let tombstone_candidate = build_sst_with_tombs(&engine, 701, 0, 200, 50, 1);
    let shard = install_layout(
        &engine,
        vec![],
        vec![(bottommost_level, vec![tombstone_candidate])],
    );
    set_gc_safe_point(&engine, 100);
    let (level, tombs_ratio) = shard
        .find_file_gc_candidate_level(100)
        .expect("expected bottommost-level tombstone hit");
    assert_eq!(level, bottommost_level);
    assert!(shard.try_set_file_gc_triggered_l1_plus_priority(level, tombs_ratio));
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    // In-place encoding: empty upper (top) slot, candidate in the bottom slot,
    // routed at bottommost_level - 1 so the output writes back to the bottommost
    // level.
    assert!(comp.get_top_deletes().is_empty());
    assert!(comp.get_bottom_deletes().contains(&701));
    assert_eq!(comp.get_level() as usize, bottommost_level - 1);
    // Every entry was a gc-able tombstone, so the in-place rewrite drops them
    // all and produces no output table.
    assert!(comp.get_table_creates().is_empty());
}

#[test]
fn test_compact_dispatch_size_driven_l1_plus_moves_down_when_lower_empty() {
    let engine = new_engine_with_ratio(0.3);
    let upper = build_sst_with_tombs(&engine, 431, 0, 200, 50, 0);
    let shard = install_layout(&engine, vec![], vec![(2, vec![upper])]);
    {
        let mut priority = shard.compaction_priority.write().unwrap();
        *priority = Some(CompactionPriority::L1Plus {
            cf: WRITE_CF as isize,
            level: 2,
            score: 1.5,
            file_gc: false,
        });
    }
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    // Size-driven trigger with empty lower overlap keeps the move-down fast
    // path: the upper table id is preserved in table_creates.
    assert!(comp.get_top_deletes().contains(&431));
    assert!(comp.get_table_creates().iter().any(|t| t.get_id() == 431));
}

#[test]
fn test_compact_dispatch_tombstone_l0_compacts_candidate_overlap_set() {
    let engine = new_engine_with_ratio(0.3);
    // A=[0,100) overlaps B=[50,150); C=[5000,5100) is disjoint and must stay
    // out of the tombstone-driven input set. Distinct max_ts per table so
    // overlapping keys never collide on the same version.
    let l0_a = build_l0_with_tombs(&engine, 501, 0, 100, 50, 2);
    let l0_b = build_l0_with_tombs(&engine, 502, 50, 150, 60, 2);
    let l0_c = build_l0_with_tombs(&engine, 503, 5000, 5100, 50, 2);
    let shard = install_layout(&engine, vec![l0_a, l0_b, l0_c], vec![]);
    set_gc_safe_point(&engine, 100);
    let (tombstone_candidate_id, ratio) = shard
        .find_file_gc_candidate_l0(100)
        .expect("expected tombstone candidate L0");
    assert!(shard.try_set_file_gc_triggered_l0_priority(tombstone_candidate_id, ratio));
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    assert!(comp.get_top_deletes().contains(&501));
    assert!(comp.get_top_deletes().contains(&502));
    assert!(!comp.get_top_deletes().contains(&503));
}

#[test]
fn test_compact_dispatch_tombstone_l0_moves_down_movable_candidate() {
    let engine = new_engine_with_ratio(0.3);
    // Write-cf-only tombstone candidate with empty L1: the L0 rewrite keeps
    // every tombstone anyway, so the metadata-only move-down reaches the
    // same end state for free and the L1 trigger picks the file up there.
    let candidate = build_split_l0_with_tombs(&engine, 801, 0, 200, 50, 2);
    let shard = install_layout(&engine, vec![candidate], vec![]);
    set_gc_safe_point(&engine, 100);
    let (candidate_id, ratio) = shard
        .find_file_gc_candidate_l0(100)
        .expect("expected tombstone candidate L0");
    assert_eq!(candidate_id, 801);
    assert!(shard.try_set_file_gc_triggered_l0_priority(candidate_id, ratio));
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    assert_eq!(comp.get_level(), 0);
    assert!(comp.get_top_deletes().contains(&801));
    // Metadata-only: the same file id is re-created at L1.
    assert!(
        comp.get_table_creates()
            .iter()
            .any(|t| t.get_id() == 801 && t.get_level() == 1)
    );
}

#[test]
fn test_compact_dispatch_tombstone_l0_l1_overlap_falls_back_to_rewrite() {
    let engine = new_engine_with_ratio(0.3);
    // The split candidate sits on overlapping L1 data: not movable, so the
    // trigger falls back to the rewrite that merges the L1 overlap in.
    let candidate = build_split_l0_with_tombs(&engine, 811, 0, 200, 50, 2);
    let l1 = build_sst_with_tombs(&engine, 812, 0, 20, 40, 0);
    let shard = install_layout(&engine, vec![candidate], vec![(1, vec![l1])]);
    set_gc_safe_point(&engine, 100);
    let (candidate_id, ratio) = shard
        .find_file_gc_candidate_l0(100)
        .expect("expected tombstone candidate L0");
    assert_eq!(candidate_id, 811);
    assert!(shard.try_set_file_gc_triggered_l0_priority(candidate_id, ratio));
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    assert!(comp.get_top_deletes().contains(&811));
    assert!(comp.get_bottom_deletes().contains(&812));
    // Rewritten, not moved: no output reuses the candidate id.
    assert!(comp.get_table_creates().iter().all(|t| t.get_id() != 811));
}

#[test]
fn test_compact_dispatch_tombstone_l0_bystander_move_down_not_adopted() {
    let engine = new_engine_with_ratio(0.3);
    // The candidate is pinned by L1 overlap while a distant split L0 is
    // movable: the trigger must not spend the round moving the bystander —
    // the candidate's rewrite runs instead.
    let candidate = build_split_l0_with_tombs(&engine, 821, 0, 200, 50, 2);
    let bystander = build_split_l0_with_tombs(&engine, 822, 5000, 5100, 60, 0);
    let l1 = build_sst_with_tombs(&engine, 823, 0, 20, 40, 0);
    let shard = install_layout(&engine, vec![candidate, bystander], vec![(1, vec![l1])]);
    set_gc_safe_point(&engine, 100);
    let (candidate_id, ratio) = shard
        .find_file_gc_candidate_l0(100)
        .expect("expected tombstone candidate L0");
    assert_eq!(candidate_id, 821);
    assert!(shard.try_set_file_gc_triggered_l0_priority(candidate_id, ratio));
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    // Rewrite of the candidate, not a move-down of the bystander.
    assert!(comp.get_top_deletes().contains(&821));
    assert!(!comp.get_top_deletes().contains(&822));
    assert!(comp.get_table_creates().iter().all(|t| t.get_id() != 821));
}

#[test]
fn test_compact_dispatch_size_driven_l0_all_path() {
    let engine = new_engine_with_ratio(0.3);
    // Two L0 tables plus an overlapping WRITE_CF L1 table: the L1 overlap
    // blocks the move-down fast path, so the all-L0 input collection runs and
    // pulls the L1 table into the compaction.
    let l0_a = build_l0_with_tombs(&engine, 521, 0, 100, 50, 2);
    let l0_b = build_l0_with_tombs(&engine, 522, 0, 100, 60, 2);
    let l1 = build_sst_with_tombs(&engine, 524, 0, 100, 40, 0);
    let shard = install_layout(&engine, vec![l0_a, l0_b], vec![(1, vec![l1])]);
    set_gc_safe_point(&engine, 100);
    {
        let mut priority = shard.compaction_priority.write().unwrap();
        *priority = Some(CompactionPriority::L0 {
            score: 2.0,
            file_candidate: None,
        });
    }
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    assert!(comp.get_top_deletes().contains(&521));
    assert!(comp.get_top_deletes().contains(&522));
    assert!(comp.get_bottom_deletes().contains(&524));
}

// Size-driven L0 keeps the metadata-only move-down. A movable split L0 over an
// empty L1 must move down with its id preserved. This is the `None`
// (size-driven) arm of the `candidate: Option<u64>` parameter the file-GC work
// threaded through `trigger_l0_compaction`, so it must behave exactly as
// before.
#[test]
fn test_compact_dispatch_size_driven_l0_moves_down_movable_split_l0() {
    let engine = new_engine_with_ratio(0.3);
    let movable = build_split_l0_with_tombs(&engine, 8021, 0, 200, 50, 0);
    let shard = install_layout(&engine, vec![movable], vec![]);
    {
        let mut priority = shard.compaction_priority.write().unwrap();
        *priority = Some(CompactionPriority::L0 {
            score: 2.0,
            file_candidate: None,
        });
    }
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    assert_eq!(comp.get_level(), 0);
    assert!(comp.get_top_deletes().contains(&8021));
    // Metadata-only move-down: the same file id is re-created one level down at L1.
    assert!(
        comp.get_table_creates()
            .iter()
            .any(|t| t.get_id() == 8021 && t.get_level() == 1)
    );
}

#[test]
fn test_compact_dispatch_tombstone_l1_moves_down_when_l2_empty() {
    let engine = new_engine_with_ratio(0.3);
    // An L1 candidate with empty L2: above bottommost_level-1 the tombstone-driven
    // trigger takes the metadata-only move-down so the next tick re-picks
    // the file at L2 for the final dropping rewrite.
    let tombstone_candidate = build_sst_with_tombs(&engine, 601, 0, 200, 50, 2);
    let shard = install_layout(&engine, vec![], vec![(1, vec![tombstone_candidate])]);
    set_gc_safe_point(&engine, 100);
    let (level, tombs_ratio) = shard
        .find_file_gc_candidate_level(100)
        .expect("expected L1 hit");
    assert_eq!(level, 1);
    assert!(shard.try_set_file_gc_triggered_l1_plus_priority(level, tombs_ratio));
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    assert!(comp.get_top_deletes().contains(&601));
    // Metadata-only move-down: the same table id is re-created at L2.
    assert!(
        comp.get_table_creates()
            .iter()
            .any(|t| t.get_id() == 601 && t.get_level() == 2)
    );
}

#[test]
fn test_compact_dispatch_tombstone_l1_with_l2_overlap_rewrites() {
    let engine = new_engine_with_ratio(0.3);
    // An L1 candidate overlapping an L2 file: move-down is impossible, so the
    // trigger rewrites L1+L2 into L2. L3 is empty, so the rewrite drops the
    // tombstones right here.
    let tombstone_candidate = build_sst_with_tombs(&engine, 611, 0, 200, 50, 1);
    let lower = build_sst_with_tombs(&engine, 612, 0, 20, 40, 0);
    let shard = install_layout(
        &engine,
        vec![],
        vec![(1, vec![tombstone_candidate]), (2, vec![lower])],
    );
    set_gc_safe_point(&engine, 100);
    let (level, tombs_ratio) = shard
        .find_file_gc_candidate_level(100)
        .expect("expected L1 hit");
    assert_eq!(level, 1);
    assert!(shard.try_set_file_gc_triggered_l1_plus_priority(level, tombs_ratio));
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    assert!(comp.get_top_deletes().contains(&611));
    assert!(comp.get_bottom_deletes().contains(&612));
    // Rewritten, not moved down: no output table reuses the candidate id.
    assert!(comp.get_table_creates().iter().all(|t| t.get_id() != 611));
}

// The crown-jewel correctness pair: a file-gc merge must KEEP the latest
// obsolete tombstone while a deeper level still overlaps the input (dropping
// it would resurrect any older value the deeper level shadows), and may only
// DROP it once nothing deeper overlaps. Both inputs are all-tombstone and
// gc-able, so the keep/drop decision shows up as output emptiness, and the only
// difference between the two tests is whether a deeper (L3) file overlaps.

#[test]
fn test_compact_dispatch_tombstone_keeps_tombstones_when_deeper_level_overlaps() {
    let engine = new_engine_with_ratio(0.3);
    // L1 all-tombstone candidate, overlapping all-tombstone L2 (forces a merge,
    // not a move-down), and an overlapping L3. The L1->L2 merge sees a deeper
    // (L3) overlap, so keep_latest_obsolete_tombstone must be set: the latest
    // tombstones survive into the output rather than being dropped.
    let candidate = build_sst_with_tombs(&engine, 941, 0, 200, 50, 1);
    let l2 = build_sst_with_tombs(&engine, 942, 0, 200, 40, 1);
    let l3 = build_sst_with_tombs(&engine, 943, 0, 200, 30, 1);
    let shard = install_layout(
        &engine,
        vec![],
        vec![(1, vec![candidate]), (2, vec![l2]), (3, vec![l3])],
    );
    set_gc_safe_point(&engine, 100);
    assert!(shard.try_set_file_gc_triggered_l1_plus_priority(1, 1.0));
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    assert!(comp.get_top_deletes().contains(&941));
    assert!(
        !comp.get_table_creates().is_empty(),
        "tombstones must be kept while a deeper level overlaps"
    );
}

#[test]
fn test_compact_dispatch_tombstone_drops_tombstones_when_no_deeper_overlap() {
    let engine = new_engine_with_ratio(0.3);
    // Same all-tombstone L1+L2 merge but with no L3: nothing deeper overlaps,
    // so the obsolete tombstones are dropped and the all-tombstone input
    // produces no output. This isolates the keep flag against the test above.
    let candidate = build_sst_with_tombs(&engine, 951, 0, 200, 50, 1);
    let l2 = build_sst_with_tombs(&engine, 952, 0, 200, 40, 1);
    let shard = install_layout(&engine, vec![], vec![(1, vec![candidate]), (2, vec![l2])]);
    set_gc_safe_point(&engine, 100);
    assert!(shard.try_set_file_gc_triggered_l1_plus_priority(1, 1.0));
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    assert!(comp.get_top_deletes().contains(&951));
    assert!(
        comp.get_table_creates().is_empty(),
        "obsolete tombstones must drop when nothing deeper overlaps"
    );
}

#[test]
fn test_compact_dispatch_size_driven_l1_plus_expands_to_neighbors() {
    let engine = new_engine_with_ratio(0.3);
    // Three L2 tables with the middle one largest and no L3 overlap: the
    // size-driven pick seeds on the middle table (max upper/lower ratio) and
    // the shared expansion pulls in both neighbors before the move-down.
    let left = build_sst_with_tombs(&engine, 621, 0, 50, 50, 0);
    let mid = build_sst_with_tombs(&engine, 622, 100, 400, 50, 0);
    let right = build_sst_with_tombs(&engine, 623, 450, 500, 50, 0);
    let shard = install_layout(&engine, vec![], vec![(2, vec![left, mid, right])]);
    {
        let mut priority = shard.compaction_priority.write().unwrap();
        *priority = Some(CompactionPriority::L1Plus {
            cf: WRITE_CF as isize,
            level: 2,
            score: 1.5,
            file_gc: false,
        });
    }
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    for id in [621, 622, 623] {
        assert!(
            comp.get_top_deletes().contains(&id),
            "expansion should include table {}",
            id
        );
    }
}

// Size-driven L1+ seeds on the max upper/lower ratio table and stops the shared
// expansion once folding a neighbor would drop the ratio. With a non-empty
// lower this is the path the file-GC refactor must leave untouched: only the
// high-ratio seed and its overlapping lower file are rewritten, the low-ratio
// neighbors and their lower files stay out. `require_lower_overlap` is `false`
// here, so the neighbor-overlap gate added for file-GC must not fire.
#[test]
fn test_compact_dispatch_size_driven_l1_plus_seeds_on_max_ratio_and_stops_expansion() {
    let engine = new_engine_with_ratio(0.3);
    // Only the middle L2 table overlaps a tiny L3 file, so its upper/lower ratio
    // dwarfs the two flanking tables that each sit over a full-size L3 file.
    let left = build_sst_with_tombs(&engine, 8001, 0, 50, 50, 0);
    let mid = build_sst_with_tombs(&engine, 8002, 100, 400, 50, 0);
    let right = build_sst_with_tombs(&engine, 8003, 450, 500, 50, 0);
    // L3 holds older versions; cross-level same-version overlap is invalid MVCC.
    let left_low = build_sst_with_tombs(&engine, 8004, 0, 50, 40, 0);
    let mid_low = build_sst_with_tombs(&engine, 8005, 200, 210, 40, 0);
    let right_low = build_sst_with_tombs(&engine, 8006, 450, 500, 40, 0);
    let shard = install_layout(
        &engine,
        vec![],
        vec![
            (2, vec![left, mid, right]),
            (3, vec![left_low, mid_low, right_low]),
        ],
    );
    {
        let mut priority = shard.compaction_priority.write().unwrap();
        *priority = Some(CompactionPriority::L1Plus {
            cf: WRITE_CF as isize,
            level: 2,
            score: 1.5,
            file_gc: false,
        });
    }
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    // Only the seed and its overlapping lower file are inputs.
    assert!(comp.get_top_deletes().contains(&8002));
    assert!(
        !comp.get_top_deletes().contains(&8001) && !comp.get_top_deletes().contains(&8003),
        "low-ratio neighbors must not be absorbed: top_deletes={:?}",
        comp.get_top_deletes()
    );
    assert!(comp.get_bottom_deletes().contains(&8005));
    assert!(
        !comp.get_bottom_deletes().contains(&8004) && !comp.get_bottom_deletes().contains(&8006),
        "neighbor lower files must not be inputs: bottom_deletes={:?}",
        comp.get_bottom_deletes()
    );
    // Non-empty lower means a rewrite, not a move-down: no output reuses the id.
    assert!(comp.get_table_creates().iter().all(|t| t.get_id() != 8002));
}

// Size-driven L1+ batches a whole disjoint level into one move-down. Three
// disjoint L2 tables over an empty L3 expand into a single metadata-only
// move-down that preserves every id. This is the behavior the file-GC
// `require_lower_overlap` gate must keep for the size-driven (`false`) path:
// disjoint neighbors are still folded in to batch the free whole-level
// move-down.
#[test]
fn test_compact_dispatch_size_driven_l1_plus_moves_down_multiple_files_preserving_ids() {
    let engine = new_engine_with_ratio(0.3);
    let a = build_sst_with_tombs(&engine, 8011, 0, 50, 50, 0);
    let b = build_sst_with_tombs(&engine, 8012, 100, 150, 50, 0);
    let c = build_sst_with_tombs(&engine, 8013, 200, 250, 50, 0);
    let shard = install_layout(&engine, vec![], vec![(2, vec![a, b, c])]);
    {
        let mut priority = shard.compaction_priority.write().unwrap();
        *priority = Some(CompactionPriority::L1Plus {
            cf: WRITE_CF as isize,
            level: 2,
            score: 1.5,
            file_gc: false,
        });
    }
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    for id in [8011, 8012, 8013] {
        assert!(
            comp.get_top_deletes().contains(&id),
            "table {} should move down",
            id
        );
        // Metadata-only move-down: every id is re-created one level down at L3.
        assert!(
            comp.get_table_creates()
                .iter()
                .any(|t| t.get_id() == id && t.get_level() == 3),
            "table {} should be re-created at L3",
            id
        );
    }
}

// Size-driven L1+ bails cleanly when its upper level is empty. The empty-upper
// guard predates the file-GC work; lock it so the refactored trigger keeps
// returning `None` instead of indexing an empty level.
#[test]
fn test_compact_dispatch_size_driven_l1_plus_none_when_upper_empty() {
    let engine = new_engine_with_ratio(0.3);
    // Only L3 holds data, so the L2 upper level for a level-2 trigger is empty.
    let lower = build_sst_with_tombs(&engine, 8051, 0, 100, 40, 0);
    let shard = install_layout(&engine, vec![], vec![(3, vec![lower])]);
    {
        let mut priority = shard.compaction_priority.write().unwrap();
        *priority = Some(CompactionPriority::L1Plus {
            cf: WRITE_CF as isize,
            level: 2,
            score: 1.5,
            file_gc: false,
        });
    }
    assert!(run_compact(&engine, &shard).is_none());
}

// File-GC must not absorb same-level neighbors that share no lower file with
// the seed. Three adjacent L2 files -- clean left/right around a
// tombstone-dense middle seed -- over an empty L3. The seed shares no lower
// file with either neighbor, so the clean cut compacts the seed alone. The
// shared size-driven expansion would instead swallow both clean neighbors
// (empty lower -> ratio degenerates to the growing upper size) and rewrite the
// whole region for nothing.
#[test]
fn test_compact_dispatch_file_gc_l1_plus_does_not_absorb_disjoint_neighbors() {
    let engine = new_engine_with_ratio(0.3);
    let left = build_sst_with_tombs(&engine, 741, 1, 50, 50, 0);
    let seed = build_sst_with_tombs(&engine, 742, 50, 80, 50, 2);
    let right = build_sst_with_tombs(&engine, 743, 80, 100, 50, 0);
    let shard = install_layout(&engine, vec![], vec![(2, vec![left, seed, right])]);
    set_gc_safe_point(&engine, 100);
    assert!(shard.try_set_file_gc_triggered_l1_plus_priority(2, 0.5));
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    assert!(
        comp.get_top_deletes().contains(&742),
        "the seed must be compacted"
    );
    assert!(
        !comp.get_top_deletes().contains(&741) && !comp.get_top_deletes().contains(&743),
        "clean neighbors sharing no lower file must not be absorbed: top_deletes={:?}",
        comp.get_top_deletes()
    );
}

// File-GC SHOULD fold in same-level neighbors that share the lower file it must
// rewrite. Same three L2 files, but one wide L3 file overlaps all three:
// rewriting it for the seed alone would leave it overlapping the neighbors, so
// the clean cut pulls them in. Locks this behavior so the disjoint-neighbor fix
// does not over-restrict.
#[test]
fn test_compact_dispatch_file_gc_l1_plus_folds_in_neighbors_sharing_lower() {
    let engine = new_engine_with_ratio(0.3);
    let left = build_sst_with_tombs(&engine, 751, 1, 50, 50, 0);
    let seed = build_sst_with_tombs(&engine, 752, 50, 80, 50, 2);
    let right = build_sst_with_tombs(&engine, 753, 80, 100, 50, 0);
    // The L3 file holds older versions of the same keys, as a real LSM layers
    // them; same-version overlap across levels would be invalid MVCC.
    let lower = build_sst_with_tombs(&engine, 760, 1, 100, 40, 0);
    let shard = install_layout(
        &engine,
        vec![],
        vec![(2, vec![left, seed, right]), (3, vec![lower])],
    );
    set_gc_safe_point(&engine, 100);
    assert!(shard.try_set_file_gc_triggered_l1_plus_priority(2, 0.5));
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    for id in [751, 752, 753] {
        assert!(
            comp.get_top_deletes().contains(&id),
            "neighbor {} sharing the lower file must be folded in: top_deletes={:?}",
            id,
            comp.get_top_deletes()
        );
    }
    assert!(
        comp.get_bottom_deletes().contains(&760),
        "the shared lower file must be an input"
    );
}

// File-GC must not over-select even when the seed has a real lower file. The
// seed overlaps its own L3 file; the clean left/right neighbors overlap no L3
// file. `require_lower_overlap` keeps the rewrite to the seed plus its lower
// file -- folding the disjoint clean neighbors would be pure write
// amplification.
#[test]
fn test_compact_dispatch_file_gc_l1_plus_keeps_seed_alone_with_nonempty_disjoint_lower() {
    let engine = new_engine_with_ratio(0.3);
    let left = build_sst_with_tombs(&engine, 8031, 1, 50, 50, 0);
    let seed = build_sst_with_tombs(&engine, 8032, 50, 80, 50, 2);
    let right = build_sst_with_tombs(&engine, 8033, 80, 100, 50, 0);
    // Older-version lower file overlapping the seed only.
    let seed_low = build_sst_with_tombs(&engine, 8034, 50, 80, 40, 0);
    let shard = install_layout(
        &engine,
        vec![],
        vec![(2, vec![left, seed, right]), (3, vec![seed_low])],
    );
    set_gc_safe_point(&engine, 100);
    assert!(shard.try_set_file_gc_triggered_l1_plus_priority(2, 0.5));
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    assert!(
        comp.get_top_deletes().contains(&8032),
        "the seed must be compacted"
    );
    assert_eq!(
        comp.get_top_deletes().len(),
        1,
        "only the seed must be an upper input: top_deletes={:?}",
        comp.get_top_deletes()
    );
    assert!(
        comp.get_bottom_deletes().contains(&8034),
        "the seed's lower file must be rewritten"
    );
}

// File-GC expansion is directional: it folds in only the neighbors that share
// the lower file it must rewrite and stops at the first neighbor that does not.
// A wide L3 file overlaps the left neighbor and the seed but not the right
// neighbor, so the pick folds left, keeps the shared lower, and excludes right.
#[test]
fn test_compact_dispatch_file_gc_l1_plus_folds_only_lower_sharing_neighbor() {
    let engine = new_engine_with_ratio(0.3);
    let left = build_sst_with_tombs(&engine, 8041, 1, 50, 50, 0);
    let seed = build_sst_with_tombs(&engine, 8042, 50, 80, 50, 2);
    let right = build_sst_with_tombs(&engine, 8043, 80, 100, 50, 0);
    // Older-version lower file spanning left + seed but not right.
    let wide_low = build_sst_with_tombs(&engine, 8044, 1, 80, 40, 0);
    let shard = install_layout(
        &engine,
        vec![],
        vec![(2, vec![left, seed, right]), (3, vec![wide_low])],
    );
    set_gc_safe_point(&engine, 100);
    assert!(shard.try_set_file_gc_triggered_l1_plus_priority(2, 0.5));
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    assert!(
        comp.get_top_deletes().contains(&8041) && comp.get_top_deletes().contains(&8042),
        "the lower-sharing left neighbor and the seed must be folded in: top_deletes={:?}",
        comp.get_top_deletes()
    );
    assert!(
        !comp.get_top_deletes().contains(&8043),
        "the disjoint right neighbor must be excluded: top_deletes={:?}",
        comp.get_top_deletes()
    );
    assert!(
        comp.get_bottom_deletes().contains(&8044),
        "the shared lower file must be an input"
    );
}

#[test]
fn test_compact_dispatch_tombstone_l1_plus_skips_candidate_above_safe_point_at_pick() {
    let engine = new_engine_with_ratio(0.3);
    // The only candidate's max_ts is above the safe point at pick time (the
    // safe point can regress between tick and pick): no candidate, no compaction.
    let upper = build_sst_with_tombs(&engine, 631, 0, 200, 300, 2);
    let lower = build_sst_with_tombs(&engine, 632, 0, 20, 40, 0);
    let shard = install_layout(&engine, vec![], vec![(2, vec![upper]), (3, vec![lower])]);
    set_gc_safe_point(&engine, 100);
    assert!(shard.try_set_file_gc_triggered_l1_plus_priority(2, 0.5));
    assert!(run_compact(&engine, &shard).is_none());
}

#[test]
fn test_compact_dispatch_tombstone_l1_plus_skips_overlap_bloat_at_pick() {
    let engine = new_engine_with_ratio(0.3);
    // The candidate passes the ratio gate but its L3 overlap dwarfs it: the
    // pick-time overlap-bloat re-check must bail instead of compacting.
    let tombstone_candidate = build_sst_with_tombs(&engine, 641, 0, 60, 50, 1);
    let lower = build_huge_lower_ssts(&engine, 642);
    let shard = install_layout(
        &engine,
        vec![],
        vec![(2, vec![tombstone_candidate]), (3, lower)],
    );
    set_gc_safe_point(&engine, 100);
    assert!(shard.try_set_file_gc_triggered_l1_plus_priority(2, 1.0));
    assert!(run_compact(&engine, &shard).is_none());
}

// ---------------------------------------------------------------------------
// Dispatch: bottommost-level in-place rewrite
// ---------------------------------------------------------------------------

#[test]
fn test_compact_dispatch_tombstone_last_level_inplace_rewrites_candidate() {
    let engine = new_engine_with_ratio(0.3);
    // 50% tombstone candidate at the bottommost level: the rewrite reuses the
    // L1Plus wire format with the candidate in the bottom slot so the output
    // lands back at L3 and older remote compactors handle the request as-is.
    let tombstone_candidate = build_sst_with_tombs(&engine, 701, 0, 200, 50, 2);
    let shard = install_layout(&engine, vec![], vec![(3, vec![tombstone_candidate])]);
    set_gc_safe_point(&engine, 100);
    let (level, tombs_ratio) = shard
        .find_file_gc_candidate_level(100)
        .expect("expected bottommost-level hit");
    assert_eq!(level, 3);
    assert!(shard.try_set_file_gc_triggered_l1_plus_priority(level, tombs_ratio));
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    // In-place encoding: empty upper slot, candidate in the bottom slot, level
    // encoded as bottommost_level - 1 so the output level is the bottommost level
    // again.
    assert_eq!(comp.get_level(), 2);
    assert!(comp.get_top_deletes().is_empty());
    assert!(comp.get_bottom_deletes().contains(&701));
    // Rewritten, not moved down: fresh output files stay at the bottommost level.
    assert!(!comp.get_table_creates().is_empty());
    for t in comp.get_table_creates() {
        assert_eq!(t.get_level(), 3);
        assert_ne!(t.get_id(), 701);
    }
}

#[test]
fn test_compact_dispatch_tombstone_last_level_inplace_drops_all_tombstones() {
    let engine = new_engine_with_ratio(0.3);
    // All-tombstone bottommost-level file: with nothing below the bottommost level
    // the rewrite drops every record and produces no output at all.
    let tombstone_candidate = build_sst_with_tombs(&engine, 711, 0, 200, 50, 1);
    let shard = install_layout(&engine, vec![], vec![(3, vec![tombstone_candidate])]);
    set_gc_safe_point(&engine, 100);
    let (level, tombs_ratio) = shard
        .find_file_gc_candidate_level(100)
        .expect("expected bottommost-level hit");
    assert_eq!(level, 3);
    assert!(tombs_ratio > 0.9);
    assert!(shard.try_set_file_gc_triggered_l1_plus_priority(level, tombs_ratio));
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    assert!(comp.get_top_deletes().is_empty());
    assert!(comp.get_bottom_deletes().contains(&711));
    assert!(comp.get_table_creates().is_empty());
}

// ---------------------------------------------------------------------------
// Rolling-upgrade compatibility: the `file_gc_compaction_allow_l3` gate must
// stop the producer from emitting the empty change set that crashes a peer
// whose `is_move_down` lacks the empty-`top_deletes` guard.
// ---------------------------------------------------------------------------

/// The legacy `is_move_down` (release-nextgen-20251011 meta.rs:1688): no
/// empty-`top_deletes` guard, so it indexes `top_deletes[0]` whenever
/// `top_deletes.len() == table_creates.len()`. An L3 full drop emits exactly
/// `top_deletes=[]`, `table_creates=[]`, so `0 == 0` holds and the index
/// panics.
fn is_move_down_without_empty_guard(comp: &kvenginepb::Compaction) -> bool {
    comp.get_top_deletes().len() == comp.get_table_creates().len()
        && comp.get_top_deletes()[0] == comp.get_table_creates()[0].id
}

#[test]
#[should_panic(expected = "index out of bounds")]
fn test_is_move_down_without_empty_guard_panics_on_full_drop() {
    // The L3 full drop's change set: empty top_deletes AND empty table_creates
    // (the candidate rides in bottom_deletes). This is the exact input that
    // crashes an un-upgraded peer.
    let mut comp = kvenginepb::Compaction::new();
    comp.set_bottom_deletes(vec![711]);
    is_move_down_without_empty_guard(&comp);
}

#[test]
fn test_l3_gate_blocks_the_full_drop_change_set() {
    // With L3 enabled the bottommost-level full drop produces the empty/empty
    // change set that test_is_move_down_without_empty_guard_panics_on_full_drop
    // shows crashes an un-upgraded peer.
    let engine = new_engine_with_ratio(0.3);
    let all_tombstones = build_sst_with_tombs(&engine, 811, 0, 200, 50, 1);
    let shard = install_layout(&engine, vec![], vec![(3, vec![all_tombstones])]);
    set_gc_safe_point(&engine, 100);
    let (level, ratio) = shard
        .find_file_gc_candidate_level(100)
        .expect("L3 enabled: bottommost-level candidate is found");
    assert_eq!(level, 3);
    assert!(shard.try_set_file_gc_triggered_l1_plus_priority(level, ratio));
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    assert!(comp.get_top_deletes().is_empty());
    assert!(comp.get_table_creates().is_empty());

    // With L3 gated off (production default) the finder never selects the
    // bottommost-level candidate, so the producer never emits that change set.
    let (engine_off, _) =
        new_test_engine_opt_with_custom_options(true, DEF_BLOCK_SIZE, TABLE_KEY_PREFIX, |opts| {
            opts.set_file_gc_compaction_ratio(0.3);
        });
    let off_tombstones = build_sst_with_tombs(&engine_off, 812, 0, 200, 50, 1);
    let shard_off = install_layout(&engine_off, vec![], vec![(3, vec![off_tombstones])]);
    set_gc_safe_point(&engine_off, 100);
    assert!(
        shard_off.find_file_gc_candidate_level(100).is_none(),
        "L3 gate off: the bottommost-level full-drop candidate must not be selected"
    );
}

#[test]
fn test_compact_dispatch_tombstone_last_level_single_candidate_no_expansion() {
    let engine = new_engine_with_ratio(0.3);
    // Two disjoint candidates at the bottommost level: the in-place pick takes
    // only the highest-ratio one; the other file waits for the next tick.
    let picked = build_sst_with_tombs(&engine, 721, 0, 200, 50, 1); // 100%
    let waiting = build_sst_with_tombs(&engine, 722, 1000, 1200, 50, 2); // 50%
    let shard = install_layout(&engine, vec![], vec![(3, vec![picked, waiting])]);
    set_gc_safe_point(&engine, 100);
    assert!(shard.try_set_file_gc_triggered_l1_plus_priority(3, 1.0));
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    assert!(comp.get_bottom_deletes().contains(&721));
    assert!(
        !comp.get_bottom_deletes().contains(&722),
        "the in-place pick must not run the generic expansion"
    );
}

#[test]
fn test_compact_dispatch_size_driven_at_last_level_bails() {
    let engine = new_engine_with_ratio(0.3);
    // Size-driven priorities never legitimately target the bottommost level; a
    // stale or hand-crafted one must bail instead of panicking on a missing
    // lower level.
    let tbl = build_sst_with_tombs(&engine, 771, 0, 200, 50, 0);
    let shard = install_layout(&engine, vec![], vec![(3, vec![tbl])]);
    {
        let mut priority = shard.compaction_priority.write().unwrap();
        *priority = Some(CompactionPriority::L1Plus {
            cf: WRITE_CF as isize,
            level: 3,
            score: 1.5,
            file_gc: false,
        });
    }
    assert!(run_compact(&engine, &shard).is_none());
}

// ---------------------------------------------------------------------------
// Split-leftover files: the finder and the wakeup hint no longer apply a
// shard-bound filter, so out-of-bound files are considered like any other.
// ---------------------------------------------------------------------------

/// A shard whose range is narrowed to `[i_to_outer_key(lo),
/// i_to_outer_key(hi))`, simulating the post-split state where old files may
/// fall outside the range.
fn new_narrow_range_shard(engine: &TestEngine, lo: usize, hi: usize) -> Shard {
    let base = engine.get_shard(1).unwrap();
    Shard::new(
        base.engine_id,
        &base.properties.to_pb(base.id),
        base.ver,
        ShardRange::new(
            &engine.key_builder.i_to_outer_key(lo),
            &engine.key_builder.i_to_outer_key(hi),
        ),
        base.get_data().inner_key_off,
        engine.opts.clone(),
        &engine.master_key,
        engine.encryption_key_manager.clone(),
    )
}

fn install_single_l0(shard: &Shard, l0: L0Table) {
    let mut builder = ShardDataBuilder::new(shard.get_data());
    builder.set_l0_tbls(vec![l0]);
    builder.set_cfs([
        ShardCfBuilder::new(WRITE_CF).build(),
        ShardCf::new(LOCK_CF),
        ShardCf::new(EXTRA_CF),
    ]);
    builder.set_persisted_version(SnapVersion::new(100, 100));
    shard.set_data(builder.build());
    shard.refresh_estimated_size_and_entries();
}

#[test]
fn test_candidate_l0_considers_out_of_bound_split_leftover() {
    let engine = new_engine_with_ratio(0.3);
    // An all-tombstone, gc-able L0 below the shard range [key20, key80). The
    // finder no longer applies a shard-bound filter, so this split-leftover is
    // now considered like any other reclaimable file.
    let out_shard = new_narrow_range_shard(&engine, 20, 80);
    install_single_l0(&out_shard, build_l0_with_tombs(&engine, 901, 0, 10, 50, 1));
    assert!(
        out_shard.find_file_gc_candidate_l0(100).is_some(),
        "out-of-bound split-leftover L0 is now considered"
    );

    // An in-bound L0 is found just the same.
    let in_shard = new_narrow_range_shard(&engine, 20, 80);
    install_single_l0(&in_shard, build_l0_with_tombs(&engine, 902, 30, 70, 50, 1));
    assert!(
        in_shard.find_file_gc_candidate_l0(100).is_some(),
        "in-bound L0 should still be found"
    );
}

#[test]
fn test_file_gc_hint_includes_out_of_bound_split_leftover() {
    let engine = new_engine_with_ratio(0.3);
    // The wakeup hint no longer applies a shard-bound filter either, so an
    // out-of-bound split-leftover feeds the hint like any other file.
    let out_shard = new_narrow_range_shard(&engine, 20, 80);
    install_single_l0(&out_shard, build_l0_with_tombs(&engine, 903, 0, 10, 55, 1));
    assert_eq!(
        out_shard.file_gc_candidate_min_max_ts(),
        55,
        "out-of-bound split-leftover now contributes to the wakeup hint"
    );
}

/// Install a WRITE_CF `(level -> ssts)` layout into a detached shard, used for
/// the post-split out-of-bound dispatch tests where the shard range is narrow.
fn install_write_cf_levels(shard: &Shard, write_cf: Vec<(usize, Vec<SsTable>)>) {
    let mut write_cf_builder = ShardCfBuilder::new(WRITE_CF);
    for (level, ssts) in write_cf {
        for sst in ssts {
            write_cf_builder.add_table(sst, level);
        }
    }
    let mut builder = ShardDataBuilder::new(shard.get_data());
    builder.set_cfs([
        write_cf_builder.build(),
        ShardCf::new(LOCK_CF),
        ShardCf::new(EXTRA_CF),
    ]);
    builder.set_persisted_version(SnapVersion::new(100, 100));
    shard.set_data(builder.build());
    shard.refresh_estimated_size_and_entries();
}

#[test]
fn test_compact_dispatch_file_gc_l1_plus_reclaims_interior_past_split_leftover() {
    let engine = new_engine_with_ratio(0.3);
    // The shard owns [key100, key200). A clean split-leftover file straddles
    // the left edge, so the whole L2 level reads as out-of-bound. An
    // all-tombstone interior file sits fully inside the range and is gc-able.
    // `find_file_gc_candidate_level` no longer applies a shard-bound filter, so
    // it flags L2 on the interior file; the dispatch must reclaim that interior
    // file instead of collapsing the candidate set to the clean boundary file
    // and bailing.
    let boundary_leftover = build_sst_with_tombs(&engine, 950, 80, 120, 50, 0);
    let interior_candidate = build_sst_with_tombs(&engine, 951, 130, 170, 50, 1);
    let shard = Arc::new(new_narrow_range_shard(&engine, 100, 200));
    install_write_cf_levels(
        &shard,
        vec![(2, vec![boundary_leftover, interior_candidate])],
    );
    shard.set_active(true);
    shard.initial_flushed.store(true, Ordering::Release);
    set_gc_safe_point(&engine, 100);

    let (level, ratio) = shard
        .find_file_gc_candidate_level(100)
        .expect("find_file_gc_candidate_level sees the interior candidate past the leftover");
    assert_eq!(level, 2);
    assert!(shard.try_set_file_gc_triggered_l1_plus_priority(level, ratio));

    let cs = run_compact(&engine, &shard)
        .expect("file-GC must pick the in-bound interior candidate, not bail on the leftover")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    // The interior candidate is reclaimed; the clean leftover stays untouched.
    assert!(comp.get_top_deletes().contains(&951));
    assert!(!comp.get_top_deletes().contains(&950));
}

// Right-edge mirror of the interior reclaim test. The clean split-leftover
// straddles the UPPER edge, so the level reads out-of-bound on the high side --
// the clamping's take-last branch, which would collapse the candidate set to
// the clean boundary file. File-GC must still reclaim the in-bound interior
// file.
#[test]
fn test_compact_dispatch_file_gc_l1_plus_reclaims_interior_before_right_split_leftover() {
    let engine = new_engine_with_ratio(0.3);
    // The shard owns [key100, key200). The interior all-tombstone candidate is
    // fully in-bound; a clean split-leftover straddles the right edge, so the
    // first table is in-bound (take-last path), not the first (take-first path).
    let interior_candidate = build_sst_with_tombs(&engine, 952, 120, 160, 50, 1);
    let boundary_leftover = build_sst_with_tombs(&engine, 953, 180, 220, 50, 0);
    let shard = Arc::new(new_narrow_range_shard(&engine, 100, 200));
    install_write_cf_levels(
        &shard,
        vec![(2, vec![interior_candidate, boundary_leftover])],
    );
    shard.set_active(true);
    shard.initial_flushed.store(true, Ordering::Release);
    set_gc_safe_point(&engine, 100);

    let (level, ratio) = shard
        .find_file_gc_candidate_level(100)
        .expect("find_file_gc_candidate_level sees the interior candidate before the leftover");
    assert_eq!(level, 2);
    assert!(shard.try_set_file_gc_triggered_l1_plus_priority(level, ratio));

    let cs = run_compact(&engine, &shard)
        .expect("file-GC must pick the in-bound interior candidate, not bail on the leftover")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    assert!(comp.get_top_deletes().contains(&952));
    assert!(!comp.get_top_deletes().contains(&953));
}

// L0 counterpart end-to-end: the L0 path has no shard-bound clamping, so an
// out-of-bound split-leftover candidate must be reclaimed through the full
// dispatch, not just seen by the finder.
#[test]
fn test_compact_dispatch_file_gc_l0_reclaims_out_of_bound_split_leftover() {
    let engine = new_engine_with_ratio(0.3);
    // The shard owns [key20, key80). An all-tombstone, gc-able L0 straddles the
    // left edge (out-of-bound), the only L0 with an empty L1 below it.
    let leftover = build_l0_with_tombs(&engine, 954, 0, 30, 50, 1);
    let shard = Arc::new(new_narrow_range_shard(&engine, 20, 80));
    install_single_l0(&shard, leftover);
    shard.set_active(true);
    shard.initial_flushed.store(true, Ordering::Release);
    set_gc_safe_point(&engine, 100);

    let (candidate_id, ratio) = shard
        .find_file_gc_candidate_l0(100)
        .expect("find_file_gc_candidate_l0 sees the out-of-bound split-leftover");
    assert_eq!(candidate_id, 954);
    assert!(shard.try_set_file_gc_triggered_l0_priority(candidate_id, ratio));

    let cs = run_compact(&engine, &shard)
        .expect("file-GC must reclaim the out-of-bound L0, not bail")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    assert!(comp.get_top_deletes().contains(&954));
}

// ---------------------------------------------------------------------------
// OldVersion driver: find_file_gc_candidate_level / find_file_gc_candidate_l0
// ---------------------------------------------------------------------------

#[test]
/// A file whose old_data/size ratio exceeds the threshold triggers a hit.
fn test_old_version_candidate_level_basic_hit() {
    let engine = new_engine_with_ratio(0.1);
    // Two MVCC versions per key: ~half the SST bytes land in old-data section.
    let sst = build_sst_with_old_versions(&engine, 1001, 0, 200, 50);
    let lower = build_sst_with_tombs(&engine, 1002, 0, 20, 40, 0);
    install_layout(&engine, vec![], vec![(2, vec![sst]), (3, vec![lower])]);
    let (level, ratio) = engine
        .get_shard(1)
        .unwrap()
        .find_file_gc_candidate_level(200)
        .expect("expected old-version hit");
    assert_eq!(level, 2);
    assert!(
        ratio > 0.1,
        "old_data ratio {} should exceed threshold",
        ratio
    );
}

#[test]
/// A file below the old-data threshold produces no candidate.
fn test_old_version_candidate_level_miss_when_ratio_below_threshold() {
    // Threshold of 0.9 is nearly impossible to satisfy — the SST won't qualify.
    let engine = new_engine_with_ratio(0.9);
    let sst = build_sst_with_old_versions(&engine, 1010, 0, 200, 50);
    let lower = build_sst_with_tombs(&engine, 1011, 0, 20, 40, 0);
    install_layout(&engine, vec![], vec![(2, vec![sst]), (3, vec![lower])]);
    assert!(
        engine
            .get_shard(1)
            .unwrap()
            .find_file_gc_candidate_level(200)
            .is_none()
    );
}

#[test]
/// max_ts above gc_safe_point gates OldVersion just as it gates Tombstone.
fn test_old_version_candidate_level_miss_when_max_ts_too_new() {
    let engine = new_engine_with_ratio(0.1);
    // max_ts=200 > gc_safe_point=100: the file is not yet safe to compact.
    let sst = build_sst_with_old_versions(&engine, 1020, 0, 200, 200);
    let lower = build_sst_with_tombs(&engine, 1021, 0, 20, 40, 0);
    install_layout(&engine, vec![], vec![(2, vec![sst]), (3, vec![lower])]);
    assert!(
        engine
            .get_shard(1)
            .unwrap()
            .find_file_gc_candidate_level(100)
            .is_none()
    );
}

#[test]
/// L0 old-version candidate: a file with high old_data/size ratio triggers a
/// hit when max_ts is at or below the gc safe point.
fn test_old_version_candidate_l0_basic_hit() {
    let engine = new_engine_with_ratio(0.1);
    let l0 = build_l0_with_old_versions(&engine, 1030, 0, 200, 50);
    install_layout(&engine, vec![l0], vec![]);
    let (file_id, ratio) = engine
        .get_shard(1)
        .unwrap()
        .find_file_gc_candidate_l0(100)
        .expect("expected old-version L0 hit");
    assert_eq!(file_id, 1030);
    assert!(
        ratio > 0.1,
        "old_data ratio {} should exceed threshold",
        ratio
    );
}

#[test]
/// Dispatch: setting an OldVersion L1+ priority triggers Engine::compact.
fn test_compact_dispatch_old_version_l1_plus_rewrites_candidate() {
    let engine = new_engine_with_ratio(0.1);
    // Candidate at L2 with two versions per key, max_ts below safe point.
    let sst = build_sst_with_old_versions(&engine, 1040, 0, 200, 50);
    let lower = build_sst_with_tombs(&engine, 1041, 0, 20, 40, 0);
    let shard = install_layout(&engine, vec![], vec![(2, vec![sst]), (3, vec![lower])]);
    set_gc_safe_point(&engine, 100);
    let (level, ratio) = shard
        .find_file_gc_candidate_level(100)
        .expect("expected old-version candidate");
    assert!(shard.try_set_file_gc_triggered_l1_plus_priority(level, ratio));
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    assert!(comp.get_top_deletes().contains(&1040));
}

// ---------------------------------------------------------------------------
// Unified ratio: a file that mixes tombstones and old versions, where neither
// share alone clears the threshold but the combined ratio does.
// ---------------------------------------------------------------------------

/// Build an SST mixing tombstones and old versions so neither sub-ratio
/// (tombs/entries, old/entries) clears 0.3 but the unified
/// `(tombs + old_entries) / (entries + old_entries)` does. Every 5th key's
/// latest is a DELETE (tombs/entries = 0.2); a disjoint 1-in-5 set carries an
/// extra older version (old_entries/entries = 0.2).
fn build_sst_mixed_tombs_and_old(
    engine: &TestEngine,
    id: u64,
    begin: usize,
    end: usize,
    max_ts: u64,
) -> SsTable {
    let block_size = engine.opts.table_builder_options.block_size;
    let comp_tp = engine.opts.table_builder_options.compression_tps[0];
    let comp_lvl = engine.opts.table_builder_options.compression_lvl;
    let mut builder = table::sstable::builder::Builder::new(
        id,
        block_size,
        comp_tp,
        comp_lvl,
        ChecksumType::Crc32,
        None,
    );
    for i in begin..end {
        let key = engine.key_builder.i_to_inner_key(i);
        let payload = key.as_ref().repeat(2);
        if i % 5 == 0 {
            // Latest is a tombstone, single version.
            let tomb = table::Value::new_with_meta_version(BIT_DELETE, max_ts, 0, &[]);
            builder.add(key.as_ref(), &tomb, None).unwrap();
        } else if i % 5 == 1 {
            // Two versions: the older one becomes an old-section entry.
            let newer = table::Value::new_with_meta_version(0, max_ts, 0, &payload);
            builder.add(key.as_ref(), &newer, None).unwrap();
            let older = table::Value::new_with_meta_version(0, max_ts - 1, 0, &payload);
            builder.add(key.as_ref(), &older, None).unwrap();
        } else {
            let put = table::Value::new_with_meta_version(0, max_ts, 0, &payload);
            builder.add(key.as_ref(), &put, None).unwrap();
        }
    }
    let mut data_buf = Vec::new();
    builder.finish(0, &mut data_buf).unwrap();
    let data = Bytes::from(data_buf);
    let opts = crate::dfs::Options::default();
    let fs = engine.fs.as_ref();
    let runtime = fs.get_runtime();
    runtime.block_on(fs.create(id, data.clone(), opts)).unwrap();
    let file = InMemFile::new(id, data);
    SsTable::new(Arc::new(file), BlockCache::None, EncryptionState::Disabled).unwrap()
}

/// L0 counterpart of `build_sst_mixed_tombs_and_old`; entries go to WRITE_CF.
fn build_l0_mixed_tombs_and_old(
    engine: &TestEngine,
    id: u64,
    begin: usize,
    end: usize,
    max_ts: u64,
) -> L0Table {
    let block_size = engine.opts.table_builder_options.block_size;
    let mut builder = L0Builder::new(id, block_size, max_ts.into(), ChecksumType::Crc32, None);
    for i in begin..end {
        let key = engine.key_builder.i_to_inner_key(i);
        let payload = key.as_ref().repeat(2);
        if i % 5 == 0 {
            let tomb = table::Value::new_with_meta_version(BIT_DELETE, max_ts, 0, &[]);
            builder.add(WRITE_CF, key.as_ref(), &tomb, None).unwrap();
        } else if i % 5 == 1 {
            let newer = table::Value::new_with_meta_version(0, max_ts, 0, &payload);
            builder.add(WRITE_CF, key.as_ref(), &newer, None).unwrap();
            let older = table::Value::new_with_meta_version(0, max_ts - 1, 0, &payload);
            builder.add(WRITE_CF, key.as_ref(), &older, None).unwrap();
        } else {
            let put = table::Value::new_with_meta_version(0, max_ts, 0, &payload);
            builder.add(WRITE_CF, key.as_ref(), &put, None).unwrap();
        }
    }
    let (_, data) = builder.finish().unwrap();
    let opts = crate::dfs::Options::default();
    let fs = engine.fs.as_ref();
    let runtime = fs.get_runtime();
    runtime.block_on(fs.create(id, data.clone(), opts)).unwrap();
    let file = Arc::new(InMemFile::new(id, data));
    L0Table::new(file, BlockCache::None, false, EncryptionState::Disabled)
        .unwrap()
        .unwrap()
}

#[test]
fn test_mixed_file_triggers_under_unified_ratio() {
    // tombs/entries = 0.2 and old/entries = 0.2 — neither old per-driver 0.3
    // threshold would have fired, but the unified
    // (tombs + old_entries) / (entries + old_entries) = 0.33 does.
    let engine = new_engine_with_ratio(0.3);
    let mixed = build_sst_mixed_tombs_and_old(&engine, 1100, 0, 500, 50);
    let lower = build_sst_with_tombs(&engine, 1101, 0, 20, 40, 0);
    install_layout(&engine, vec![], vec![(2, vec![mixed]), (3, vec![lower])]);
    let (level, ratio) = engine
        .get_shard(1)
        .unwrap()
        .find_file_gc_candidate_level(100)
        .expect("mixed file should trigger under the unified ratio");
    assert_eq!(level, 2);
    assert!(
        (0.3..0.4).contains(&ratio),
        "unified ratio {} should clear 0.3 while neither sub-ratio (0.2) would",
        ratio
    );
}

#[test]
fn test_l0_mixed_triggers_under_unified_ratio() {
    // The L0 ratio uses entries_write_cf() + old_entries_write_cf(): tombs and
    // old versions each at 0.2 combine to clear the 0.3 threshold.
    let engine = new_engine_with_ratio(0.3);
    let l0 = build_l0_mixed_tombs_and_old(&engine, 1110, 0, 500, 50);
    install_layout(&engine, vec![l0], vec![]);
    let (file_id, ratio) = engine
        .get_shard(1)
        .unwrap()
        .find_file_gc_candidate_l0(100)
        .expect("mixed L0 should trigger under the unified ratio");
    assert_eq!(file_id, 1110);
    assert!(
        (0.3..0.4).contains(&ratio),
        "unified L0 ratio {} should clear 0.3 while neither sub-ratio (0.2) would",
        ratio
    );
}

// ---------------------------------------------------------------------------
// Per-shard wakeup hint: refresh computes it; the finder is gated below it.
// ---------------------------------------------------------------------------

#[test]
fn test_refresh_sets_file_gc_hint_to_min_candidate_max_ts() {
    let engine = new_engine_with_ratio(0.3);
    // Two qualifying tombstone files; the hint must be the smaller max_ts.
    let l2 = build_sst_with_tombs(&engine, 1210, 0, 200, 60, 2);
    let l3 = build_sst_with_tombs(&engine, 1211, 0, 200, 80, 2);
    let shard = install_layout(&engine, vec![], vec![(2, vec![l2]), (3, vec![l3])]);
    assert_eq!(shard.file_gc_candidate_min_max_ts(), 60);
}

#[test]
fn test_refresh_sets_file_gc_hint_to_u64_max_when_no_candidate() {
    let engine = new_engine_with_ratio(0.3);
    // 25% tombstones is below the 0.3 ratio, so no file qualifies.
    let l2 = build_sst_with_tombs(&engine, 1220, 0, 400, 60, 4);
    let shard = install_layout(&engine, vec![], vec![(2, vec![l2])]);
    assert_eq!(shard.file_gc_candidate_min_max_ts(), u64::MAX);
}

#[test]
fn test_file_gc_hint_is_a_safe_lower_bound_for_the_finder() {
    let engine = new_engine_with_ratio(0.3);
    // The hint is the min cheap-gate max_ts. Below it the finder (which also
    // gates on max_ts <= gc_safe_point) finds nothing, so the tick's
    // `gc_safe_point < hint` skip can never drop a real trigger.
    let candidate = build_sst_with_tombs(&engine, 1200, 0, 200, 200, 2);
    let lower = build_sst_with_tombs(&engine, 1201, 0, 20, 40, 0);
    let shard = install_layout(
        &engine,
        vec![],
        vec![(2, vec![candidate]), (3, vec![lower])],
    );
    let hint = shard.file_gc_candidate_min_max_ts();
    assert_eq!(hint, 200, "hint is the candidate's max_ts");
    // gc_safe_point below the hint: nothing has matured, finder is empty.
    assert!(shard.find_file_gc_candidate_level(hint - 1).is_none());
    // gc_safe_point at the hint: the candidate matures and is found.
    assert!(shard.find_file_gc_candidate_level(hint).is_some());
}

#[test]
fn test_refresh_recomputes_parked_file_gc_hint() {
    let engine = new_engine_with_ratio(0.3);
    let l2 = build_sst_with_tombs(&engine, 1230, 0, 200, 60, 2);
    let shard = install_layout(&engine, vec![], vec![(2, vec![l2])]);
    // Park the hint at u64::MAX, as an overlap-gated scan does.
    shard.set_file_gc_candidate_min_max_ts(u64::MAX);
    assert_eq!(shard.file_gc_candidate_min_max_ts(), u64::MAX);
    // A refresh (every file-set mutation routes through one) recomputes the
    // hint from the live layout, so a reclaimable file reopens the scan.
    shard.refresh_estimated_size_and_entries();
    assert_eq!(shard.file_gc_candidate_min_max_ts(), 60);
}

#[test]
fn test_refresh_candidate_hint_recomputes_parked_hint() {
    let engine = new_engine_with_ratio(0.3);
    let l2 = build_sst_with_tombs(&engine, 1231, 0, 200, 60, 2);
    let shard = install_layout(&engine, vec![], vec![(2, vec![l2])]);
    // The live layout has a reclaimable candidate, so the hint is set.
    let live_hint = shard.file_gc_candidate_min_max_ts();
    assert_ne!(live_hint, u64::MAX);
    // Park the hint at u64::MAX, as an overlap-gated scan does.
    shard.set_file_gc_candidate_min_max_ts(u64::MAX);
    // The leader tick calls this public entry after an online config change;
    // it must recompute the hint from the live layout under the read lock.
    shard.refresh_candidate_hint();
    assert_eq!(shard.file_gc_candidate_min_max_ts(), live_hint);
}

#[test]
fn test_park_keeps_a_hint_refreshed_after_the_tick_read_it() {
    let engine = new_engine_with_ratio(0.3);
    let l2 = build_sst_with_tombs(&engine, 1260, 0, 200, 60, 2);
    let shard = install_layout(&engine, vec![], vec![(2, vec![l2])]);
    // The tick reads the hint before it scans.
    let observed = shard.file_gc_candidate_min_max_ts();
    assert_eq!(observed, 60);
    // A concurrent refresh recomputes a fresh hint before the tick parks.
    shard.set_file_gc_candidate_min_max_ts(70);
    // Parking against the stale observed value must leave the fresh hint, not
    // clobber it to u64::MAX.
    shard.park_file_gc_candidate_hint(observed);
    assert_eq!(shard.file_gc_candidate_min_max_ts(), 70);
}

#[test]
fn test_park_sets_max_when_the_hint_is_unchanged() {
    let engine = new_engine_with_ratio(0.3);
    let l2 = build_sst_with_tombs(&engine, 1261, 0, 200, 60, 2);
    let shard = install_layout(&engine, vec![], vec![(2, vec![l2])]);
    let observed = shard.file_gc_candidate_min_max_ts();
    // No refresh raced, so the park applies.
    shard.park_file_gc_candidate_hint(observed);
    assert_eq!(shard.file_gc_candidate_min_max_ts(), u64::MAX);
}

#[test]
fn test_file_gc_hint_matches_finder_for_l2_old_version_file() {
    let engine = new_engine_with_ratio(0.3);
    // An L2 old-version file (tombs = 0) is reachable only when the narrowing
    // counter includes lv2plus_old_entries. The hint and the finder must agree.
    let l2 = build_sst_with_old_versions(&engine, 1240, 0, 200, 60);
    let lower = build_sst_with_tombs(&engine, 1241, 0, 20, 40, 0);
    let shard = install_layout(&engine, vec![], vec![(2, vec![l2]), (3, vec![lower])]);
    // The narrowing counter is driven by old entries here, not tombstones.
    assert!(shard.lv2plus_old_entries.load(Ordering::Relaxed) > 0);
    let (level, ratio) = shard
        .find_file_gc_candidate_level(100)
        .expect("L2 old-version file must be found");
    assert_eq!(level, 2);
    assert!(ratio >= 0.3);
    // The hint sees the same file, so it never suppresses this trigger.
    assert_eq!(shard.file_gc_candidate_min_max_ts(), 60);
}

#[test]
fn test_file_gc_hint_mirrors_allow_l3_gate() {
    // A reclaimable file only at the bottommost level. The hint must mirror the
    // finder's allow_l3 gate: counted when L3 is on, ignored when L3 is gated
    // off (production default), so an idle shard is not woken for a scan the
    // finder would always reject.

    // L3 enabled: the bottommost-level file sets the hint, and the finder picks it.
    let engine_on = new_engine_with_ratio(0.3);
    let on_file = build_sst_with_tombs(&engine_on, 813, 0, 200, 60, 1);
    let shard_on = install_layout(&engine_on, vec![], vec![(3, vec![on_file])]);
    assert_eq!(shard_on.file_gc_candidate_min_max_ts(), 60);
    assert!(shard_on.find_file_gc_candidate_level(100).is_some());

    // L3 gated off: the lv2+ counter still routes the scan to the bottommost level,
    // but the allow_l3 gate now excludes it, so the hint parks at u64::MAX and
    // the finder agrees there is nothing to do.
    let (engine_off, _) =
        new_test_engine_opt_with_custom_options(true, DEF_BLOCK_SIZE, TABLE_KEY_PREFIX, |opts| {
            opts.set_file_gc_compaction_ratio(0.3);
        });
    let off_file = build_sst_with_tombs(&engine_off, 814, 0, 200, 60, 1);
    let shard_off = install_layout(&engine_off, vec![], vec![(3, vec![off_file])]);
    assert!(shard_off.lv2plus_tombs.load(Ordering::Relaxed) > 0);
    assert_eq!(shard_off.file_gc_candidate_min_max_ts(), u64::MAX);
    assert!(shard_off.find_file_gc_candidate_level(100).is_none());
}

#[test]
fn test_file_gc_hint_excludes_overlap_gated_l0() {
    // An L0 file that passes ratio but is L1-overlap-gated must not enter the
    // hint, mirroring find_file_gc_candidate_l0. Otherwise it would wake the
    // tick for a scan the finder always rejects.
    let engine = new_engine_with_ratio(0.3);
    let gated = build_l0_with_tombs(&engine, 960, 0, 60, 50, 1);
    let l1 = build_huge_lower_ssts(&engine, 961);
    let shard = install_layout(&engine, vec![gated], vec![(1, l1)]);
    assert_eq!(shard.file_gc_candidate_min_max_ts(), u64::MAX);
    assert!(shard.find_file_gc_candidate_l0(100).is_none());
}

#[test]
fn test_file_gc_overlap_gated_files_do_not_bury_later_candidates() {
    // A (ts=10) and C (ts=30) are overlap-gated at L1; B (ts=20) and D (ts=40)
    // are eligible at the bottommost level. The hint must skip A/C and track the
    // earliest eligible file, so B then D each trigger as the safe point
    // advances. The pre-fix ratio-only hint took A's ts=10, scanned empty, and
    // parked at u64::MAX, burying both B and D.
    let engine = new_engine_with_ratio(0.3);
    let a = build_sst_with_tombs(&engine, 901, 0, 60, 10, 1);
    let b = build_sst_with_tombs(&engine, 902, 200, 260, 20, 1);
    let c = build_sst_with_tombs(&engine, 903, 100, 160, 30, 1);
    let d = build_sst_with_tombs(&engine, 904, 300, 360, 40, 1);
    let huge = build_huge_lower_ssts(&engine, 910);
    let shard = install_layout(
        &engine,
        vec![],
        vec![(1, vec![a, c]), (2, huge), (3, vec![b, d])],
    );

    // The hint points at the earliest eligible file B, not the gated A.
    assert_eq!(shard.file_gc_candidate_min_max_ts(), 20);
    assert!(shard.find_file_gc_candidate_level(10).is_none());

    // First trigger: B reclaimed at the bottommost level.
    let (level, ratio) = shard
        .find_file_gc_candidate_level(20)
        .expect("B must be the first candidate");
    assert_eq!(level, 3);
    assert!(shard.try_set_file_gc_triggered_l1_plus_priority(level, ratio));
    set_gc_safe_point(&engine, 20);
    let cs = run_compact(&engine, &shard)
        .expect("first compaction picked")
        .expect("first compaction succeeded");
    assert!(cs.get_compaction().get_bottom_deletes().contains(&902));

    // Reinstall the post-compaction layout (B gone). A and C are still gated,
    // so the hint advances to D rather than parking at u64::MAX.
    let a = build_sst_with_tombs(&engine, 921, 0, 60, 10, 1);
    let c = build_sst_with_tombs(&engine, 923, 100, 160, 30, 1);
    let d = build_sst_with_tombs(&engine, 924, 300, 360, 40, 1);
    let huge = build_huge_lower_ssts(&engine, 930);
    let shard = install_layout(
        &engine,
        vec![],
        vec![(1, vec![a, c]), (2, huge), (3, vec![d])],
    );
    assert_eq!(shard.file_gc_candidate_min_max_ts(), 40);
    assert!(shard.find_file_gc_candidate_level(30).is_none());

    // Second trigger: D reclaimed, proving it was never buried.
    let (level, ratio) = shard
        .find_file_gc_candidate_level(40)
        .expect("D must be the second candidate");
    assert_eq!(level, 3);
    *shard.compaction_priority.write().unwrap() = None;
    assert!(shard.try_set_file_gc_triggered_l1_plus_priority(level, ratio));
    set_gc_safe_point(&engine, 40);
    let cs = run_compact(&engine, &shard)
        .expect("second compaction picked")
        .expect("second compaction succeeded");
    assert!(cs.get_compaction().get_bottom_deletes().contains(&924));
}

// ---------------------------------------------------------------------------
// File-GC obsolete-file distribution: classify_write_cf_obsolete.
// ---------------------------------------------------------------------------

#[test]
fn test_classify_write_cf_obsolete_counts() {
    let engine = new_engine_with_ratio(0.3);
    // 50%-tombstone SST at L2 -> obsolete ratio 0.5 -> ratio bucket 5; zero
    // old-data-size -> size bucket 0; ratio-passing over empty L3 -> overlap
    // bucket 0.
    let tomb_sst = build_sst_with_tombs(&engine, 201, 0, 200, 50, 2);
    // A clean SST (no tombstones, no old versions) -> early-out, not counted.
    let clean = build_sst_with_tombs(&engine, 202, 1000, 1100, 50, 0);
    install_layout(&engine, vec![], vec![(2, vec![tomb_sst, clean])]);

    let shard = engine.get_shard(1).unwrap();
    let mut buckets = ObsoleteFileBuckets::default();
    shard.classify_write_cf_obsolete(&mut buckets);

    // Only the obsolete file is counted, once in each distribution.
    assert_eq!(
        buckets.ratio.iter().sum::<u64>(),
        1,
        "clean file must be skipped"
    );
    assert_eq!(buckets.ratio[5], 1);
    assert_eq!(buckets.size.iter().sum::<u64>(), 1);
    assert_eq!(buckets.size[0], 1);
    assert_eq!(buckets.overlap.iter().sum::<u64>(), 1);
    assert_eq!(buckets.overlap[0], 1);
}

#[test]
fn test_classify_write_cf_obsolete_overlap_above_cap() {
    let engine = new_engine_with_ratio(0.3);
    // Tiny ratio-passing candidate at L2 over huge L3 -> overlap far above cap.
    let candidate = build_sst_with_tombs(&engine, 210, 0, 60, 50, 2);
    let lower = build_huge_lower_ssts(&engine, 211);
    install_layout(&engine, vec![], vec![(2, vec![candidate]), (3, lower)]);

    let shard = engine.get_shard(1).unwrap();
    let mut buckets = ObsoleteFileBuckets::default();
    shard.classify_write_cf_obsolete(&mut buckets);

    // The candidate is counted once in the overlap distribution, in a bucket
    // above the default cap (15x): nothing in the <15x buckets.
    assert_eq!(buckets.overlap.iter().sum::<u64>(), 1);
    assert_eq!(
        buckets.overlap[0] + buckets.overlap[1] + buckets.overlap[2],
        0
    );
    // It also shows in the ratio distribution (ratio 0.5).
    assert_eq!(buckets.ratio[5], 1);
}

// ---------------------------------------------------------------------------
// Expansion boundaries: skipped bottom tables, multi-lower picks, partial
// overlap
// ---------------------------------------------------------------------------

// File-GC right expansion stops at a skipped bottom table. The neighbor
// overlaps a lower file, but a different lower file lies between it and the
// seed's lower that neither covers; folding the neighbor would skip that bottom
// table, so the shared `left > pick.lower_right_idx` guard breaks the
// expansion.
#[test]
fn test_compact_dispatch_file_gc_l1_plus_right_expansion_stops_at_gap() {
    let engine = new_engine_with_ratio(0.3);
    let seed = build_sst_with_tombs(&engine, 8101, 50, 80, 50, 2);
    let neighbor = build_sst_with_tombs(&engine, 8102, 300, 350, 50, 0);
    let low_seed = build_sst_with_tombs(&engine, 8103, 50, 80, 40, 0);
    let low_gap = build_sst_with_tombs(&engine, 8104, 150, 200, 40, 0);
    let low_nb = build_sst_with_tombs(&engine, 8105, 300, 350, 40, 0);
    let shard = install_layout(
        &engine,
        vec![],
        vec![
            (2, vec![seed, neighbor]),
            (3, vec![low_seed, low_gap, low_nb]),
        ],
    );
    set_gc_safe_point(&engine, 100);
    assert!(shard.try_set_file_gc_triggered_l1_plus_priority(2, 0.5));
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    assert!(comp.get_top_deletes().contains(&8101));
    assert_eq!(
        comp.get_top_deletes().len(),
        1,
        "the gap must stop expansion: top_deletes={:?}",
        comp.get_top_deletes()
    );
    assert!(comp.get_bottom_deletes().contains(&8103));
    assert!(
        !comp.get_bottom_deletes().contains(&8104) && !comp.get_bottom_deletes().contains(&8105),
        "the gap and the neighbor's lower file must stay out: bottom_deletes={:?}",
        comp.get_bottom_deletes()
    );
}

// File-GC left expansion stops at a skipped bottom table (mirror of the right
// case): the shared `right < pick.lower_left_idx` guard breaks the leftward
// expansion.
#[test]
fn test_compact_dispatch_file_gc_l1_plus_left_expansion_stops_at_gap() {
    let engine = new_engine_with_ratio(0.3);
    let neighbor = build_sst_with_tombs(&engine, 8111, 50, 100, 50, 0);
    let seed = build_sst_with_tombs(&engine, 8112, 300, 330, 50, 2);
    let low_nb = build_sst_with_tombs(&engine, 8113, 50, 100, 40, 0);
    let low_gap = build_sst_with_tombs(&engine, 8114, 150, 200, 40, 0);
    let low_seed = build_sst_with_tombs(&engine, 8115, 300, 330, 40, 0);
    let shard = install_layout(
        &engine,
        vec![],
        vec![
            (2, vec![neighbor, seed]),
            (3, vec![low_nb, low_gap, low_seed]),
        ],
    );
    set_gc_safe_point(&engine, 100);
    assert!(shard.try_set_file_gc_triggered_l1_plus_priority(2, 0.5));
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    assert!(comp.get_top_deletes().contains(&8112));
    assert_eq!(
        comp.get_top_deletes().len(),
        1,
        "the gap must stop expansion: top_deletes={:?}",
        comp.get_top_deletes()
    );
    assert!(comp.get_bottom_deletes().contains(&8115));
}

// Size-driven expansion respects the same skipped-bottom-table guard. Here
// `require_lower_overlap` is false, so only the shared gap guard applies --
// this locks that the file-GC param did not change the size-driven boundary.
#[test]
fn test_compact_dispatch_size_driven_l1_plus_expansion_stops_at_gap() {
    let engine = new_engine_with_ratio(0.3);
    let seed = build_sst_with_tombs(&engine, 8121, 50, 80, 50, 0);
    let neighbor = build_sst_with_tombs(&engine, 8122, 300, 350, 50, 0);
    // A tiny lower under the seed gives it the max ratio; a full-size lower
    // under the neighbor keeps the neighbor's ratio low.
    let low_seed = build_sst_with_tombs(&engine, 8123, 50, 60, 40, 0);
    let low_gap = build_sst_with_tombs(&engine, 8124, 150, 200, 40, 0);
    let low_nb = build_sst_with_tombs(&engine, 8125, 300, 350, 40, 0);
    let shard = install_layout(
        &engine,
        vec![],
        vec![
            (2, vec![seed, neighbor]),
            (3, vec![low_seed, low_gap, low_nb]),
        ],
    );
    {
        let mut priority = shard.compaction_priority.write().unwrap();
        *priority = Some(CompactionPriority::L1Plus {
            cf: WRITE_CF as isize,
            level: 2,
            score: 1.5,
            file_gc: false,
        });
    }
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    assert!(
        comp.get_top_deletes().contains(&8121),
        "size-driven seeds on the max-ratio table"
    );
    assert!(
        !comp.get_top_deletes().contains(&8122),
        "the gap stops size-driven expansion too: top_deletes={:?}",
        comp.get_top_deletes()
    );
}

// A file-GC candidate spanning several lower files pulls every overlapping
// lower file into the rewrite, so no lower file is left straddling the output.
#[test]
fn test_compact_dispatch_file_gc_l1_plus_includes_all_overlapping_lower_files() {
    let engine = new_engine_with_ratio(0.3);
    let seed = build_sst_with_tombs(&engine, 8131, 0, 300, 50, 2);
    let low_a = build_sst_with_tombs(&engine, 8132, 0, 100, 40, 0);
    let low_b = build_sst_with_tombs(&engine, 8133, 100, 200, 40, 0);
    let low_c = build_sst_with_tombs(&engine, 8134, 200, 300, 40, 0);
    let shard = install_layout(
        &engine,
        vec![],
        vec![(2, vec![seed]), (3, vec![low_a, low_b, low_c])],
    );
    set_gc_safe_point(&engine, 100);
    assert!(shard.try_set_file_gc_triggered_l1_plus_priority(2, 0.5));
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    assert!(comp.get_top_deletes().contains(&8131));
    for id in [8132, 8133, 8134] {
        assert!(
            comp.get_bottom_deletes().contains(&id),
            "lower file {} overlapping the seed must be an input: bottom={:?}",
            id,
            comp.get_bottom_deletes()
        );
    }
}

// A neighbor that only partially overlaps the shared lower file is still folded
// in: leaving it out would leave the rewritten lower file straddling it. The
// neighbor extends past the lower file's upper bound but shares its lower half.
#[test]
fn test_compact_dispatch_file_gc_l1_plus_folds_partially_overlapping_neighbor() {
    let engine = new_engine_with_ratio(0.3);
    let seed = build_sst_with_tombs(&engine, 8141, 50, 100, 50, 2);
    let neighbor = build_sst_with_tombs(&engine, 8142, 100, 250, 50, 0);
    // The lower file spans the seed fully and the neighbor's lower half only.
    let wide_low = build_sst_with_tombs(&engine, 8143, 50, 200, 40, 0);
    let shard = install_layout(
        &engine,
        vec![],
        vec![(2, vec![seed, neighbor]), (3, vec![wide_low])],
    );
    set_gc_safe_point(&engine, 100);
    assert!(shard.try_set_file_gc_triggered_l1_plus_priority(2, 0.5));
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    assert!(
        comp.get_top_deletes().contains(&8141) && comp.get_top_deletes().contains(&8142),
        "the partially-overlapping neighbor must be folded in: top_deletes={:?}",
        comp.get_top_deletes()
    );
    assert!(comp.get_bottom_deletes().contains(&8143));
}

// ---------------------------------------------------------------------------
// Pure helpers: rate limiter, scheduling score, reclaimable ratio
// ---------------------------------------------------------------------------

// The per-node trigger limiter allows the first trigger, refuses a second
// inside the window, allows again once the window elapses, and disables on
// interval 0.
#[test]
fn test_file_gc_trigger_rate_limiter_bounds_frequency() {
    let limiter = FileGcCompactionTriggerLimiter::new();
    let interval = 1_000u64;
    // First trigger is always allowed and claims the window.
    assert!(limiter.try_trigger(5_000, interval));
    // A second trigger inside the window is refused, and peeks as limited.
    assert!(!limiter.try_trigger(5_500, interval));
    assert!(limiter.is_rate_limited(5_500, interval));
    // Once the window elapses the next trigger is allowed again.
    assert!(!limiter.is_rate_limited(6_000, interval));
    assert!(limiter.try_trigger(6_000, interval));
    // interval == 0 disables limiting entirely.
    assert!(limiter.try_trigger(6_001, 0));
    assert!(!limiter.is_rate_limited(6_001, 0));
}

// The file-GC scheduling score never reaches the foreground L0/L1 range (>
// 1.0), so a reclaim compaction can never preempt user-facing compaction.
#[test]
fn test_file_gc_priority_score_stays_below_foreground() {
    assert!((file_gc_priority_score(0.3) - 0.69).abs() < 1e-9);
    assert!((file_gc_priority_score(1.0) - 0.90).abs() < 1e-9);
    assert!(file_gc_priority_score(1.0) < 1.0);
    let file_gc = CompactionPriority::L1Plus {
        cf: WRITE_CF as isize,
        level: 2,
        score: file_gc_priority_score(1.0),
        file_gc: true,
    };
    let foreground = CompactionPriority::L1Plus {
        cf: WRITE_CF as isize,
        level: 2,
        score: 1.5,
        file_gc: false,
    };
    assert!(file_gc.score() < foreground.score());
}

// The reclaimable ratio counts both tombstones and old versions over the live
// entry count: all-tombstone is 1.0, a clean single-version file is 0.0, and a
// two-versions-per-key file is about half. The L0 helper uses the same formula.
#[test]
fn test_file_gc_ratio_counts_tombstones_and_old_versions() {
    let engine = new_engine_with_ratio(0.3);
    let all_tomb = build_sst_with_tombs(&engine, 8161, 0, 100, 50, 1);
    assert!((file_gc_ratio(&all_tomb).unwrap() - 1.0).abs() < 1e-9);
    let clean = build_sst_with_tombs(&engine, 8162, 0, 100, 50, 0);
    assert!(file_gc_ratio(&clean).unwrap().abs() < 1e-9);
    let old_ver = build_sst_with_old_versions(&engine, 8163, 0, 100, 50);
    let r = file_gc_ratio(&old_ver).unwrap();
    assert!(r > 0.4 && r < 0.6, "old-version ratio out of range: {}", r);
    let l0_all_tomb = build_l0_with_tombs(&engine, 8164, 0, 100, 50, 1);
    assert!((l0_file_gc_ratio(&l0_all_tomb).unwrap() - 1.0).abs() < 1e-9);
}

// ---------------------------------------------------------------------------
// Old-version bottommost-level in-place rewrite
// ---------------------------------------------------------------------------

// A bottommost-level file whose keys each carry an old plus a new version: the
// in-place rewrite drops the old versions but keeps the latest, so it still
// produces output (unlike an all-tombstone drop that produces nothing).
#[test]
fn test_compact_dispatch_old_version_last_level_inplace_keeps_latest() {
    let engine = new_engine_with_ratio(0.3);
    let bottommost_level = engine
        .get_shard(1)
        .unwrap()
        .get_data()
        .get_cf(WRITE_CF)
        .levels
        .last()
        .unwrap()
        .level;
    let candidate = build_sst_with_old_versions(&engine, 8151, 0, 200, 50);
    let shard = install_layout(&engine, vec![], vec![(bottommost_level, vec![candidate])]);
    set_gc_safe_point(&engine, 100);
    let (level, ratio) = shard
        .find_file_gc_candidate_level(100)
        .expect("old-version bottommost-level hit");
    assert_eq!(level, bottommost_level);
    assert!(shard.try_set_file_gc_triggered_l1_plus_priority(level, ratio));
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    // In-place encoding: empty upper slot, candidate in the bottom slot at
    // bottommost_level - 1 so the output writes back to the bottommost level.
    assert_eq!(comp.get_level() as usize, bottommost_level - 1);
    assert!(comp.get_top_deletes().is_empty());
    assert!(comp.get_bottom_deletes().contains(&8151));
    // The latest versions survive, so the rewrite produces output.
    assert!(!comp.get_table_creates().is_empty());
}

// ---------------------------------------------------------------------------
// Online config change end-to-end: real dispatch path changes finder behavior
// ---------------------------------------------------------------------------

// Lowering the ratio through the real `KvEngineConfigManager::dispatch` path
// lets the finder accept a candidate the previous high ratio rejected.
#[test]
fn test_online_lowering_ratio_enables_finder_candidate() {
    let engine = new_engine_with_ratio(0.9);
    let candidate = build_sst_with_tombs(&engine, 8171, 0, 200, 50, 2);
    let lower = build_sst_with_tombs(&engine, 8172, 0, 20, 40, 0);
    let shard = install_layout(
        &engine,
        vec![],
        vec![(2, vec![candidate]), (3, vec![lower])],
    );
    set_gc_safe_point(&engine, 100);
    // At ratio 0.9 the 50%-tombstone candidate is below threshold.
    assert!(shard.find_file_gc_candidate_level(100).is_none());
    let cfg = Config {
        file_gc_compaction_ratio: 0.9,
        file_gc_compaction_allow_l3: true,
        ..Default::default()
    };
    let new_cfg = Config {
        file_gc_compaction_ratio: 0.3,
        ..cfg.clone()
    };
    let change = cfg.diff(&new_cfg);
    let mut manager = KvEngineConfigManager::new(
        Arc::new(VersionTrack::new(cfg.clone())),
        IaCtx::disabled(&cfg.ia),
        engine.opts.clone(),
    );
    manager.dispatch(change).unwrap();
    assert_eq!(engine.opts.file_gc_compaction_ratio(), 0.3);
    // The finder now accepts the candidate the high ratio used to reject.
    assert_eq!(
        shard.find_file_gc_candidate_level(100).map(|(l, _)| l),
        Some(2)
    );
}

// Flipping `allow_l3` on through the real dispatch path exposes a
// bottommost-level candidate the production default (off) keeps gated.
#[test]
fn test_online_enabling_allow_l3_exposes_last_level_candidate() {
    let (engine, _) =
        new_test_engine_opt_with_custom_options(true, DEF_BLOCK_SIZE, TABLE_KEY_PREFIX, |opts| {
            opts.set_file_gc_compaction_ratio(0.3);
            opts.set_file_gc_compaction_allow_l3(false);
        });
    let bottommost_level = engine
        .get_shard(1)
        .unwrap()
        .get_data()
        .get_cf(WRITE_CF)
        .levels
        .last()
        .unwrap()
        .level;
    let candidate = build_sst_with_tombs(&engine, 8181, 0, 200, 50, 2);
    let shard = install_layout(&engine, vec![], vec![(bottommost_level, vec![candidate])]);
    set_gc_safe_point(&engine, 100);
    // allow_l3 off: the bottommost-level candidate is gated.
    assert!(shard.find_file_gc_candidate_level(100).is_none());
    let cfg = Config {
        file_gc_compaction_ratio: 0.3,
        ..Default::default()
    };
    let new_cfg = Config {
        file_gc_compaction_allow_l3: true,
        ..cfg.clone()
    };
    let change = cfg.diff(&new_cfg);
    let mut manager = KvEngineConfigManager::new(
        Arc::new(VersionTrack::new(cfg.clone())),
        IaCtx::disabled(&cfg.ia),
        engine.opts.clone(),
    );
    manager.dispatch(change).unwrap();
    assert!(engine.opts.file_gc_compaction_allow_l3());
    assert_eq!(
        shard.find_file_gc_candidate_level(100).map(|(l, _)| l),
        Some(bottommost_level)
    );
}

// ---------------------------------------------------------------------------
// LSM invariant: a file-GC rewrite leaves no same-level overlap
// ---------------------------------------------------------------------------

// The rewrite output spans the whole candidate range, so any lower file inside
// that range that was NOT taken as an input would overlap the output once
// applied. The candidate spans three overlapping L3 files plus a disjoint L3
// sibling: all three overlapping files must be absorbed and the sibling left
// out, so the rewritten L3 stays non-overlapping. (The companion MVCC-ordering
// invariant -- a newer version never sits below an older one -- is proven at
// the data level by the cross-level resurrection integration test.)
#[test]
fn test_file_gc_l1_plus_absorbs_all_overlapping_lower_keeps_disjoint_sibling() {
    let engine = new_engine_with_ratio(0.3);
    let seed = build_sst_with_tombs(&engine, 8201, 0, 300, 50, 2);
    let low_a = build_sst_with_tombs(&engine, 8202, 0, 100, 40, 0);
    let low_b = build_sst_with_tombs(&engine, 8203, 100, 200, 40, 0);
    let low_c = build_sst_with_tombs(&engine, 8204, 200, 300, 40, 0);
    let sibling = build_sst_with_tombs(&engine, 8205, 400, 500, 40, 0);
    let shard = install_layout(
        &engine,
        vec![],
        vec![(2, vec![seed]), (3, vec![low_a, low_b, low_c, sibling])],
    );
    set_gc_safe_point(&engine, 100);
    assert!(shard.try_set_file_gc_triggered_l1_plus_priority(2, 0.5));
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    assert!(comp.get_top_deletes().contains(&8201));
    assert_eq!(
        comp.get_top_deletes().len(),
        1,
        "only the seed is the upper input: {:?}",
        comp.get_top_deletes()
    );
    // Every lower file overlapping the candidate range is absorbed.
    for id in [8202, 8203, 8204] {
        assert!(
            comp.get_bottom_deletes().contains(&id),
            "overlapping lower {} must be absorbed: {:?}",
            id,
            comp.get_bottom_deletes()
        );
    }
    // The disjoint sibling is left out, so it cannot overlap the output.
    assert!(
        !comp.get_bottom_deletes().contains(&8205),
        "disjoint sibling must not be touched: {:?}",
        comp.get_bottom_deletes()
    );
}

// ---------------------------------------------------------------------------
// Move-down eligibility and lower-gap edge cases
// ---------------------------------------------------------------------------

// A multi-CF-format L0 (magic != SPLIT_L0) can never move down, even with no
// overlap: `try_move_down_l0` bails when no L0 is `is_write_cf_only`. So a lone
// regular L0 candidate is rewritten to L1 (fresh id), not relocated -- unlike
// the split-L0 move-down.
#[test]
fn test_compact_dispatch_file_gc_l0_multi_cf_rewrites_not_moves_down() {
    let engine = new_engine_with_ratio(0.3);
    // build_l0_with_tombs produces a regular (multi-CF-format) L0; empty L1.
    let candidate = build_l0_with_tombs(&engine, 8301, 0, 200, 50, 2);
    let shard = install_layout(&engine, vec![candidate], vec![]);
    set_gc_safe_point(&engine, 100);
    let (candidate_id, ratio) = shard
        .find_file_gc_candidate_l0(100)
        .expect("expected L0 candidate");
    assert_eq!(candidate_id, 8301);
    assert!(shard.try_set_file_gc_triggered_l0_priority(candidate_id, ratio));
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    assert!(comp.get_top_deletes().contains(&8301));
    // Rewritten, not moved: no output reuses the candidate id.
    assert!(comp.get_table_creates().iter().all(|t| t.get_id() != 8301));
}

// A multi-CF-format L0 that overlaps L1 is likewise rewritten, merging the L1
// overlap into the output.
#[test]
fn test_compact_dispatch_file_gc_l0_multi_cf_with_l1_overlap_rewrites() {
    let engine = new_engine_with_ratio(0.3);
    let candidate = build_l0_with_tombs(&engine, 8311, 0, 200, 50, 2);
    let l1 = build_sst_with_tombs(&engine, 8312, 0, 20, 40, 0);
    let shard = install_layout(&engine, vec![candidate], vec![(1, vec![l1])]);
    set_gc_safe_point(&engine, 100);
    let (candidate_id, ratio) = shard
        .find_file_gc_candidate_l0(100)
        .expect("expected L0 candidate");
    assert_eq!(candidate_id, 8311);
    assert!(shard.try_set_file_gc_triggered_l0_priority(candidate_id, ratio));
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    assert!(comp.get_top_deletes().contains(&8311));
    assert!(comp.get_bottom_deletes().contains(&8312));
    assert!(comp.get_table_creates().iter().all(|t| t.get_id() != 8311));
}

// A split-L0 candidate blocked from move-down specifically by an overlapping
// sibling split-L0 (`overlap_below_l0`), forcing a rewrite of the overlap set.
// The sibling is movable on its own, but the candidate is not, so the dispatch
// drops the sibling move-down and rewrites the candidate instead.
#[test]
fn test_compact_dispatch_file_gc_l0_overlap_below_blocks_move_down() {
    let engine = new_engine_with_ratio(0.3);
    // Candidate (all-tomb, higher ratio) overlaps an older split-L0 sibling.
    let candidate = build_split_l0_with_tombs(&engine, 8341, 50, 150, 60, 1);
    let sibling = build_split_l0_with_tombs(&engine, 8342, 100, 200, 50, 2);
    let shard = install_layout(&engine, vec![candidate, sibling], vec![]);
    set_gc_safe_point(&engine, 100);
    let (candidate_id, ratio) = shard
        .find_file_gc_candidate_l0(100)
        .expect("expected L0 candidate");
    assert_eq!(candidate_id, 8341);
    assert!(shard.try_set_file_gc_triggered_l0_priority(candidate_id, ratio));
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    assert!(comp.get_top_deletes().contains(&8341));
    // Rewritten (merging the overlapping sibling), not moved: id not preserved.
    assert!(comp.get_table_creates().iter().all(|t| t.get_id() != 8341));
}

// Multi-lower pick AND neighbor folding at once: the seed spans two L3 files,
// and the left neighbor shares the first of them. The rewrite must fold the
// neighbor in and take both lower files.
#[test]
fn test_compact_dispatch_file_gc_l1_plus_folds_neighbor_over_multi_lower() {
    let engine = new_engine_with_ratio(0.3);
    let neighbor = build_sst_with_tombs(&engine, 8321, 0, 50, 50, 0);
    let seed = build_sst_with_tombs(&engine, 8322, 50, 150, 50, 2);
    // Two older L3 files under the seed; low_a also underlies the neighbor.
    let low_a = build_sst_with_tombs(&engine, 8323, 0, 80, 40, 0);
    let low_b = build_sst_with_tombs(&engine, 8324, 80, 150, 40, 0);
    let shard = install_layout(
        &engine,
        vec![],
        vec![(2, vec![neighbor, seed]), (3, vec![low_a, low_b])],
    );
    set_gc_safe_point(&engine, 100);
    assert!(shard.try_set_file_gc_triggered_l1_plus_priority(2, 0.5));
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    assert!(
        comp.get_top_deletes().contains(&8321) && comp.get_top_deletes().contains(&8322),
        "the neighbor sharing the first lower file must be folded in: {:?}",
        comp.get_top_deletes()
    );
    for id in [8323, 8324] {
        assert!(
            comp.get_bottom_deletes().contains(&id),
            "both overlapping lower files must be inputs: {:?}",
            comp.get_bottom_deletes()
        );
    }
}

// A file-gc candidate that overlaps NO lower file while the lower level is
// non-empty: the pick takes the candidate alone with an empty lower slot, and
// the disjoint lower file stays untouched.
#[test]
fn test_compact_dispatch_file_gc_l1_plus_candidate_in_lower_gap() {
    let engine = new_engine_with_ratio(0.3);
    let candidate = build_sst_with_tombs(&engine, 8331, 100, 150, 50, 2);
    // A non-empty L3 that lies entirely beyond the candidate's range.
    let far_low = build_sst_with_tombs(&engine, 8332, 500, 600, 40, 0);
    let shard = install_layout(
        &engine,
        vec![],
        vec![(2, vec![candidate]), (3, vec![far_low])],
    );
    set_gc_safe_point(&engine, 100);
    assert!(shard.try_set_file_gc_triggered_l1_plus_priority(2, 0.5));
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    assert!(comp.get_top_deletes().contains(&8331));
    assert!(
        !comp.get_bottom_deletes().contains(&8332),
        "the disjoint lower file must not be an input: {:?}",
        comp.get_bottom_deletes()
    );
}

// ---------------------------------------------------------------------------
// Driver dispatch and maturity boundary
// ---------------------------------------------------------------------------

// Old-version driver through the full L0 dispatch (finder-only was covered).
#[test]
fn test_compact_dispatch_old_version_l0_rewrites_candidate() {
    let engine = new_engine_with_ratio(0.1);
    let l0 = build_l0_with_old_versions(&engine, 8401, 0, 200, 50);
    let shard = install_layout(&engine, vec![l0], vec![]);
    set_gc_safe_point(&engine, 100);
    let (candidate_id, ratio) = shard
        .find_file_gc_candidate_l0(100)
        .expect("expected old-version L0 candidate");
    assert_eq!(candidate_id, 8401);
    assert!(shard.try_set_file_gc_triggered_l0_priority(candidate_id, ratio));
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    assert!(cs.get_compaction().get_top_deletes().contains(&8401));
}

// Unified driver (mixed tombstones + old versions) through the full L1+
// dispatch (finder-only was covered).
#[test]
fn test_compact_dispatch_unified_l1_plus_rewrites_candidate() {
    let engine = new_engine_with_ratio(0.3);
    let mixed = build_sst_mixed_tombs_and_old(&engine, 8411, 0, 500, 50);
    let lower = build_sst_with_tombs(&engine, 8412, 0, 20, 40, 0);
    let shard = install_layout(&engine, vec![], vec![(2, vec![mixed]), (3, vec![lower])]);
    set_gc_safe_point(&engine, 100);
    let (level, ratio) = shard
        .find_file_gc_candidate_level(100)
        .expect("expected unified-ratio candidate");
    assert!(shard.try_set_file_gc_triggered_l1_plus_priority(level, ratio));
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    assert!(cs.get_compaction().get_top_deletes().contains(&8411));
}

// Maturity boundary: the gate is `max_ts > safe_point`, so max_ts == safe_point
// is mature and max_ts == safe_point + 1 is too new. file_a is picked, file_b
// is skipped.
#[test]
fn test_file_gc_maturity_boundary_at_exact_safe_point() {
    let engine = new_engine_with_ratio(0.3);
    let file_a = build_sst_with_tombs(&engine, 8421, 0, 100, 100, 2);
    let file_b = build_sst_with_tombs(&engine, 8423, 200, 300, 101, 2);
    let shard = install_layout(&engine, vec![], vec![(2, vec![file_a, file_b])]);
    set_gc_safe_point(&engine, 100);
    let (level, ratio) = shard
        .find_file_gc_candidate_level(100)
        .expect("max_ts == safe point is mature and must be found");
    assert_eq!(level, 2);
    assert!(shard.try_set_file_gc_triggered_l1_plus_priority(level, ratio));
    let cs = run_compact(&engine, &shard)
        .expect("compaction picked")
        .expect("compaction succeeded");
    let comp = cs.get_compaction();
    assert!(
        comp.get_top_deletes().contains(&8421),
        "max_ts == safe point must be compacted"
    );
    assert!(
        !comp.get_top_deletes().contains(&8423),
        "max_ts == safe point + 1 must be skipped"
    );
}