wbftree 0.1.0

//! RangeIndex 管理器实现 (1:1 对标 Garnet RangeIndexManager.cs)
//!
//! 负责协调 BfTree 的生命周期、数据文件预分阶段 (Pre-Stage)、检查点 CPR 快照与全量故障恢复。

use std::{
  fs,
  hint::spin_loop,
  io::Read,
  iter::repeat_n,
  path::{Path, PathBuf},
  str,
  sync::{
    Arc,
    atomic::{AtomicBool, Ordering},
  },
  thread::yield_now,
};

use parking_lot::RwLock;
use whasher::{GxPapayaMap, fast_hash, hash128, new_papaya_map};

use crate::{
  error::{Error, Result},
  service::BfTreeService,
  stub::{RANGE_INDEX_STUB_SIZE, RangeIndexStub},
  types::{BfTreeConfig, StorageBackend, StorageBackendType, TreeTuning},
};

/// 短临界区自旋等待退避阈值，超过后让出 CPU 时间片
const SPIN_LIMIT: usize = 32;

/// 编译期计算的 128 位哈希前缀种子密钥
const PREFIX_SEED_1: u64 = 0x27bb_2ee6_87b0_b0fd;
const PREFIX_SEED_2: u64 = 0x517c_c1b7_2722_0a95;

/// 将 u128 格式化为零填充到指定位数的十六进制字符串
#[inline]
fn hex_padded_128(value: u128, width: usize) -> String {
  const HEX: &[u8; 16] = b"0123456789abcdef";
  let mut buf = [0u8; 32];
  let mut v = value;
  let mut pos = 32;
  while v > 0 {
    pos -= 1;
    buf[pos] = HEX[(v & 0xf) as usize];
    v >>= 4;
  }
  let hex_len = 32 - pos;
  let pad = width.saturating_sub(hex_len);
  let mut s = String::with_capacity(pad + hex_len);
  s.extend(repeat_n('0', pad));
  s.push_str(unsafe { str::from_utf8_unchecked(&buf[pos..]) });
  s
}

/// 键哈希前缀长度 (128 位哈希的 32 字符小写十六进制编码，1:1 对标 Garnet HashPrefixLength)
const HASH_PREFIX_LEN: usize = 32;

/// 刷盘快照文件名中逻辑地址段的十六进制位数 (1:1 对标 Garnet AddrHexLength，格式 {addr:x16})
const ADDR_HEX_LEN: usize = 16;

/// 数据文件标准后缀
const DATA_FILE_SUFFIX: &str = ".data.bftree";
/// 刷盘快照文件标准后缀
const FLUSH_FILE_SUFFIX: &str = ".flush.bftree";
/// 树文件通用后缀
const TREE_FILE_SUFFIX: &str = ".bftree";

/// CPR 快照文件魔数 (底层 bf-tree 快照格式头部标识)
const CPR_MAGIC: &[u8; 16] = b"BF-TREE-V0-BEGIN";
/// CPR 快照魔数长度
const CPR_MAGIC_LEN: usize = CPR_MAGIC.len();

/// 锁条带默认数量 (128 分段降低热路径锁竞争，1:1 对标 Garnet rangeIndexLocks)
pub const NUM_LOCK_STRIPES: usize = 128;
const STRIPE_MASK: usize = NUM_LOCK_STRIPES - 1;

/// 缓存行 128 字节对齐的读写锁包装器，消除相邻条带锁在 ARM64/Apple Silicon 及 x86-64 上的 CPU 伪共享 (False Sharing)
#[repr(align(128))]
pub struct CacheAlignedLock(pub RwLock<()>);

impl CacheAlignedLock {
  /// 创建新的缓存行对齐锁
  #[inline]
  pub fn new() -> Self {
    Self(RwLock::new(()))
  }
}

impl Default for CacheAlignedLock {
  #[inline]
  fn default() -> Self {
    Self::new()
  }
}

/// 针对键哈希分段的读写条带锁 (1:1 对标 Garnet rangeIndexLocks 与 ReadOptimizedLock)
///
/// 条带数恒为 [`NUM_LOCK_STRIPES`]，由构造函数保证。
pub struct RangeIndexLocks {
  stripes: Box<[CacheAlignedLock]>,
}

impl Default for RangeIndexLocks {
  fn default() -> Self {
    Self::new()
  }
}

impl RangeIndexLocks {
  /// 创建锁条带实例（逐条带堆上就地构造，避免 16KB 数组先在栈上成型再整体拷贝）
  pub fn new() -> Self {
    let mut stripes = Vec::with_capacity(NUM_LOCK_STRIPES);
    stripes.extend((0..NUM_LOCK_STRIPES).map(|_| CacheAlignedLock::new()));
    Self {
      stripes: stripes.into_boxed_slice(),
    }
  }

  /// 获取指定键哈希的读锁（共享锁）
  #[inline]
  pub fn read(&self, key_hash: u64) -> parking_lot::RwLockReadGuard<'_, ()> {
    // 安全：STRIPE_MASK = 127 保证下标严格在 [0..128) 范围内，完全避免越界
    let idx = (key_hash as usize) & STRIPE_MASK;
    unsafe { self.stripes.get_unchecked(idx) }.0.read()
  }

  /// 获取指定键哈希的写锁（互斥锁）
  #[inline]
  pub fn write(&self, key_hash: u64) -> parking_lot::RwLockWriteGuard<'_, ()> {
    // 安全：STRIPE_MASK = 127 保证下标严格在 [0..128) 范围内，完全避免越界
    let idx = (key_hash as usize) & STRIPE_MASK;
    unsafe { self.stripes.get_unchecked(idx) }.0.write()
  }
}

/// 待复制的 RangeIndex 文件条目 (1:1 对标 Garnet RangeIndexFileEntry)
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RangeIndexFileEntry {
  /// 主节点上的完整文件路径
  pub path: PathBuf,
  /// 32 字符键哈希前缀
  pub key_hash: String,
  /// HybridLog 逻辑地址 (用于刷盘文件；快照文件为 0)
  pub address: i64,
  /// 是否为刷盘文件 (true 为 flush 文件，false 为检查点快照)
  pub is_flush_file: bool,
}

/// 单树条目 (1:1 对标 Garnet TreeEntry)
pub struct TreeEntry {
  /// 托管的在线 BfTreeService 实例
  pub tree: RwLock<Option<Arc<BfTreeService>>>,
  /// 键哈希值 (用于锁分段)
  pub key_hash: u64,
  /// 128 位唯一键 ID
  pub key_id: u128,
  /// 32 位十六进制前缀
  pub hash_prefix: String,
  /// 是否处于快照中
  pub snapshot_pending: AtomicBool,
  /// 快照防重入原子锁
  pub snapshot_in_progress: AtomicBool,
}

impl TreeEntry {
  /// 创建新条目
  pub fn new(
    tree: Option<Arc<BfTreeService>>,
    key_hash: u64,
    key_id: u128,
    hash_prefix: String,
  ) -> Self {
    Self {
      tree: RwLock::new(tree),
      key_hash,
      key_id,
      hash_prefix,
      snapshot_pending: AtomicBool::new(false),
      snapshot_in_progress: AtomicBool::new(false),
    }
  }

  /// 尝试获取快照原子锁
  #[inline]
  pub fn try_claim_snapshot(&self) -> bool {
    self
      .snapshot_in_progress
      .compare_exchange(false, true, Ordering::AcqRel, Ordering::Relaxed)
      .is_ok()
  }

  /// 释放快照原子锁
  #[inline]
  pub fn release_snapshot(&self) {
    self.snapshot_in_progress.store(false, Ordering::Release);
  }

  /// 在防重入快照锁保护下执行 CPR 快照 (1:1 对标 Garnet TreeEntry.SnapshotUnderClaim)
  pub fn snapshot_under_claim(&self, tree: &BfTreeService, destination_path: &Path) -> Result<()> {
    let mut spins = 0usize;
    while !self.try_claim_snapshot() {
      if spins < SPIN_LIMIT {
        spin_loop();
      } else {
        yield_now();
      }
      spins = spins.wrapping_add(1);
    }
    let _guard = SnapshotClaimGuard(self);
    tree.cpr_snapshot(destination_path)
  }
}

/// 快照防重入 RAII 守卫，确保在任何退出或异常路径下释放 snapshot_in_progress
struct SnapshotClaimGuard<'a>(&'a TreeEntry);
impl Drop for SnapshotClaimGuard<'_> {
  fn drop(&mut self) {
    self.0.release_snapshot();
  }
}

/// 检查点 RAII 守卫，确保在任何退出路径下重置 checkpoint_in_progress 并清理 snapshot_pending
struct CheckpointGuard<'a>(&'a RangeIndexManager);
impl Drop for CheckpointGuard<'_> {
  fn drop(&mut self) {
    self.0.clear_checkpoint_barrier();
  }
}

/// 快照待处理 RAII 守卫，确保单树快照结束（无论成功或异常退出）后重置 snapshot_pending
struct SnapshotPendingGuard<'a>(&'a TreeEntry);
impl Drop for SnapshotPendingGuard<'_> {
  fn drop(&mut self) {
    self.0.snapshot_pending.store(false, Ordering::SeqCst);
  }
}

/// 默认迁移分块大小 (256KB，1:1 对标 Garnet RangeIndexManager.DefaultMigrationChunkSize)
pub const DEFAULT_MIGRATION_CHUNK_SIZE: usize = 256 * 1024;
/// 索引存根字节大小 (35 字节，1:1 对标 Garnet RangeIndexManager.IndexSizeBytes)
pub const INDEX_SIZE_BYTES: usize = RANGE_INDEX_STUB_SIZE;

/// RangeIndex 管理器 (1:1 对标 Garnet RangeIndexManager)
pub struct RangeIndexManager {
  /// 数据文件根目录 ({ri_log_root}/{hashPrefix}.data.bftree)
  ri_log_root: PathBuf,
  /// 检查点快照根目录 ({cpr_dir}/{token}/rangeindex/{hashPrefix}.bftree)
  cpr_dir: PathBuf,
  /// 迁移临时目录 ({ri_log_root}/migration-tmp/)
  migration_temp_dir: PathBuf,
  /// 在线索引字典 (按 128 位 key_id 纯整数索引，零堆分配，基于 papaya 高性能无锁并发字典与硬件向量加速 gxhash)
  live_indexes: GxPapayaMap<u128, Arc<TreeEntry>>,
  /// 全局检查点进行中标记
  checkpoint_in_progress: AtomicBool,
  /// 键哈希分段读写条带锁
  locks: RangeIndexLocks,
}

impl RangeIndexManager {
  /// 从根目录创建管理器实例 (cpr 目录默认为 ri_log_root/cpr，1:1 对标 Garnet new RangeIndexManager(rootPath, null))
  pub fn from_root(ri_log_root: impl Into<PathBuf>) -> Self {
    let root = ri_log_root.into();
    let cpr = root.join("cpr");
    Self::new(root, cpr)
  }

  /// 创建管理器实例
  pub fn new(ri_log_root: impl Into<PathBuf>, cpr_dir: impl Into<PathBuf>) -> Self {
    let ri_log_root = ri_log_root.into();
    let cpr_dir = cpr_dir.into();

    let _ = fs::create_dir_all(&ri_log_root);
    let _ = fs::create_dir_all(&cpr_dir);

    let migration_temp_dir = ri_log_root.join("migration-tmp");
    if migration_temp_dir.exists() {
      let _ = fs::remove_dir_all(&migration_temp_dir);
    }
    let _ = fs::create_dir_all(&migration_temp_dir);

    Self {
      ri_log_root,
      cpr_dir,
      migration_temp_dir,
      live_indexes: new_papaya_map(),
      checkpoint_in_progress: AtomicBool::new(false),
      locks: RangeIndexLocks::new(),
    }
  }

  /// 生成临时迁移文件路径 ({ri_log_root}/migration-tmp/{random_id}.bftree) (1:1 对标 Garnet DeriveTempMigrationPath)
  #[inline]
  pub fn derive_temp_migration_path(&self) -> PathBuf {
    let rand_id = fastrand::u128(..);
    let mut s = hex_padded_128(rand_id, 32);
    s.push_str(TREE_FILE_SUFFIX);
    self.migration_temp_dir.join(s)
  }

  /// 释放管理器资源并释放所有在线树 (1:1 对标 Garnet IDisposable.Dispose)
  pub fn dispose(&self) {
    let pin = self.live_indexes.pin();
    for entry in pin.values() {
      if let Some(tree) = entry.tree.write().take() {
        tree.dispose();
      }
    }
    pin.clear();
  }

  /// 获取锁条带管理器引用
  #[inline]
  pub fn locks(&self) -> &RangeIndexLocks {
    &self.locks
  }

  /// 计算键的 128 位唯一 ID (零堆分配，用于内存字典极速索引，1:1 对标 Garnet KeyId)
  #[inline]
  pub fn key_id_of(key: &[u8]) -> u128 {
    hash128(key, PREFIX_SEED_1, PREFIX_SEED_2)
  }

  /// 根据键计算 32 字符十六进制前缀 (1:1 对标 Garnet HashKeyToPrefix)
  #[inline]
  pub fn hash_prefix_of(key: &[u8]) -> String {
    let id = Self::key_id_of(key);
    hex_padded_128(id, HASH_PREFIX_LEN)
  }

  /// 根据键计算 64 位哈希值 (用于锁分段与索引，默认采用 gxhash 硬件向量加速)
  #[inline]
  pub fn key_hash_of(key: &[u8]) -> u64 {
    fast_hash(key)
  }

  /// 根据最大记录大小动态计算叶子页面大小 (1:1 对标 Garnet ComputeLeafPageSize)
  ///
  /// ≤2KB → 4KB；否则 2.5 倍封顶 32KB 后向上取 2 的幂 (纯整数运算，floor(5n/2) 与 C# 浮点截断一致)
  #[inline]
  pub fn compute_leaf_page_size(max_record_size: usize) -> usize {
    if max_record_size <= 2048 {
      return 4096;
    }
    (max_record_size * 5 / 2).min(32768).next_power_of_two()
  }

  /// 数据文件标准路径
  pub fn data_file_path(&self, hash_prefix: &str) -> PathBuf {
    let mut file_name = String::with_capacity(hash_prefix.len() + DATA_FILE_SUFFIX.len());
    file_name.push_str(hash_prefix);
    file_name.push_str(DATA_FILE_SUFFIX);
    self.ri_log_root.join(file_name)
  }

  /// 根据原始 key 获取数据文件标准路径 (1:1 对标 Garnet LogDataPathFor)
  #[inline]
  pub fn data_file_path_for_key(&self, key: &[u8]) -> PathBuf {
    let hash_prefix = Self::hash_prefix_of(key);
    self.data_file_path(&hash_prefix)
  }

  /// 刷盘快照文件标准路径 ({ri_log_root}/{hash_prefix}.{logical_address:016x}.flush.bftree)
  pub fn log_flush_path(&self, hash_prefix: &str, logical_address: i64) -> PathBuf {
    // i64 按二进制补码转 u64 后十六进制格式化，与 std {:#x} 对负数的行为逐字节一致
    let mut s = String::from(hash_prefix);
    s.push('.');
    s.push_str(&hex_padded_128(logical_address as u64 as u128, 16));
    s.push_str(FLUSH_FILE_SUFFIX);
    self.ri_log_root.join(s)
  }

  /// 获取在线索引字典引用 (用于检查点遍历与恢复注册)
  #[inline]
  pub fn live_indexes(&self) -> &GxPapayaMap<u128, Arc<TreeEntry>> {
    &self.live_indexes
  }

  /// 获取所有活跃与就绪的索引条目快照
  fn live_entries(&self) -> Vec<Arc<TreeEntry>> {
    let pin = self.live_indexes.pin();
    pin.values().cloned().collect()
  }

  /// 获取在线树实例（快速共享读路径，1:1 对标 Garnet liveIndexes.TryGetValue）
  #[inline]
  pub fn get_tree(&self, key: &[u8]) -> Option<Arc<BfTreeService>> {
    let key_id = Self::key_id_of(key);
    let pin = self.live_indexes.pin();
    pin
      .get(&key_id)
      .and_then(|e| e.tree.read().as_ref().cloned())
  }

  /// 按 key_id 查询在线树条目 (papaya 无锁读 + 条目与树双 Arc 保活)
  #[inline]
  fn live_tree_of(&self, key_id: u128) -> Option<(Arc<TreeEntry>, Arc<BfTreeService>)> {
    let pin = self.live_indexes.pin();
    pin
      .get(&key_id)
      .and_then(|e| e.tree.read().as_ref().cloned().map(|t| (Arc::clone(e), t)))
  }

  /// 检查点快照文件标准路径
  pub fn checkpoint_snapshot_path(&self, token: &str, hash_prefix: &str) -> PathBuf {
    let mut file_name = String::with_capacity(hash_prefix.len() + TREE_FILE_SUFFIX.len());
    file_name.push_str(hash_prefix);
    file_name.push_str(TREE_FILE_SUFFIX);
    self.cpr_dir.join(token).join("rangeindex").join(file_name)
  }

  /// 获取当前活跃与待激活索引数量
  #[inline]
  pub fn live_index_count(&self) -> usize {
    self.live_indexes.pin().len()
  }

  /// 检查是否正处于全局检查点快照中
  #[inline]
  pub fn is_checkpoint_in_progress(&self) -> bool {
    self.checkpoint_in_progress.load(Ordering::Acquire)
  }

  /// 设置全局检查点屏障 (1:1 对标 Garnet SetCheckpointBarrier)
  ///
  /// 将所有在线与就绪条目的 snapshot_pending 设为 true，并设置 checkpoint_in_progress 为 true
  pub fn set_checkpoint_barrier(&self) {
    let pin = self.live_indexes.pin();
    for entry in pin.values() {
      entry.snapshot_pending.store(true, Ordering::SeqCst);
    }
    self.checkpoint_in_progress.store(true, Ordering::SeqCst);
  }

  /// 清除全局检查点屏障 (1:1 对标 Garnet ClearCheckpointBarrier)
  ///
  /// 将 checkpoint_in_progress 设为 false，并将所有条目的 snapshot_pending 设为 false
  pub fn clear_checkpoint_barrier(&self) {
    self.checkpoint_in_progress.store(false, Ordering::SeqCst);
    let pin = self.live_indexes.pin();
    for entry in pin.values() {
      entry.snapshot_pending.store(false, Ordering::SeqCst);
    }
  }

  /// 等待单树快照完成屏障 (1:1 对标 Garnet WaitForTreeCheckpoint)
  ///
  /// 如果处于检查点中且对应树的 snapshot_pending 为 true，自旋并让出 CPU 等待完成并返回 true
  pub fn wait_for_tree_checkpoint(&self, key: &[u8]) -> bool {
    if !self.checkpoint_in_progress.load(Ordering::Acquire) {
      return false;
    }
    let key_id = Self::key_id_of(key);
    let entry_opt = {
      let pin = self.live_indexes.pin();
      pin.get(&key_id).cloned()
    };
    if let Some(entry) = entry_opt
      && entry.snapshot_pending.load(Ordering::Acquire)
    {
      let mut spins = 0usize;
      while entry.snapshot_pending.load(Ordering::Acquire) {
        if spins < SPIN_LIMIT {
          spin_loop();
        } else {
          yield_now();
        }
        spins = spins.wrapping_add(1);
      }
      return true;
    }
    false
  }

  /// 等待全局检查点完全解除
  #[inline]
  pub fn wait_for_global_checkpoint(&self) {
    let mut spins = 0usize;
    while self.is_checkpoint_in_progress() {
      if spins < SPIN_LIMIT {
        spin_loop();
      } else {
        yield_now();
      }
      spins = spins.wrapping_add(1);
    }
  }

  /// 创建并注册全新的 BfTreeService (持有条带互斥写锁保证并发唯一性)
  pub fn create_bftree(
    &self,
    key: &[u8],
    storage_backend: StorageBackend,
    tuning: TreeTuning,
  ) -> Result<Arc<BfTreeService>> {
    let key_hash = Self::key_hash_of(key);
    let _stripe_lock = self.locks.write(key_hash);
    let key_id = Self::key_id_of(key);
    let hash_prefix = Self::hash_prefix_of(key);
    self.create_bftree_internal(key_id, key_hash, &hash_prefix, storage_backend, tuning)
  }

  /// 仅构建 BfTreeService 实例，不操作 live_indexes 字典
  fn instantiate_tree(
    &self,
    hash_prefix: &str,
    storage_backend: StorageBackend,
    tuning: TreeTuning,
  ) -> Result<Arc<BfTreeService>> {
    let mut config = BfTreeConfig::default();

    let (file_path_str, backend_type) = if storage_backend == StorageBackend::Memory {
      config.cache_only(true);
      (None, StorageBackendType::Memory)
    } else {
      // file_path 同时配置 Std 磁盘后端与数据文件路径
      let data_path = self.data_file_path(hash_prefix);
      config.file_path(&data_path);
      (
        Some(data_path.to_string_lossy().into_owned()),
        StorageBackendType::Disk,
      )
    };

    if tuning.cache_size > 0 {
      config.cb_size_byte(tuning.cache_size);
    }
    if tuning.min_record_size > 0 {
      config.cb_min_record_size(tuning.min_record_size);
    }
    if tuning.max_record_size > 0 {
      config.cb_max_record_size(tuning.max_record_size);
    }
    if tuning.max_key_len > 0 {
      config.cb_max_key_len(tuning.max_key_len);
    }

    let actual_leaf_page_size = if tuning.leaf_page_size > 0 {
      tuning.leaf_page_size
    } else if tuning.max_record_size > 0 {
      Self::compute_leaf_page_size(tuning.max_record_size)
    } else {
      0
    };
    if actual_leaf_page_size > 0 {
      config.leaf_page_size(actual_leaf_page_size);
    }
    config.use_snapshot(true);

    Ok(Arc::new(BfTreeService::new_with_backend(
      config,
      backend_type,
      file_path_str,
    )?))
  }

  fn create_bftree_internal(
    &self,
    key_id: u128,
    key_hash: u64,
    hash_prefix: &str,
    storage_backend: StorageBackend,
    tuning: TreeTuning,
  ) -> Result<Arc<BfTreeService>> {
    // 单次 pin 贯穿查重与注册：调用方已持条带写锁，同 key 并发被串行化
    let pin = self.live_indexes.pin();
    if pin.contains_key(&key_id) {
      return Err(Error::IndexExists);
    }

    let tree = self.instantiate_tree(hash_prefix, storage_backend, tuning)?;
    let entry = Arc::new(TreeEntry::new(
      Some(Arc::clone(&tree)),
      key_hash,
      key_id,
      hash_prefix.to_string(),
    ));

    pin.insert(key_id, entry);
    Ok(tree)
  }

  /// 获取或按需打开在线 BfTreeService (双重检查锁与条带锁保证并发安全性与恢复正确性)
  pub fn get_or_open_tree(&self, key: &[u8], stub: &RangeIndexStub) -> Result<Arc<BfTreeService>> {
    let key_hash = Self::key_hash_of(key);
    let _stripe_lock = self.locks.write(key_hash);
    let key_id = Self::key_id_of(key);

    {
      let pin = self.live_indexes.pin();
      if let Some(entry) = pin.get(&key_id) {
        let tree_guard = entry.tree.read();
        if let Some(t) = tree_guard.as_ref() {
          return Ok(Arc::clone(t));
        }
      }
    }

    let hash_prefix = Self::hash_prefix_of(key);
    let backend = StorageBackendType::from_u8(stub.storage_backend);
    let data_path = self.data_file_path(&hash_prefix);
    let mut flush_name = String::with_capacity(hash_prefix.len() + FLUSH_FILE_SUFFIX.len());
    flush_name.push_str(&hash_prefix);
    flush_name.push_str(FLUSH_FILE_SUFFIX);
    let flush_path = self.ri_log_root.join(flush_name);

    // 磁盘后端才存在可恢复的磁盘工件；内存后端的刷盘文件与工作文件均无意义
    if backend == StorageBackendType::Disk {
      if flush_path.exists() {
        // 拷贝失败必须传播：静默吞掉会回退到陈旧/部分写入的 data.bftree，
        // 恢复出错误树版本 (1:1 对标 C# File.Copy 异常传播语义)
        fs::copy(&flush_path, &data_path)?;
      } else if let Ok(entries) = fs::read_dir(&self.ri_log_root) {
        let mut latest_candidate: Option<(i64, PathBuf)> = None;
        for entry in entries.flatten() {
          let path = entry.path();
          if let Some(name_str) = path.file_name().and_then(|n| n.to_str())
            && let Some((prefix, addr)) = Self::parse_flush_file_name(name_str)
            && prefix == hash_prefix
            && latest_candidate
              .as_ref()
              .is_none_or(|(max_addr, _)| addr > *max_addr)
          {
            latest_candidate = Some((addr, path));
          }
        }
        if let Some((_, path)) = latest_candidate {
          fs::copy(path, &data_path)?;
        }
      }

      // 1:1 对标 C# RestoreTree：pre-stage 不变量保证 TreeHandle=0 的存根必有已预置的
      // data.bftree；缺失说明不变量被破坏（pre-stage 失败或文件被外部删除）。
      // 返回错误显式暴露数据丢失，绝不静默创建空树掩盖问题。
      if !data_path.exists() {
        let mut msg = String::from("数据文件缺失且无可用刷盘快照: ");
        msg.push_str(&data_path.display().to_string());
        return Err(Error::Recovery(msg));
      }
    }

    let is_cpr = backend == StorageBackendType::Disk
      && data_path
        .metadata()
        .map(|m| m.len() >= CPR_MAGIC_LEN as u64)
        .unwrap_or(false)
      && fs::File::open(&data_path).is_ok_and(|mut f| {
        let mut magic = [0u8; CPR_MAGIC_LEN];
        f.read_exact(&mut magic).is_ok() && magic == *CPR_MAGIC
      });

    // 1:1 对标 Garnet RestoreTree: 如果磁盘上已存在数据文件且为快照，严格从快照恢复，否则以已有文件重新打开
    let tree = if is_cpr {
      Arc::new(BfTreeService::recover_from_cpr_snapshot(
        &data_path,
        true,
        StorageBackendType::Disk,
      )?)
    } else {
      self.instantiate_tree(&hash_prefix, backend.into(), TreeTuning::from(stub))?
    };

    // 原地激活现有条目（如 pre_stage 或恢复阶段注册的 pending entry），或者注册全新条目
    let pin = self.live_indexes.pin();
    if let Some(entry) = pin.get(&key_id) {
      *entry.tree.write() = Some(Arc::clone(&tree));
    } else {
      let entry = Arc::new(TreeEntry::new(
        Some(Arc::clone(&tree)),
        key_hash,
        key_id,
        hash_prefix,
      ));
      pin.insert(key_id, entry);
    }
    Ok(tree)
  }

  /// 预分阶段复制并注册就绪条目 (1:1 对标 Garnet PreStageAndRegisterPending)
  ///
  /// 源刷盘快照缺失属于不变量破坏 (1:1 对标 C# 不变量 violation 处理)：绝不回退到其他
  /// 刷盘文件以免恢复出错误树版本，且不注册 pending 条目，让后续 get_or_open_tree
  /// 显式报错暴露数据丢失，而非静默恢复不正确的数据。
  pub fn pre_stage_and_register_pending(&self, key: &[u8], src_flush_address: i64) -> Result<()> {
    let key_hash = Self::key_hash_of(key);
    let _stripe_lock = self.locks.write(key_hash);
    let key_id = Self::key_id_of(key);
    let hash_prefix = Self::hash_prefix_of(key);
    let snapshot_path = self.log_flush_path(&hash_prefix, src_flush_address);
    if !snapshot_path.exists() {
      return Ok(());
    }
    let data_path = self.data_file_path(&hash_prefix);
    fs::copy(&snapshot_path, &data_path)?;

    let entry = Arc::new(TreeEntry::new(None, key_hash, key_id, hash_prefix));
    self.live_indexes.pin().insert(key_id, entry);
    Ok(())
  }

  /// 从内存字典中移除条目并释放其底层树实例
  #[inline]
  fn remove_and_dispose_entry(&self, key_id: u128) -> bool {
    if let Some(entry) = self.live_indexes.pin().remove(&key_id).cloned() {
      if let Some(tree) = entry.tree.write().take() {
        tree.dispose();
      }
      true
    } else {
      false
    }
  }

  /// 注销并释放指定树条目 (1:1 对标 Garnet UnregisterIndex)
  pub fn unregister_index(&self, key: &[u8]) -> bool {
    let key_hash = Self::key_hash_of(key);
    let _stripe_lock = self.locks.write(key_hash);
    let key_id = Self::key_id_of(key);
    self.remove_and_dispose_entry(key_id)
  }

  /// 删除指定索引并彻底清理磁盘文件 (1:1 对标 Garnet DisposeTreeUnderLock)
  ///
  /// 先注销并释放在线树（关闭引擎持有的文件句柄），再删除工作文件：
  /// 与 C# DisposeAndDeleteFilesDeferred 的「先 dispose 后 delete」顺序一致
  pub fn delete_index(&self, key: &[u8]) -> bool {
    let key_hash = Self::key_hash_of(key);
    let _stripe_lock = self.locks.write(key_hash);
    let removed = self.remove_and_dispose_entry(Self::key_id_of(key));
    let data_path = self.data_file_path(&Self::hash_prefix_of(key));
    if data_path.exists() {
      let _ = fs::remove_file(data_path);
    }
    removed
  }

  /// 销毁并释放指定树条目 (1:1 对标 Garnet DisposeTreeUnderLock)
  ///
  /// delete_file=true (DEL/UNLINK) 时同时删除工作文件 data.bftree (刷盘快照保留，
  /// 由 on_truncate 按日志地址回收)；false (淘汰) 时仅注销条目保留文件供惰性恢复
  pub fn dispose_tree(&self, key: &[u8], delete_file: bool) -> bool {
    if delete_file {
      self.delete_index(key)
    } else {
      self.unregister_index(key)
    }
  }

  /// 销毁并释放指定树条目，校验存根转移标志 (1:1 对标 Garnet DisposeTreeUnderLock)
  pub fn dispose_tree_under_lock(
    &self,
    key: &[u8],
    stub: &RangeIndexStub,
    delete_files: bool,
  ) -> bool {
    if !delete_files && stub.is_transferred() {
      return false;
    }
    self.dispose_tree(key, delete_files)
  }

  /// 注册已存在的 BfTreeService 实例到管理器中 (1:1 对标 Garnet RegisterIndex)
  ///
  /// 持有条带互斥写锁，与 RestoreTree / UnregisterIndex / 检查点快照等路径串行化
  pub fn register_tree(&self, key: &[u8], tree: Arc<BfTreeService>) {
    let key_hash = Self::key_hash_of(key);
    let _stripe_lock = self.locks.write(key_hash);
    let key_id = Self::key_id_of(key);
    let hash_prefix = Self::hash_prefix_of(key);
    let pin = self.live_indexes.pin();
    if let Some(existing) = pin.get(&key_id) {
      *existing.tree.write() = Some(tree);
    } else {
      let entry = Arc::new(TreeEntry::new(Some(tree), key_hash, key_id, hash_prefix));
      pin.insert(key_id, entry);
    }
  }

  /// 刷盘事件触发快照与存根标记 (使用防重入快照锁)
  pub fn on_flush(&self, key: &[u8], stub: &mut RangeIndexStub) -> Result<()> {
    self.on_flush_internal(key, stub, None)
  }

  /// 带有逻辑地址的刷盘事件触发快照与存根标记
  pub fn on_flush_address(
    &self,
    key: &[u8],
    stub: &mut RangeIndexStub,
    logical_address: i64,
  ) -> Result<()> {
    self.on_flush_internal(key, stub, Some(logical_address))
  }

  fn on_flush_internal(
    &self,
    key: &[u8],
    stub: &mut RangeIndexStub,
    logical_address: Option<i64>,
  ) -> Result<()> {
    // 过期源存根 no-op (1:1 对标 C# SnapshotTreeForFlush)：所有权已转移至尾部新记录时，
    // 既不快照过期视图也不置位 IsFlushed，避免把陈旧数据误标为已刷盘
    if stub.is_transferred() {
      return Ok(());
    }

    let key_id = Self::key_id_of(key);
    let hash_prefix = Self::hash_prefix_of(key);
    let flush_path = match logical_address {
      Some(addr) => self.log_flush_path(&hash_prefix, addr),
      None => {
        let mut flush_name = String::with_capacity(hash_prefix.len() + FLUSH_FILE_SUFFIX.len());
        flush_name.push_str(&hash_prefix);
        flush_name.push_str(FLUSH_FILE_SUFFIX);
        self.ri_log_root.join(flush_name)
      }
    };

    // 热路径 (1:1 对标 C# SnapshotTreeForFlush 活跃分支)：在线树直接 CPR 快照，
    // CPR 与工作线程并发安全，全程不持条带锁 (Arc 保活使快照期间树实例不可被释放)
    if let Some((entry, tree)) = self.live_tree_of(key_id) {
      entry.snapshot_under_claim(&tree, &flush_path)?;
      stub.set_flushed(true);
      return Ok(());
    }

    // 冷路径 (1:1 对标 C# SnapshotForFlushCold)：持条带写锁与 RestoreTree / 注销路径
    // 串行化，锁内复查——树可能恰在取锁前被并发恢复激活
    let key_hash = Self::key_hash_of(key);
    let _stripe_lock = self.locks.write(key_hash);
    if let Some((entry, tree)) = self.live_tree_of(key_id) {
      entry.snapshot_under_claim(&tree, &flush_path)?;
      stub.set_flushed(true);
      return Ok(());
    }

    // 无在线树时 data.bftree 无并发写者，直接复制为刷盘快照；工作文件缺失属不变量
    // 破坏 (1:1 对标 C# LogOnFlushInvariantViolation)，保持未刷盘状态交由上层显式处理
    let data_path = self.data_file_path(&hash_prefix);
    if data_path.exists() {
      fs::copy(&data_path, &flush_path)?;
      stub.set_flushed(true);
    }

    Ok(())
  }

  /// 为全局检查点快照所有活跃与就绪的 BfTree (1:1 对标 Garnet SnapshotAllTreesForCheckpoint)
  pub fn snapshot_all_trees_for_checkpoint(&self, checkpoint_token: &str) -> Result<()> {
    self.snapshot_all_trees_to_dir(&self.cpr_dir, checkpoint_token)?;
    Ok(())
  }

  /// 为指定目标目录快照所有活跃与就绪的 BfTree
  ///
  /// 屏障时序 1:1 对标 C# 两阶段语义：未处于检查点中时就地设置屏障 (单步便捷路径)；
  /// 调用方已先行 [`set_checkpoint_barrier`](Self::set_checkpoint_barrier) 则保留原屏障
  /// 时序不重设——屏障设置与快照执行之间新注册的条目 (snapshot_pending == false)
  /// 不会被纳入本次检查点，与 C# 版本切换/FlushBegin 分离语义一致
  pub fn snapshot_all_trees_to_dir(
    &self,
    target_dir: &Path,
    checkpoint_token: &str,
  ) -> Result<usize> {
    if !self.is_checkpoint_in_progress() {
      self.set_checkpoint_barrier();
    }
    let _guard = CheckpointGuard(self);

    let entries = self.live_entries();
    let count = entries.len();

    let token_snapshot_dir = target_dir.join(checkpoint_token).join("rangeindex");
    let _ = fs::create_dir_all(&token_snapshot_dir);

    for entry in &entries {
      // 仅快照屏障设置时已存在的条目参与检查点快照；屏障设置后新注册的条目
      // snapshot_pending == false 直接跳过 (1:1 对标 C# SnapshotPending == 0 → continue)，
      // 避免把检查点 hlog 快照之外新建的幻影树写进检查点文件导致恢复时复活幽灵索引
      if !entry.snapshot_pending.load(Ordering::Acquire) {
        continue;
      }
      let _pending_guard = SnapshotPendingGuard(entry);
      let tree_opt = entry.tree.read().as_ref().cloned();
      let mut dest_file_name =
        String::with_capacity(entry.hash_prefix.len() + TREE_FILE_SUFFIX.len());
      dest_file_name.push_str(&entry.hash_prefix);
      dest_file_name.push_str(TREE_FILE_SUFFIX);
      let token_dest = token_snapshot_dir.join(&dest_file_name);

      if let Some(tree) = tree_opt {
        entry.snapshot_under_claim(&tree, &token_dest)?;
      } else {
        // 冷树 / 待激活树：data.bftree 已在磁盘上就绪，直接复制到检查点目录。
        // 工作文件存在但复制失败属致命错误，向上传播 (1:1 对标 C# File.Copy 异常传播)
        let data_path = self.data_file_path(&entry.hash_prefix);
        if data_path.exists() {
          fs::copy(&data_path, &token_dest)?;
        }
      }
    }

    Ok(count)
  }

  /// 解析刷盘快照文件名 `{hash_prefix}.{logical_address:016x}.flush.bftree`
  ///
  /// 严格校验：前缀 32 位十六进制、地址段恰好 16 位十六进制（1:1 对标 C# 固定长度解析，
  /// 并拒绝 `from_str_radix` 宽松接受的前导符号形式），确保不误选/误删外来文件
  #[inline]
  fn parse_flush_file_name(file_name: &str) -> Option<(&str, i64)> {
    let rest = file_name.strip_suffix(".flush.bftree")?;
    let (prefix, addr_str) = rest.rsplit_once('.')?;
    if prefix.len() != HASH_PREFIX_LEN || !prefix.as_bytes().iter().all(|b| b.is_ascii_hexdigit()) {
      return None;
    }
    if addr_str.len() != ADDR_HEX_LEN || !addr_str.as_bytes().iter().all(|b| b.is_ascii_hexdigit())
    {
      return None;
    }
    let addr = i64::from_str_radix(addr_str, 16).ok()?;
    Some((prefix, addr))
  }

  /// 日志截断清理：删除逻辑地址小于 new_begin_address 的历史刷盘快照文件 (1:1 对标 Garnet OnTruncateImpl)
  pub fn on_truncate(&self, new_begin_address: i64) -> Result<()> {
    if !self.ri_log_root.exists() {
      return Ok(());
    }

    for entry in fs::read_dir(&self.ri_log_root)? {
      let entry = entry?;
      let path = entry.path();
      if let Some(file_name) = path.file_name().and_then(|n| n.to_str())
        && let Some((_prefix, addr)) = Self::parse_flush_file_name(file_name)
        && addr < new_begin_address
      {
        let _ = fs::remove_file(&path);
      }
    }

    Ok(())
  }

  /// 收集指定检查点与 HybridLog 地址范围内需要进行主从复制的文件 (1:1 对标 Garnet EnumerateFilesForReplication)
  pub fn enumerate_files_for_replication(
    &self,
    checkpoint_token: &str,
    hlog_start_address: i64,
    hlog_end_address: i64,
  ) -> Result<Vec<RangeIndexFileEntry>> {
    let mut result = Vec::new();

    // 1. 扫描 ri_log_root 下的 *.flush.bftree 文件
    if self.ri_log_root.exists() {
      for entry in fs::read_dir(&self.ri_log_root)? {
        let entry = entry?;
        let file_name = entry.file_name();
        let Some(name) = file_name.to_str() else {
          continue;
        };
        if let Some((prefix, addr)) = Self::parse_flush_file_name(name)
          && addr >= hlog_start_address
          && addr < hlog_end_address
        {
          result.push(RangeIndexFileEntry {
            path: entry.path(),
            key_hash: prefix.to_string(),
            address: addr,
            is_flush_file: true,
          });
        }
      }
    }

    // 2. 扫描 cpr_dir/<token>/rangeindex/*.bftree 检查点快照
    let snapshot_dir = self.cpr_dir.join(checkpoint_token).join("rangeindex");
    if snapshot_dir.exists() {
      for entry in fs::read_dir(&snapshot_dir)? {
        let entry = entry?;
        let file_name = entry.file_name();
        let Some(name) = file_name.to_str() else {
          continue;
        };
        if let Some(stem) = name.strip_suffix(".bftree")
          // 快照文件名固定为 32 位十六进制前缀，跳过外来文件 (对标 C# 长度过滤并加强校验)
          && stem.len() == HASH_PREFIX_LEN
          && stem.as_bytes().iter().all(|b| b.is_ascii_hexdigit())
        {
          result.push(RangeIndexFileEntry {
            path: entry.path(),
            key_hash: stem.to_string(),
            address: 0,
            is_flush_file: false,
          });
        }
      }
    }

    Ok(result)
  }

  /// 便捷别名：获取需要进行主从复制的文件路径列表 (1:1 对标 get_replication_file_names)
  #[inline]
  pub fn get_replication_file_names(
    &self,
    checkpoint_token: &str,
    hlog_start_address: i64,
    hlog_end_address: i64,
  ) -> Result<Vec<PathBuf>> {
    let entries = self.enumerate_files_for_replication(
      checkpoint_token,
      hlog_start_address,
      hlog_end_address,
    )?;
    Ok(entries.into_iter().map(|e| e.path).collect())
  }

  /// 从指定检查点全量恢复所有 BfTree 索引 (1:1 对标 Garnet RecoverAllTreesFromCheckpoint)
  pub fn recover_all_trees_from_checkpoint(&self, checkpoint_token: &str) -> Result<()> {
    self.recover_all_trees_from_dir(&self.cpr_dir, checkpoint_token)?;
    Ok(())
  }

  /// 从指定目标目录全量恢复所有 BfTree 索引至 ri_log_root 并恢复注册 (支持多候选路径容错)
  pub fn recover_all_trees_from_dir(
    &self,
    target_dir: &Path,
    checkpoint_token: &str,
  ) -> Result<usize> {
    let candidate_dirs = [
      target_dir.join(checkpoint_token).join("rangeindex"),
      target_dir.join("rangeindex"),
      self.cpr_dir.join(checkpoint_token).join("rangeindex"),
      self.cpr_dir.join("rangeindex"),
    ];

    let mut recovered_count = 0;
    for snapshot_dir in &candidate_dirs {
      if !snapshot_dir.exists() {
        continue;
      }
      if let Ok(entries) = fs::read_dir(snapshot_dir) {
        for entry in entries.flatten() {
          let path = entry.path();
          if path.extension().is_some_and(|ext| ext == "bftree")
            && let Some(stem) = path.file_stem().and_then(|s| s.to_str())
            // 快照文件名固定为 32 位十六进制前缀，跳过外来文件避免 key_id 误注册
            && stem.len() == HASH_PREFIX_LEN
            && stem.as_bytes().iter().all(|b| b.is_ascii_hexdigit())
          {
            let target_data_path = self.data_file_path(stem);
            if !target_data_path.exists() || target_data_path != path {
              let _ = fs::copy(&path, &target_data_path);
            }

            // stem 已严格校验为 32 位十六进制，必然可解析为 u128
            let Ok(key_id) = u128::from_str_radix(stem, 16) else {
              continue;
            };
            if let Ok(recovered_tree) = BfTreeService::recover_from_cpr_snapshot(
              &target_data_path,
              true,
              StorageBackendType::Disk,
            ) {
              let key_hash = Self::key_hash_of(stem.as_bytes());
              let tree_entry = Arc::new(TreeEntry::new(
                Some(Arc::new(recovered_tree)),
                key_hash,
                key_id,
                stem.to_string(),
              ));
              self.live_indexes.pin().insert(key_id, tree_entry);
              recovered_count += 1;
            }
          }
        }
      }
      if recovered_count > 0 {
        break;
      }
    }

    Ok(recovered_count)
  }
}

impl Drop for RangeIndexManager {
  fn drop(&mut self) {
    self.dispose();
  }
}