wbftree 0.1.0

//! BfTree 高层服务包装器 (1:1 对标 Garnet BfTreeService.cs)
//!
//! 封装 Microsoft Research 的 bf-tree 核心实例，提供零堆分配切片 API、流式扫描回调与快照恢复。

use std::{
  fs,
  panic::{self, AssertUnwindSafe},
  path::{Path, PathBuf},
  result::Result as StdResult,
  sync::{
    Arc,
    atomic::{AtomicBool, AtomicU8, AtomicUsize, Ordering},
  },
  thread::yield_now,
};

use bf_tree::{BfTree, ConfigError, LeafInsertResult, LeafReadResult, ScanIter, ScanIterError};
use parking_lot::RwLock;

use crate::{
  error::{Error, Result},
  types::{
    BfTreeConfig, BfTreeDeleteResult, BfTreeInsertResult, BfTreeReadResult, ScanRecord,
    ScanReturnField, StorageBackendType,
  },
};

/// 栈上单值读取缓冲区大小 (值 ≤ 4096 字节走零堆分配快路径)
const STACK_READ_BUF_SIZE: usize = 4096;

/// 栈上扫描缓冲区大小 (键+值 ≤ cb_max_record_size ≤ 8192 时零堆分配快路径)
const STACK_SCAN_BUF_SIZE: usize = 8192;

/// 快照等待写者清零的自旋间隔 (纳秒级 yield，避免 burn CPU)
const SNAPSHOT_DRAIN_SPINS: u32 = 64;

/// 快照文件缺失错误
fn snapshot_missing(path: &Path) -> Error {
  let mut msg = String::from(SNAPSHOT_MISSING_PREFIX);
  msg.push_str(&path.display().to_string());
  Error::Recovery(msg)
}

/// 便捷构造的默认调优参数 (1:1 对标 Garnet 默认树参数)
const PRESET_LEAF_PAGE_SIZE: usize = 16384;
const PRESET_MAX_RECORD_SIZE: usize = 4096;
const PRESET_MAX_KEY_LEN: usize = 512;
const PRESET_MIN_RECORD_SIZE: usize = 4;

/// 快照文件缺失错误消息前缀
const SNAPSHOT_MISSING_PREFIX: &str = "快照文件不存在: ";

/// 最大记录大小下限 (保证读/扫描缓冲区 ≥ 单值上限，同时作为栈缓冲路径的切换阈值)
const MIN_MAX_RECORD_SIZE: usize = STACK_READ_BUF_SIZE;

/// 将 bf_tree::ConfigError 映射为可读字符串 (替代 Debug 格式化)
#[inline]
fn config_error_to_string(e: ConfigError) -> String {
  match e {
    ConfigError::MinimumRecordSize(s) => {
      let mut msg = String::from("MinimumRecordSize: ");
      msg.push_str(&s);
      msg
    }
    ConfigError::MaximumRecordSize(s) => {
      let mut msg = String::from("MaximumRecordSize: ");
      msg.push_str(&s);
      msg
    }
    ConfigError::LeafPageSize(s) => {
      let mut msg = String::from("LeafPageSize: ");
      msg.push_str(&s);
      msg
    }
    ConfigError::MaxKeyLen(s) => {
      let mut msg = String::from("MaxKeyLen: ");
      msg.push_str(&s);
      msg
    }
    ConfigError::CircularBufferSize(s) => {
      let mut msg = String::from("CircularBufferSize: ");
      msg.push_str(&s);
      msg
    }
    ConfigError::SnapshotFileInvalid(s) => {
      let mut msg = String::from("SnapshotFileInvalid: ");
      msg.push_str(&s);
      msg
    }
    ConfigError::SnapshotDisabled => String::from("SnapshotDisabled"),
  }
}

/// 将 bf_tree::ScanIterError 映射为可读字符串 (替代 Debug 格式化)
#[inline]
fn scan_iter_error_to_string(e: ScanIterError) -> &'static str {
  match e {
    ScanIterError::CacheOnlyMode => "CacheOnlyMode",
    ScanIterError::InvalidStartKey => "InvalidStartKey",
    ScanIterError::InvalidEndKey => "InvalidEndKey",
    ScanIterError::InvalidCount => "InvalidCount",
    ScanIterError::InvalidKeyRange => "InvalidKeyRange",
  }
}

/// 高层 BfTree 服务实例 (1:1 对标 Garnet BfTreeService)
///
/// 后端/路径/记录上限支持原地恢复换树 (`recover_in_place`)：
/// 热路径字段走原子量，文件路径走冷路径读写锁。
pub struct BfTreeService {
  tree: RwLock<Option<Arc<BfTree>>>,
  storage_backend: AtomicU8,
  file_path: RwLock<Option<String>>,
  max_record_size: AtomicUsize,
  disposed: AtomicBool,
  /// 活动写者计数 (insert/delete 微守卫持有时 > 0)
  writers: AtomicUsize,
  /// 活动屏障计数 (对标 Garnet checkpoint barrier)：> 0 时新写者短暂自旋等待。
  /// 计数而非布尔位，嵌套叠加（外层屏障内嵌 cpr_snapshot / recover_in_place 的
  /// 内部屏障）时写者阻塞至最外层守卫丢弃，杜绝内层先行释放溶解外层窗口。
  barriers: AtomicUsize,
}

/// BfTree 写者 RAII 守卫：持有时写入计数 > 0，阻止快照与写入撕裂
struct WriteGuard<'a> {
  writers: &'a AtomicUsize,
}

impl<'a> WriteGuard<'a> {
  /// 获取写者守卫：快照进行中则自旋等待屏障放行后再登记
  #[inline]
  fn acquire(service: &'a BfTreeService) -> Self {
    let mut spins = 0u32;
    loop {
      if service.barriers.load(Ordering::Acquire) == 0 {
        // SeqCst 登记 + SeqCst 复读与屏障侧「store(true) → load(writers)」构成
        // Dekker/store-buffering 配对：Release/Acquire 允许两侧同时读到旧值
        // （屏障漏算在途写者且写者漏看屏障），SeqCst 全序保证任一侧必然观察到对方。
        service.writers.fetch_add(1, Ordering::SeqCst);
        // 双检：登记瞬间快照恰好开始则回退重试，保证与快照互斥
        if service.barriers.load(Ordering::SeqCst) == 0 {
          return Self {
            writers: &service.writers,
          };
        }
        service.writers.fetch_sub(1, Ordering::Release);
      }
      spins = spins.wrapping_add(1);
      if spins.is_multiple_of(SNAPSHOT_DRAIN_SPINS) {
        yield_now();
      }
    }
  }
}

impl Drop for WriteGuard<'_> {
  #[inline]
  fn drop(&mut self) {
    self.writers.fetch_sub(1, Ordering::Release);
  }
}

/// BfTree 写入屏障 RAII 守卫 (对标 Garnet SetCheckpointBarrier)
///
/// 持有期间屏障计数 > 0，全部 insert/delete 自旋等待；丢弃时递减计数，
/// 写者阻塞至最外层守卫丢弃。持有着必须保证屏障窗口内不做任何跨线程
/// 事件等待（await / driver I/O 依赖）——同步自旋的写者不会让出 executor，
/// 屏障持有着若依赖同线程事件将形成死锁。
pub struct WriteBarrierGuard<'a> {
  service: &'a BfTreeService,
}

impl Drop for WriteBarrierGuard<'_> {
  #[inline]
  fn drop(&mut self) {
    self.service.barriers.fetch_sub(1, Ordering::Release);
  }
}

impl BfTreeService {
  /// 内部构建辅助函数，直接传入已知后端和路径
  pub(crate) fn new_with_backend(
    config: impl Into<bf_tree::Config>,
    storage_backend: StorageBackendType,
    file_path: Option<String>,
  ) -> Result<Self> {
    if storage_backend == StorageBackendType::Disk && file_path.is_none() {
      return Err(Error::InvalidArgument(
        "磁盘后端必须指定数据文件路径 (file_path)".into(),
      ));
    }

    let inner_cfg: bf_tree::Config = config.into();
    let max_record_size = inner_cfg.get_cb_max_record_size().max(MIN_MAX_RECORD_SIZE);

    let tree = match BfTree::with_config(inner_cfg, None) {
      Ok(t) => Arc::new(t),
      Err(e) => return Err(Error::InvalidConfig(config_error_to_string(e))),
    };

    Ok(Self {
      tree: RwLock::new(Some(tree)),
      storage_backend: AtomicU8::new(storage_backend as u8),
      file_path: RwLock::new(file_path),
      max_record_size: AtomicUsize::new(max_record_size),
      disposed: AtomicBool::new(false),
      writers: AtomicUsize::new(0),
      barriers: AtomicUsize::new(0),
    })
  }

  /// 根据配置创建全新的 BfTreeService (零 Debug 字符串解析)
  pub fn new(config: BfTreeConfig) -> Result<Self> {
    let storage_backend = config.storage_backend;
    let file_path = config.file_path;
    Self::new_with_backend(config.inner, storage_backend, file_path)
  }

  /// 便捷构造共享的默认调优配置 (1:1 对标 Garnet 默认树参数)
  fn preset_config(cb_min_record_size: usize) -> BfTreeConfig {
    let mut config = BfTreeConfig::default();
    config
      .use_snapshot(true)
      .leaf_page_size(PRESET_LEAF_PAGE_SIZE)
      .cb_max_record_size(PRESET_MAX_RECORD_SIZE)
      .cb_max_key_len(PRESET_MAX_KEY_LEN)
      .cb_min_record_size(if cb_min_record_size > 0 {
        cb_min_record_size
      } else {
        PRESET_MIN_RECORD_SIZE
      });
    config
  }

  /// 便捷创建磁盘文件后端树实例 (1:1 对标 Garnet new BfTreeService(filePath: path, ...))
  pub fn open_disk(path: impl AsRef<Path>, cb_min_record_size: usize) -> Result<Self> {
    let p = path.as_ref();
    if p.as_os_str().is_empty() {
      return Err(Error::InvalidArgument(
        "磁盘后端必须指定有效的数据文件路径".into(),
      ));
    }
    if let Some(parent) = p.parent()
      && !parent.as_os_str().is_empty()
    {
      fs::create_dir_all(parent)?;
    }
    let mut config = Self::preset_config(cb_min_record_size);
    config.file_path(p);
    Self::new_with_backend(
      config,
      StorageBackendType::Disk,
      Some(p.to_string_lossy().into_owned()),
    )
  }

  /// 便捷创建纯内存后端树实例 (1:1 对标 Garnet new BfTreeService(storageBackend: Memory, ...))
  pub fn open_memory(cb_min_record_size: usize) -> Result<Self> {
    let mut config = Self::preset_config(cb_min_record_size);
    config.cache_only(true);
    Self::new_with_backend(config, StorageBackendType::Memory, None)
  }

  /// 获取底层 BfTree 的 Arc 实例
  #[inline]
  fn tree_arc(&self) -> Result<Arc<BfTree>> {
    self.check_disposed()?;
    self.tree.read().as_ref().cloned().ok_or(Error::Disposed)
  }

  /// 在读锁保护下安全借用底层 BfTree (零 Arc 克隆，零原子增减开销)
  #[inline]
  fn with_tree<R>(&self, f: impl FnOnce(&BfTree) -> R) -> Result<R> {
    self.check_disposed()?;
    let guard = self.tree.read();
    match guard.as_ref() {
      Some(tree) => Ok(f(tree)),
      None => Err(Error::Disposed),
    }
  }

  /// 获取裸指针标识 (用于 RangeIndexStub.tree_handle)
  #[inline]
  pub fn native_ptr(&self) -> u64 {
    if let Some(tree) = self.tree.read().as_ref() {
      Arc::as_ptr(tree) as usize as u64
    } else {
      0
    }
  }

  /// 获取数据文件路径
  #[inline]
  pub fn file_path(&self) -> Option<String> {
    self.file_path.read().clone()
  }

  /// 获取存储后端类型
  #[inline]
  pub fn storage_backend(&self) -> StorageBackendType {
    StorageBackendType::from_u8(self.storage_backend.load(Ordering::Acquire))
  }

  /// 获取最大记录大小 (读/扫描缓冲区 sizing 依据)
  #[inline]
  fn max_record_size(&self) -> usize {
    self.max_record_size.load(Ordering::Relaxed)
  }

  /// 检查是否已释放 (1:1 对标 is_disposed)
  #[inline]
  pub fn is_disposed(&self) -> bool {
    self.disposed.load(Ordering::Acquire)
  }

  /// 检查是否已释放
  #[inline]
  fn check_disposed(&self) -> Result<()> {
    if self.is_disposed() {
      Err(Error::Disposed)
    } else {
      Ok(())
    }
  }

  /// 插入键值对 (零 Arc 克隆；空值快速拒绝，与底层 min_record_size 校验语义一致)
  ///
  /// 顶部登记写者微守卫：快照期间短暂自旋等待屏障放行，保证快照不含撕裂写 (对标 Garnet checkpoint barrier)
  #[inline]
  pub fn insert(&self, key: &[u8], value: &[u8]) -> BfTreeInsertResult {
    if value.is_empty() {
      return BfTreeInsertResult::InvalidKV;
    }
    let _guard = WriteGuard::acquire(self);
    self
      .with_tree(|tree| match tree.insert(key, value) {
        LeafInsertResult::Success => BfTreeInsertResult::Success,
        LeafInsertResult::InvalidKV(_) => BfTreeInsertResult::InvalidKV,
      })
      .unwrap_or(BfTreeInsertResult::InvalidArguments)
  }

  /// 读取键对应的值 (≤4096 字节值走栈缓冲区零堆分配；更大值按 cb_max_record_size 一次性分配堆缓冲区)
  ///
  /// 底层 bf-tree 要求读取缓冲区不小于值长度（否则越界 panic），此处缓冲区恒 ≥ cb_max_record_size，绝无越界。
  pub fn read(&self, key: &[u8]) -> (BfTreeReadResult, Option<Vec<u8>>) {
    let max_record_size = self.max_record_size.load(Ordering::Relaxed);
    if max_record_size <= STACK_READ_BUF_SIZE {
      let mut stack_buf = [0u8; STACK_READ_BUF_SIZE];
      let (res, len) = self.read_direct(key, &mut stack_buf);
      (
        res,
        (res == BfTreeReadResult::Found).then(|| stack_buf[..len].to_vec()),
      )
    } else {
      let mut heap_buf = vec![0u8; max_record_size];
      let (res, len) = self.read_direct(key, &mut heap_buf);
      if res == BfTreeReadResult::Found {
        heap_buf.truncate(len);
        heap_buf.shrink_to_fit();
        (res, Some(heap_buf))
      } else {
        (res, None)
      }
    }
  }

  /// 直读：要求 out_buf 容量 ≥ cb_max_record_size（恒能容纳任意合法值，零额外开销）
  #[inline]
  fn read_direct(&self, key: &[u8], out_buf: &mut [u8]) -> (BfTreeReadResult, usize) {
    self
      .with_tree(|tree| match tree.read(key, out_buf) {
        LeafReadResult::Found(n) => (BfTreeReadResult::Found, n as usize),
        LeafReadResult::NotFound => (BfTreeReadResult::NotFound, 0),
        LeafReadResult::Deleted => (BfTreeReadResult::Deleted, 0),
        LeafReadResult::InvalidKey => (BfTreeReadResult::InvalidKey, 0),
      })
      .unwrap_or((BfTreeReadResult::InvalidArguments, 0))
  }

  /// 经临时缓冲读取后按需拷贝至 out_buf (值超出 out_buf 容量时返回 InvalidArguments)
  #[inline]
  fn read_via_scratch(
    &self,
    key: &[u8],
    out_buf: &mut [u8],
    scratch: &mut [u8],
  ) -> (BfTreeReadResult, usize) {
    let (res, len) = self.read_direct(key, scratch);
    match res {
      BfTreeReadResult::Found if out_buf.len() >= len => {
        out_buf[..len].copy_from_slice(&scratch[..len]);
        (BfTreeReadResult::Found, len)
      }
      BfTreeReadResult::Found => (BfTreeReadResult::InvalidArguments, 0),
      r => (r, 0),
    }
  }

  /// 读取键对应的值到用户提供的输出切片中（零堆分配）
  ///
  /// 当 out_buf 容量 ≥ cb_max_record_size 时走直读快路径；否则改用内部安全缓冲读取后按需拷贝，
  /// 值超出 out_buf 容量时返回 InvalidArguments（底层 bf-tree 缓冲区过小会直接越界 panic，此处彻底拦截）。
  pub fn read_into(&self, key: &[u8], out_buf: &mut [u8]) -> (BfTreeReadResult, usize) {
    let max_record_size = self.max_record_size.load(Ordering::Relaxed);
    if out_buf.len() >= max_record_size {
      return self.read_direct(key, out_buf);
    }
    if max_record_size <= STACK_READ_BUF_SIZE {
      let mut stack_buf = [0u8; STACK_READ_BUF_SIZE];
      self.read_via_scratch(key, out_buf, &mut stack_buf)
    } else {
      let mut heap_buf = vec![0u8; max_record_size];
      self.read_via_scratch(key, out_buf, &mut heap_buf)
    }
  }

  /// 删除指定键 (打入墓碑标记，与 insert 同受快照屏障保护)
  #[inline]
  pub fn delete(&self, key: &[u8]) -> BfTreeDeleteResult {
    let _guard = WriteGuard::acquire(self);
    self
      .with_tree(|tree| {
        tree.delete(key);
        BfTreeDeleteResult::Success
      })
      .unwrap_or(BfTreeDeleteResult::InvalidArguments)
  }

  /// 基于数量的流式范围扫描 (内部栈缓冲区零分配回调)
  ///
  /// count == 0 直接返回 0 (1:1 对标 Garnet 原生层允许 count=0 的行为)
  pub fn scan_with_count_callback<F>(
    &self,
    start_key: &[u8],
    count: usize,
    return_field: ScanReturnField,
    on_record: F,
  ) -> Result<usize>
  where
    F: FnMut(&[u8], &[u8]) -> bool,
  {
    if count == 0 {
      return Ok(0);
    }
    self.scan_callback(
      |tree| tree.scan_with_count(start_key, count, return_field),
      return_field,
      on_record,
    )
  }

  /// 基于数量的范围扫描并返回记录列表
  pub fn scan_with_count(
    &self,
    start_key: &[u8],
    count: usize,
    return_field: ScanReturnField,
  ) -> Result<Vec<ScanRecord>> {
    let mut records = Vec::with_capacity(count.min(1024));
    self.scan_with_count_callback(start_key, count, return_field, |k, v| {
      records.push(ScanRecord {
        key: k.to_vec(),
        value: v.to_vec(),
      });
      true
    })?;
    Ok(records)
  }

  /// 闭区间流式范围扫描 (内部栈缓冲区零分配回调)
  ///
  /// start_key > end_key 时视为空区间直接返回 0 (1:1 对标 Garnet 原生层行为)
  pub fn scan_with_end_key_callback<F>(
    &self,
    start_key: &[u8],
    end_key: &[u8],
    return_field: ScanReturnField,
    on_record: F,
  ) -> Result<usize>
  where
    F: FnMut(&[u8], &[u8]) -> bool,
  {
    if start_key > end_key {
      return Ok(0);
    }
    self.scan_callback(
      |tree| tree.scan_with_end_key(start_key, end_key, return_field),
      return_field,
      on_record,
    )
  }

  /// 闭区间范围扫描并返回记录列表
  pub fn scan_with_end_key(
    &self,
    start_key: &[u8],
    end_key: &[u8],
    return_field: ScanReturnField,
  ) -> Result<Vec<ScanRecord>> {
    let mut records = Vec::with_capacity(32);
    self.scan_with_end_key_callback(start_key, end_key, return_field, |k, v| {
      records.push(ScanRecord {
        key: k.to_vec(),
        value: v.to_vec(),
      });
      true
    })?;
    Ok(records)
  }

  /// 全表顺序扫描
  pub fn scan_all(&self, return_field: ScanReturnField) -> Result<Vec<ScanRecord>> {
    self.scan_with_count(&[0], usize::MAX, return_field)
  }

  /// 扫描统一驱动：经底层校验构造迭代器 (非法键/区间返回 Err，杜绝底层未定义行为)，逐条填充回调。
  ///
  /// 迭代前仅克隆一次底层 Arc 并随即释放包装读锁：整个用户回调期间不持有任何服务级锁，
  /// 回调因此可安全重入本服务的点读 (read/read_into) 乃至 dispose 而无包装层自死锁风险
  /// (1:1 对标 C# 扫描期间无托管锁的语义；parking_lot 写优先，持读锁跨回调时一旦有
  /// dispose 排队，回调内任何重入读取都将永久阻塞)。Arc 同时保证迭代期间底层引擎实例
  /// 存活，dispose 与扫描并发时扫描仍可在存活引擎上安全完成 (对标 C# LightEpoch 延迟释放)。
  ///
  /// 注意：回调不得对同一棵树重入写入 (insert/delete/scan)——底层引擎扫描持有叶子共享
  /// 闩锁，同线程重入写同叶子会在引擎闩锁层自死锁 (与 C# 原生层约束一致，非包装层问题)。
  fn scan_callback<F>(
    &self,
    make_iter: impl FnOnce(&BfTree) -> StdResult<ScanIter<'_, '_>, ScanIterError>,
    return_field: ScanReturnField,
    mut on_record: F,
  ) -> Result<usize>
  where
    F: FnMut(&[u8], &[u8]) -> bool,
  {
    // 单次原子引用计数开销换取回调重入安全，绝不持锁跨用户回调
    let tree = self.tree_arc()?;
    let mut iter = make_iter(&tree)
      .map_err(|e| Error::InvalidArgument(scan_iter_error_to_string(e).to_string()))?;

    // 缓冲区恒 ≥ 最大记录长度 (键+值 ≤ cb_max_record_size)，底层填充绝不会越界
    let mut stack_buf = [0u8; STACK_SCAN_BUF_SIZE];
    let mut heap_buf;
    let max_record_size = self.max_record_size.load(Ordering::Relaxed);
    let buf: &mut [u8] = if max_record_size <= STACK_SCAN_BUF_SIZE {
      &mut stack_buf
    } else {
      heap_buf = vec![0u8; max_record_size];
      &mut heap_buf
    };

    let mut scanned = 0;
    while let Some((k_len, v_len)) = iter.next(buf) {
      let k = if return_field != ScanReturnField::Value {
        &buf[..k_len]
      } else {
        &[]
      };
      let v = if return_field != ScanReturnField::Key {
        &buf[k_len..k_len + v_len]
      } else {
        &[]
      };
      scanned += 1;
      if !on_record(k, v) {
        break;
      }
    }

    Ok(scanned)
  }

  /// 自旋排空全部在途写者 (对标 Garnet SnapshotUnderClaim 的 claim 等待)
  ///
  /// 必须在 [`write_barrier`](Self::write_barrier) 置位后调用：等待已越过双检的
  /// 在途写者全部退出，此后 writers == 0 即树对写静稳，可安全快照/换树。
  fn drain_writers(&self) {
    let mut spins = 0u32;
    while self.writers.load(Ordering::SeqCst) != 0 {
      spins = spins.wrapping_add(1);
      if spins.is_multiple_of(SNAPSHOT_DRAIN_SPINS) {
        yield_now();
      }
    }
  }

  /// 开启写入屏障并返回 RAII 守卫 (对标 Garnet SetCheckpointBarrier)
  ///
  /// 计数式屏障：嵌套叠加时写者阻塞至最外层守卫丢弃；守卫丢弃递减计数。
  /// ⚠️ 持有窗口内严禁跨 await / 依赖同线程 I/O 事件（见类型文档），
  /// 因此 checkpoint 流程不持有跨 await 的外层屏障，快照互斥仅由
  /// [`cpr_snapshot`](Self::cpr_snapshot) / [`recover_in_place`](Self::recover_in_place)
  /// 内部的同步短屏障承担。
  pub fn write_barrier(&self) -> WriteBarrierGuard<'_> {
    // SeqCst 置位：与写者侧 fetch_add(SeqCst) + 双检 load(SeqCst) 构成全序配对
    self.barriers.fetch_add(1, Ordering::SeqCst);
    WriteBarrierGuard { service: self }
  }

  /// 触发 CPR 快照
  ///
  /// 底层 bf-tree 在未启用 use_snapshot 等异常场景下直接 panic 而非返回错误，
  /// 此处 catch_unwind 拦截转换为 Err (1:1 对标 Garnet 原生互操作层 bftree_cpr_snapshot 的处理方式)。
  /// 注意：release 构建全局 `panic = "abort"`，panic 路径实际以进程终止收场
  /// （检查点元数据未发布，重启一致性不受影响）；catch_unwind 仅在
  /// unwind 构建（dev/test）下生效，且 panic 点位于任何写入之前，无部分写入副作用。
  ///
  /// 快照前自置同步屏障并排空全部在途写者 (对标 Garnet SnapshotUnderClaim)，
  /// 保证快照不与 insert/delete 撕裂；屏障窗口纯同步无 await，无死锁风险。
  pub fn cpr_snapshot(&self, snapshot_path: impl AsRef<Path>) -> Result<()> {
    let tree = self.tree_arc()?;
    let p = snapshot_path.as_ref();
    if let Some(parent) = p.parent()
      && !parent.as_os_str().is_empty()
    {
      fs::create_dir_all(parent)?;
    }
    let _guard = self.write_barrier();
    self.drain_writers();
    let res = panic::catch_unwind(AssertUnwindSafe(|| tree.cpr_snapshot(p)))
      .map_err(|_| Error::Snapshot("底层引擎异常 (快照未启用或内部状态异常)".into()));
    drop(_guard);
    res
  }

  /// 从 CPR 快照原地换入恢复树 (1:1 对标 Garnet RestoreTree 的句柄重锚定)
  ///
  /// 恢复流程（保证任何时刻都不破坏仍存活的旧树，且任一步失败状态自洽）：
  /// 1. 快照拷贝至 `work_path.recovering` 临时文件（绝不覆盖旧树正在使用的 `work_path`）；
  /// 2. 从临时文件恢复出新树（其活动基文件即该临时 inode）；
  /// 3. 临时 inode 原子 rename 至 `work_path`——旧树 fd 指向原 inode 不受换名影响；
  ///    此后工作路径命名即新快照态，下次启动可凭魔数直接恢复，且 purge 回收
  ///    token 目录绝不伤及活动树；
  /// 4. 屏障内排空在途写者后写锁下换树——换树并发窗口内不存在「写入旧树成功
  ///    应答却被换树丢弃」的丢失写（屏障前已应答的写入随恢复回滚属既定语义）；
  ///    旧树摘除后移出写锁析构（其基文件为已换名的原 inode，经 fd 访问全程有效）。
  ///
  /// 所有持有 `Arc<BfTreeService>` 的使用方（如 ACL 存储）无需重绑即透明使用恢复后的树。
  pub fn recover_in_place(&self, snapshot_path: &Path, work_path: &Path) -> Result<()> {
    if !snapshot_path.exists() {
      return Err(snapshot_missing(snapshot_path));
    }
    // 预建工作路径父目录：保证同目录 rename 不因目录缺失而失败
    if let Some(parent) = work_path.parent()
      && !parent.as_os_str().is_empty()
    {
      fs::create_dir_all(parent)?;
    }
    let mut tmp_os = work_path.as_os_str().to_os_string();
    tmp_os.push(".recovering");
    let tmp_path = PathBuf::from(tmp_os);

    // 1. 预置临时文件（覆盖上一轮可能的残留）；恢复失败时清理残留后原样上抛
    fs::copy(snapshot_path, &tmp_path)?;
    let recovered = match Self::recover_from_cpr_snapshot(&tmp_path, true, StorageBackendType::Disk)
    {
      Ok(tree) => tree,
      Err(e) => {
        let _ = fs::remove_file(&tmp_path);
        return Err(e);
      }
    };
    if self.disposed.load(Ordering::Acquire) {
      let _ = fs::remove_file(&tmp_path);
      return Err(Error::Disposed);
    }

    // 2. 先换名再换树：rename 失败时树未换、盘上态未动，清理残留即可原样上抛；
    //    成功后 work_path 命名即新快照 inode（旧树经 fd 继续访问原 inode，不受影响）
    if let Err(e) = fs::rename(&tmp_path, work_path) {
      let _ = fs::remove_file(&tmp_path);
      return Err(e.into());
    }

    // 3. 屏障内排空在途写者再写锁换树：写锁内仅做指针替换，
    //    旧树析构（遍历基页刷盘，经自身 fd 访问已换名的原 inode）移至锁外
    let old_tree = {
      let _barrier = self.write_barrier();
      self.drain_writers();
      let mut guard = self.tree.write();
      // 写锁内复查：并发 dispose 抢先释放时放弃换树，保持已释放语义
      //（recovered 树随栈变量析构关闭，旧树槽位保持 None）
      if self.disposed.load(Ordering::Acquire) {
        return Err(Error::Disposed);
      }
      let old_tree = guard.take();
      *guard = recovered.tree.write().take();
      old_tree
    };
    drop(old_tree);
    self
      .storage_backend
      .store(recovered.storage_backend() as u8, Ordering::Release);
    *self.file_path.write() = Some(work_path.to_string_lossy().into_owned());
    self
      .max_record_size
      .store(recovered.max_record_size(), Ordering::Release);
    Ok(())
  }

  /// 从 CPR 快照文件恢复创建全新的 BfTreeService
  ///
  /// 损坏的快照文件 (魔数不匹配等) 会使底层断言 panic，此处 catch_unwind 拦截转换为 Err
  /// (1:1 对标 Garnet 原生互操作层 bftree_new_from_cpr_snapshot 的处理方式)。
  /// 注意：release 构建全局 `panic = "abort"` 时 panic 直接终止进程，拦截仅在
  /// unwind 构建 (dev/test) 下生效。
  pub fn recover_from_cpr_snapshot(
    recovery_path: impl AsRef<Path>,
    enable_snapshots: bool,
    storage_backend: impl Into<StorageBackendType>,
  ) -> Result<Self> {
    let p = recovery_path.as_ref();
    if !p.exists() {
      return Err(snapshot_missing(p));
    }
    let backend = storage_backend.into();
    let use_snapshot = enable_snapshots;
    match panic::catch_unwind(AssertUnwindSafe(|| {
      BfTree::new_from_cpr_snapshot(p, use_snapshot, None, None, None)
    })) {
      Ok(Ok(tree)) => {
        let max_record_size = tree
          .config()
          .get_cb_max_record_size()
          .max(MIN_MAX_RECORD_SIZE);
        Ok(Self {
          tree: RwLock::new(Some(Arc::new(tree))),
          storage_backend: AtomicU8::new(backend as u8),
          file_path: RwLock::new(Some(p.to_string_lossy().into_owned())),
          max_record_size: AtomicUsize::new(max_record_size),
          disposed: AtomicBool::new(false),
          writers: AtomicUsize::new(0),
          barriers: AtomicUsize::new(0),
        })
      }
      Ok(Err(e)) => Err(Error::Recovery(config_error_to_string(e))),
      Err(_) => {
        let mut msg = String::from("快照文件损坏或格式非法: ");
        msg.push_str(&p.display().to_string());
        Err(Error::Recovery(msg))
      }
    }
  }

  /// 释放实例并回收资源
  pub fn dispose(&self) {
    if !self.disposed.swap(true, Ordering::SeqCst) {
      self.tree.write().take();
    }
  }
}

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