pi_db 0.19.13

//! 全局 Key 版本、快照租约、TTL 回收及兼容期 LogWrite publication 门。
//!
//! 本模块只保存版本证据，不保存表数据，也不执行 WAL 或数据文件 I/O。公开载荷用于
//! `query_with_version -> prepare_with_version -> commit_with_version` 协议；其余类型均为
//! `KVDBManager` 正常装配表时使用的 crate-private 实现。

use std::borrow::Borrow;
use std::collections::BTreeMap;
use std::io::{Error, ErrorKind, Result as IOResult};
use std::mem;
use std::sync::{Arc, Weak,
                atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}};
use std::time::{Duration, Instant};

use async_channel::{Receiver, Sender, bounded};
use async_lock::RwLock;
use crossbeam_channel::{Receiver as SyncReceiver, Sender as SyncSender, TryRecvError,
                        unbounded as sync_unbounded};
use dashmap::{DashMap, mapref::entry::Entry};
use futures::{FutureExt, future::{Either, select}};
use parking_lot::Mutex;

use pi_async_rt::rt::{AsyncRuntime, multi_thread::MultiTaskRuntime};
use pi_atom::Atom;
use pi_guid::Guid;
use pi_hash::XHashMap;

use crate::{Binary, KVActionLog};

const NO_DEADLINE: u64 = u64::MAX;
const MAX_TICK: u64 = u64::MAX - 1;
const TTL_SCAN_BATCH_SIZE: usize = 256;

/// 一个 Key 的公开版本。
///
/// Upsert/Delete 都携带产生该版本的 transaction UID；首次观察使用事务管理器同一 Guid
/// 生成器独立分配 UID，但不表示发生过真实事务提交。首次观察到 Key 不存在时也使用
/// [`Version::Delete`]，其 Guid 只标识这次已发布的不存在状态，不能解释为 delete 事务回执。
///
/// `Guid` 是 owned `u128` 值；clone、相等和 Hash 均为 O(1)，不会借用或访问版本缓存，也不会
/// 刷新 TTL。后续事务发布、TTL 淘汰或数据库关闭不会原地改变已经返回的 `Version`。该类型
/// 本身不执行查询、写入、WAL、锁或 I/O；完整协议边界见
/// [CORE-PUBLIC-TYPES-001](../docs/CORE_PUBLIC_TYPES_CONTRACT.md#core-public-types-version-payload)。
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub enum Version {
    /// 最新公开状态由插入或更新产生。
    Upsert(Guid),
    /// 最新公开状态由逻辑删除产生，或首次观察确认 Key 不存在。
    Delete(Guid),
}

/// 表、Key 和公开版本的 owned 联合载荷。
///
/// 该结构同时用于 `query_with_version` 的观察结果、`prepare_with_version` 的 read-set 和
/// `commit_with_version` 的业务回执；相同字段布局不表示三个阶段可以互换。返回值是创建时的
/// 值对象而不是实时版本缓存 view，clone 会共享 `Atom`/`Binary` owner 并复制小型 `Version`。
/// 构造字段不会验证表是否存在、Key 是否为匹配 Meta 的规范 BON，也不会登记 read-set、刷新
/// TTL 或选择事务协议。相等和 Hash 按三字段内容计算，合法 Key 的比较前提与 [`crate::Binary`]
/// 相同。
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct TableKeyVersion {
    /// 表名。
    pub table: Atom,
    /// Key 的规范二进制编码。
    pub key: Binary,
    /// 与该 Table/Key 状态对应的公开版本。
    pub version: Version,
}

/// 表名和 Key 的联合标识。
///
/// 当前内部成功路径用它索引版本事务的最终写集合；版本冲突错误使用携带类型的
/// [`TableKeyConflict`]。字段均为 owned 值，不借用表或事务，也不是数据库存在性证明。clone
/// 共享 `Atom`/`Binary` owner；相等和 Hash 按字段内容计算，不访问版本缓存、表、WAL 或 I/O。
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct TableKey {
    /// 表名。
    pub table: Atom,
    /// 冲突 Key 的规范二进制编码。
    pub key: Binary,
}

/// 版本事务预提交失败时，单个 Table/Key 的主冲突类型。
///
/// 该类型只描述本次实际执行检查所观察到的阻断阶段，不穷举因 Phase 1 短路而没有执行的
/// 潜在检查。同一 Key 同时观察到两类冲突时，版本失配优先，因为外部必须先失效 Value/Version
/// 缓存并重新执行 `query_with_version`。完整分类、归并和下游映射见
/// [VERSION-CONFLICT-KIND-001](../docs/VERSION_CONFLICT_KIND_DESIGN.md#version-conflict-kind-design-index)。
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum VersionConflictKind {
    /// `read_set` 的版本在缓存中缺失、被淘汰、与当前版本不等，或对应只读表身份已变化。
    ReadSetVersionMismatch,
    /// read-set 版本未被判为失配，但数据基线、revision 或其它 prepared 预留发生冲突。
    TransactionConflict,
}

impl VersionConflictKind {
    /// 合并同一 Table/Key 的冲突类型；read-set 失配按冻结协议具有确定性优先级。
    #[inline]
    fn merge(self, other: Self) -> Self {
        if matches!(self, Self::ReadSetVersionMismatch)
            || matches!(other, Self::ReadSetVersionMismatch) {
            Self::ReadSetVersionMismatch
        } else {
            Self::TransactionConflict
        }
    }
}

/// 版本事务完整冲突集合中的单项。
///
/// 字段均为 owned 值，错误返回后不借用事务、表或版本缓存。`kind` 不携带 expected/current
/// Version；下游只能据此选择失效缓存或稍后重试，不能直接刷新 Value/Version 对。
/// 公开 wire 迁移见
/// [pi_db_server 对接](../docs/PI_DB_SERVER_KEY_VERSION_API_HANDOFF.md#pi-db-server-version-conflict-kind)。
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct TableKeyConflict {
    /// 表名。
    pub table: Atom,
    /// 冲突 Key 的规范二进制编码。
    pub key: Binary,
    /// 本次实际阻断预提交的主冲突类型。
    pub kind: VersionConflictKind,
}

/// 按公开协议要求对冲突集合执行原始字节排序和去重。
///
/// 相同 Table/Key 的多项会合并为一项；只要任一项为 read-set 版本失配，最终类型就是
/// `ReadSetVersionMismatch`。该函数只在错误路径执行，不读取数据库共享状态。
pub(crate) fn normalize_conflicts(
    mut conflicts: Vec<TableKeyConflict>,
) -> Vec<TableKeyConflict> {
    conflicts.sort_by(|left, right| {
        left
            .table
            .as_str()
            .as_bytes()
            .cmp(right.table.as_str().as_bytes())
            .then_with(|| left.key.as_ref().cmp(right.key.as_ref()))
    });
    conflicts.dedup_by(|next, previous| {
        if previous.table.as_str().as_bytes() == next.table.as_str().as_bytes()
            && previous.key.as_ref() == next.key.as_ref() {
            previous.kind = previous.kind.merge(next.kind);
            true
        } else {
            false
        }
    });
    conflicts
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum VersionSource {
    FirstObservation,
    CommittedWrite,
}

#[derive(Debug, Clone)]
pub(crate) struct VersionRecord {
    pub(crate) version: Version,
    pub(crate) source: VersionSource,
    pub(crate) revision: u64,
    deadline_tick: u64,
    generation: u64,
}

/// 首次观察一次 DashMap entry 操作的结果。
///
/// `Inserted` 允许调用方使用 entry 前读取的值；`Occupied` 表示期间已有首次观察或提交，
/// 调用方必须在 entry guard 已释放后重读值。它不描述版本自身来自观察还是写提交。
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum FirstObservation {
    Inserted(Version),
    Occupied(Version),
}

impl VersionRecord {
    fn exact_eq(&self, other: &Self) -> bool {
        self.version == other.version
            && self.source == other.source
            && self.revision == other.revision
            && self.deadline_tick == other.deadline_tick
            && self.generation == other.generation
    }
}

/// builder 中 TTL 配置换算后的单调毫秒参数。
#[derive(Debug, Clone, Copy)]
pub(crate) struct KeyVersionConfig {
    ttl_ticks: Option<u64>,
    poll_interval_ticks: u64,
}

impl KeyVersionConfig {
    /// 在数据库目录和文件副作用前校验并换算配置。
    pub(crate) fn new(ttl: Duration, poll_interval: Duration) -> IOResult<Self> {
        if ttl.is_zero() {
            return Ok(Self {
                ttl_ticks: None,
                poll_interval_ticks: 0,
            });
        }
        if poll_interval.is_zero() {
            return Err(Error::new(ErrorKind::InvalidInput,
                                  "Start database failed, reason: key version TTL is enabled but poll interval is zero"));
        }

        Ok(Self {
            ttl_ticks: Some(duration_to_ticks(ttl)),
            poll_interval_ticks: duration_to_ticks(poll_interval),
        })
    }

    fn ttl_ticks(&self) -> Option<u64> {
        self.ttl_ticks
    }

    fn poll_interval_ticks(&self) -> u64 {
        self.poll_interval_ticks
    }
}

fn duration_to_ticks(duration: Duration) -> u64 {
    // TTL 的最小单位是 1ms：非零 sub-ms 值按 1ms 处理，其余不足 1ms 的小数按契约忽略。
    // 历史 BUG-KV-TTL-001 由 deadline 基准向下取整导致；这里的配置量化不是缺陷且保持不变。
    let millis = duration.as_millis();
    if millis == 0 {
        1
    } else if millis >= MAX_TICK as u128 {
        MAX_TICK
    } else {
        millis as u64
    }
}

fn timeout_ticks(ticks: u64) -> usize {
    if ticks >= usize::MAX as u64 {
        usize::MAX
    } else {
        ticks as usize
    }
}

#[derive(Clone)]
pub(crate) struct KeyVersionRegistry(Arc<InnerKeyVersionRegistry>);

/// trace 构建中公开版本 API 的固定低基数操作种类。
#[cfg(feature = "trace")]
#[derive(Clone, Copy)]
pub(crate) enum KeyVersionApiOperation {
    Query,
    Prepare,
    Commit,
}

/// tracing loop 一次读取的版本 API 累计计数快照。
#[cfg(feature = "trace")]
#[derive(Clone, Copy, Default, Debug, PartialEq, Eq)]
pub(crate) struct KeyVersionApiMetricsSnapshot {
    pub(crate) query_success: u64,
    pub(crate) query_failure: u64,
    pub(crate) prepare_success: u64,
    pub(crate) prepare_failure: u64,
    pub(crate) commit_success: u64,
    pub(crate) commit_failure: u64,
}

#[cfg(feature = "trace")]
impl KeyVersionApiMetricsSnapshot {
    /// 计算两次累计快照之间的 Counter 增量；wrapping 使单次 u64 回绕仍保持模运算正确。
    pub(crate) fn delta_since(self, previous: Self) -> Self {
        Self {
            query_success: self.query_success.wrapping_sub(previous.query_success),
            query_failure: self.query_failure.wrapping_sub(previous.query_failure),
            prepare_success: self.prepare_success.wrapping_sub(previous.prepare_success),
            prepare_failure: self.prepare_failure.wrapping_sub(previous.prepare_failure),
            commit_success: self.commit_success.wrapping_sub(previous.commit_success),
            commit_failure: self.commit_failure.wrapping_sub(previous.commit_failure),
        }
    }
}

/// 每个合法版本 API 调用只向 success/failure 中一个原子提交一次结果。
///
/// guard 不持有 registry owner 或任何锁；未显式完成即表示 future 被取消或发生 unwind，Drop
/// 将其归入 failure。指标是旁路观测，不参与 API 错误和事务状态判断。
#[cfg(feature = "trace")]
pub(crate) struct KeyVersionApiCallGuard<'a> {
    success: &'a AtomicU64,
    failure: &'a AtomicU64,
    completed: bool,
}

#[cfg(feature = "trace")]
impl KeyVersionApiCallGuard<'_> {
    pub(crate) fn finish(mut self, success: bool) {
        self.completed = true;
        if success {
            self.success.fetch_add(1, Ordering::Relaxed);
        } else {
            self.failure.fetch_add(1, Ordering::Relaxed);
        }
    }
}

#[cfg(feature = "trace")]
impl Drop for KeyVersionApiCallGuard<'_> {
    fn drop(&mut self) {
        if !self.completed {
            self.failure.fetch_add(1, Ordering::Relaxed);
        }
    }
}

#[cfg(feature = "trace")]
#[derive(Default)]
struct KeyVersionApiMetrics {
    query_success: AtomicU64,
    query_failure: AtomicU64,
    prepare_success: AtomicU64,
    prepare_failure: AtomicU64,
    commit_success: AtomicU64,
    commit_failure: AtomicU64,
}

struct InnerKeyVersionRegistry {
    tables: DashMap<Atom, KeyVersions>,
    ttl_ticks: Option<u64>,
    poll_interval_ticks: u64,
    origin: Instant,
    shutdown_tx: Option<Sender<()>>,
    #[cfg(feature = "trace")]
    api_metrics: KeyVersionApiMetrics,
}

impl KeyVersionRegistry {
    pub(crate) fn new(config: KeyVersionConfig) -> (Self, Option<Receiver<()>>) {
        let (shutdown_tx, shutdown_rx) = if config.ttl_ticks().is_some() {
            let (tx, rx) = bounded(1);
            (Some(tx), Some(rx))
        } else {
            (None, None)
        };
        let inner = InnerKeyVersionRegistry {
            tables: DashMap::new(),
            ttl_ticks: config.ttl_ticks(),
            poll_interval_ticks: config.poll_interval_ticks(),
            origin: Instant::now(),
            shutdown_tx,
            #[cfg(feature = "trace")]
            api_metrics: KeyVersionApiMetrics::default(),
        };
        (Self(Arc::new(inner)), shutdown_rx)
    }

    pub(crate) fn create_table_versions(&self) -> KeyVersions {
        KeyVersions::new(Arc::downgrade(&self.0), self.0.ttl_ticks.is_some())
    }

    pub(crate) fn install(&self, table: Atom, versions: KeyVersions) {
        self.0.tables.insert(table, versions);
    }

    pub(crate) fn remove_exact(&self, table: &Atom, versions: &KeyVersions) {
        let _ = self
            .0
            .tables
            .remove_if(table, |_name, current| current.ptr_eq(versions));
    }

    /// 创建一次公开版本 API 结果 guard；只在 trace 构建中存在且不获取任何锁。
    #[cfg(feature = "trace")]
    pub(crate) fn begin_api_call(&self,
                                 operation: KeyVersionApiOperation)
        -> KeyVersionApiCallGuard<'_> {
        let metrics = &self.0.api_metrics;
        let (success, failure) = match operation {
            KeyVersionApiOperation::Query => {
                (&metrics.query_success, &metrics.query_failure)
            },
            KeyVersionApiOperation::Prepare => {
                (&metrics.prepare_success, &metrics.prepare_failure)
            },
            KeyVersionApiOperation::Commit => {
                (&metrics.commit_success, &metrics.commit_failure)
            },
        };
        KeyVersionApiCallGuard {
            success,
            failure,
            completed: false,
        }
    }

    /// 原子读取当前累计 API 计数；不同字段是最终收敛的观测值，不构成事务型成对快照。
    #[cfg(feature = "trace")]
    pub(crate) fn api_metrics_snapshot(&self) -> KeyVersionApiMetricsSnapshot {
        let metrics = &self.0.api_metrics;
        KeyVersionApiMetricsSnapshot {
            query_success: metrics.query_success.load(Ordering::Relaxed),
            query_failure: metrics.query_failure.load(Ordering::Relaxed),
            prepare_success: metrics.prepare_success.load(Ordering::Relaxed),
            prepare_failure: metrics.prepare_failure.load(Ordering::Relaxed),
            commit_success: metrics.commit_success.load(Ordering::Relaxed),
            commit_failure: metrics.commit_failure.load(Ordering::Relaxed),
        }
    }

    /// 修复完成后清空外部不可见的恢复期版本记录，但保留单调 revision。
    pub(crate) fn clear_records(&self) {
        for entry in self.0.tables.iter() {
            entry.value().clear_records();
        }
    }

    /// 启动唯一固定周期 TTL 任务。TTL 关闭时 receiver 为 None，不创建任务。
    pub(crate) fn start_ttl_task(&self,
                                 rt: MultiTaskRuntime<()>,
                                 receiver: Option<Receiver<()>>) {
        let Some(receiver) = receiver else {
            return;
        };
        let weak = Arc::downgrade(&self.0);
        let task_rt = rt.clone();
        let _ = rt.spawn(async move {
            ttl_loop(task_rt, weak, receiver).await;
        });
    }
}

async fn ttl_loop(rt: MultiTaskRuntime<()>,
                  registry: Weak<InnerKeyVersionRegistry>,
                  receiver: Receiver<()>) {
    let Some(initial) = registry.upgrade() else {
        return;
    };
    let interval = initial.poll_interval_ticks;
    let ttl = initial.ttl_ticks.unwrap_or(0);
    drop(initial);

    log::info!(target: "pi_db::key_version_ttl",
               "Key version TTL task started, ttl_ms: {}, interval_ms: {}",
               ttl,
               interval);
    let mut round = 0u64;
    loop {
        let timeout = rt.timeout(timeout_ticks(interval)).fuse();
        let shutdown = receiver.recv().fuse();
        futures::pin_mut!(timeout, shutdown);
        match select(timeout, shutdown).await {
            Either::Left((_timeout_result, _shutdown_future)) => (),
            Either::Right((_shutdown_result, _timeout_future)) => break,
        }

        let Some(registry_ref) = registry.upgrade() else {
            break;
        };
        round = round.saturating_add(1);
        let statistics = collect_ttl_round(&rt, &registry_ref, round).await;
        drop(registry_ref);
        statistics.log();
    }
    log::info!(target: "pi_db::key_version_ttl",
               "Key version TTL task stopped, ttl_ms: {}, interval_ms: {}, completed_rounds: {}",
               ttl,
               interval,
               round);
}

#[derive(Default)]
struct TtlCollectStatistics {
    round: u64,
    ttl_ticks: u64,
    interval_ticks: u64,
    elapsed: Duration,
    registered_tables: usize,
    due_tables: usize,
    scanned_tables: usize,
    scanned_records: usize,
    due_candidates: usize,
    removed_records: usize,
    removed_first_observations: usize,
    removed_committed_writes: usize,
    first_observation_blocked: usize,
    snapshot_blocked: usize,
    stale_candidates: usize,
    records_before: usize,
    records_after: usize,
    batches: usize,
    yields: usize,
    next_deadline: u64,
}

impl TtlCollectStatistics {
    fn log(&self) {
        log::info!(target: "pi_db::key_version_ttl",
                   "Key version TTL round completed, round: {}, ttl_ms: {}, interval_ms: {}, elapsed_ms: {}, registered_tables: {}, due_tables: {}, scanned_tables: {}, scanned_records: {}, due_candidates: {}, removed_records: {}, removed_first_observations: {}, removed_committed_writes: {}, first_observation_blocked: {}, snapshot_blocked: {}, stale_candidates: {}, records_before: {}, records_after: {}, batches: {}, yields: {}, next_deadline_tick: {}",
                   self.round,
                   self.ttl_ticks,
                   self.interval_ticks,
                   self.elapsed.as_millis(),
                   self.registered_tables,
                   self.due_tables,
                   self.scanned_tables,
                   self.scanned_records,
                   self.due_candidates,
                   self.removed_records,
                   self.removed_first_observations,
                   self.removed_committed_writes,
                   self.first_observation_blocked,
                   self.snapshot_blocked,
                   self.stale_candidates,
                   self.records_before,
                   self.records_after,
                   self.batches,
                   self.yields,
                   self.next_deadline);
    }
}

async fn collect_ttl_round(rt: &MultiTaskRuntime<()>,
                           registry: &Arc<InnerKeyVersionRegistry>,
                           round: u64) -> TtlCollectStatistics {
    let started = Instant::now();
    let now = monotonic_tick(registry.origin);
    let mut statistics = TtlCollectStatistics {
        round,
        ttl_ticks: registry.ttl_ticks.unwrap_or(0),
        interval_ticks: registry.poll_interval_ticks,
        next_deadline: NO_DEADLINE,
        ..TtlCollectStatistics::default()
    };
    // 这是 TTL 唯一保留的 DashMap iterator：它只枚举通常远少于 Key 数的表级 registry，
    // 在同步表达式内克隆 owner 后立即释放全部分片 guard，且不与 DDL 写、Key Map 写或 await
    // 交叠。它仍可能短暂延迟同分片 DDL；当前合法协议不允许 DDL 与事务读写混用，若未来放宽
    // 该边界必须改用独立表索引。后续统计复用同一快照，避免本轮第二次枚举。验收见
    // docs/KEY_VERSION_TTL_FIFO_ACCEPTANCE.md#kv-ttl-fifo-outer-registry。
    let tables: Vec<KeyVersions> = registry
        .tables
        .iter()
        .map(|entry| entry.value().clone())
        .collect();
    statistics.registered_tables = tables.len();
    statistics.records_before = tables.iter().map(KeyVersions::len).sum();

    for versions in &tables {
        let earliest = versions.0.earliest_deadline.load(Ordering::Acquire);
        let blocked_changed = versions.0.has_blocked_expiry.load(Ordering::Acquire)
            && versions.0.lease_epoch.load(Ordering::Acquire)
                != versions.0.blocked_lease_epoch.load(Ordering::Acquire);
        if earliest > now && !blocked_changed {
            statistics.next_deadline = statistics.next_deadline.min(earliest);
            continue;
        }

        statistics.due_tables += 1;
        statistics.scanned_tables += 1;
        versions
            .collect_expired(rt, now, &mut statistics)
            .await;
        statistics.next_deadline = statistics
            .next_deadline
            .min(versions.0.earliest_deadline.load(Ordering::Acquire));
    }

    statistics.records_after = tables.iter().map(KeyVersions::len).sum();
    statistics.elapsed = started.elapsed();
    statistics
}

fn monotonic_tick(origin: Instant) -> u64 {
    let elapsed = origin.elapsed().as_millis();
    if elapsed >= MAX_TICK as u128 {
        MAX_TICK
    } else {
        elapsed as u64
    }
}

fn deadline_tick(elapsed: Duration, ttl_ticks: u64) -> u64 {
    let elapsed_millis = elapsed.as_millis();
    let rounded_millis = if elapsed.subsec_nanos() % 1_000_000 == 0 {
        elapsed_millis
    } else {
        elapsed_millis.saturating_add(1)
    };
    let base_tick = if rounded_millis >= MAX_TICK as u128 {
        MAX_TICK
    } else {
        rounded_millis as u64
    };
    base_tick.saturating_add(ttl_ticks).min(MAX_TICK)
}

/// 每表 TTL 活动 Key 的非等待 FIFO 索引。
///
/// DashMap 仍是版本状态的唯一事实来源。索引只保存 O(1) 的 `Binary` 共享 owner，使 scanner
/// 可以分批执行单 Key 点读而无需持有 DashMap 分片 iterator。sender/receiver 由同一对象共同
/// 持有，因此 unbounded channel 在对象存活期间不会断开，也不会因容量产生 Full。
/// 设计、竞态和生命周期证据见
/// `docs/KEY_VERSION_TTL_FIFO_ACCEPTANCE.md#kv-ttl-fifo-algorithm`。
struct TtlKeyIndex {
    sender: SyncSender<Binary>,
    receiver: SyncReceiver<Binary>,
}

impl TtlKeyIndex {
    fn new() -> Self {
        let (sender, receiver) = sync_unbounded();
        Self { sender, receiver }
    }

    fn len(&self) -> usize {
        self.receiver.len()
    }

    fn push(&self, key: Binary) {
        if self.sender.try_send(key).is_err() {
            unreachable!("TTL key index channel disconnected while its receiver is alive");
        }
    }

    fn push_batch(&self, keys: Vec<Binary>) {
        for key in keys {
            self.push(key);
        }
    }

    fn take_batch(&self, limit: usize) -> Vec<Binary> {
        let mut keys = Vec::with_capacity(limit);
        for _ in 0..limit {
            match self.receiver.try_recv() {
                Ok(key) => keys.push(key),
                Err(TryRecvError::Empty) => break,
                Err(TryRecvError::Disconnected) => {
                    unreachable!("TTL key index channel disconnected while its sender is alive");
                },
            }
        }
        keys
    }

    fn clear(&self) {
        loop {
            match self.receiver.try_recv() {
                Ok(_) => (),
                Err(TryRecvError::Empty) => break,
                Err(TryRecvError::Disconnected) => {
                    unreachable!("TTL key index channel disconnected while its sender is alive");
                },
            }
        }
    }
}

/// 估算一个活动版本记录可归属于版本缓存的动态字节数。
///
/// 该 O(1) 公式使用 Key 的实际 Vec capacity，并计算 Map 逻辑 entry、Arc<Vec> 控制字段及
/// TTL 开启时的 FIFO owner/slot 状态。它不读取 DashMap capacity，也不声称包含 allocator、
/// shard 空闲 bucket 或 channel 空闲 block；完整口径见
/// `docs/KEY_VERSION_CACHE_METRICS_DESIGN.md#key-version-metrics-memory`。
#[cfg(feature = "trace")]
fn estimated_record_memory_bytes(key: &Binary, ttl_enabled: bool) -> u64 {
    let map_entry = mem::size_of::<(Binary, VersionRecord)>();
    let arc_vec = mem::size_of::<AtomicUsize>()
        .saturating_mul(2)
        .saturating_add(mem::size_of::<Vec<u8>>())
        .saturating_add(key.0.capacity());
    let ttl_slot = if ttl_enabled {
        mem::size_of::<(Binary, AtomicUsize)>()
    } else {
        0
    };
    map_entry
        .saturating_add(arc_vec)
        .saturating_add(ttl_slot) as u64
}

#[derive(Clone)]
pub(crate) struct KeyVersions(Arc<InnerKeyVersions>);

/// tracing loop 一次读取的每表版本缓存状态。
#[cfg(feature = "trace")]
#[derive(Clone, Copy, Default, Debug, PartialEq, Eq)]
pub(crate) struct KeyVersionCacheMetricsSnapshot {
    pub(crate) record_count: u64,
    pub(crate) estimated_memory_bytes: u64,
}

#[cfg(feature = "trace")]
#[derive(Default)]
struct KeyVersionCacheMetrics {
    record_count: AtomicU64,
    estimated_memory_bytes: AtomicU64,
}

/// 单个已注册表实例的版本状态。
///
/// `versions` 是 `(Binary -> VersionRecord)` 的唯一事实来源；`ttl_keys` 只是避免遍历
/// DashMap 的活动 Key 索引。`publication` 仅为暂停重构且禁止外部使用的 LogWrite 保留；
/// Meta、Memory、LogOrdered 和 Btree 不再取得它。TTL 与在用表依赖 DashMap 单 Key 原子操作、
/// 完整记录签名和 `active_first_observations` 安全处理竞态。后者不是锁，只在 TTL 开启且首次
/// 版本点读缺失时短暂计数，禁止 scanner 在值读取和首次观察登记之间删除版本。
/// `completed_revision` 与 `active_snapshots` 共同保存普通事务识别同值写和 ABA 所需的最小历史窗口。
///
/// 强引用方向固定为 registry/table/transaction -> KeyVersions，反向只允许 Weak registry；
/// TTL task 同样只持 Weak registry，禁止形成数据库或表无法释放的 Arc 环。
struct InnerKeyVersions {
    versions: DashMap<Binary, VersionRecord>,
    ttl_keys: TtlKeyIndex,
    pub(crate) publication: RwLock<()>,
    ttl_enabled: bool,
    active_first_observations: AtomicUsize,
    completed_revision: AtomicU64,
    active_snapshots: Mutex<BTreeMap<u64, usize>>,
    lease_epoch: AtomicU64,
    earliest_deadline: AtomicU64,
    has_blocked_expiry: AtomicBool,
    blocked_lease_epoch: AtomicU64,
    registry: Weak<InnerKeyVersionRegistry>,
    #[cfg(feature = "trace")]
    metrics: KeyVersionCacheMetrics,
}

impl KeyVersions {
    fn new(registry: Weak<InnerKeyVersionRegistry>, ttl_enabled: bool) -> Self {
        Self(Arc::new(InnerKeyVersions {
            versions: DashMap::new(),
            ttl_keys: TtlKeyIndex::new(),
            publication: RwLock::new(()),
            ttl_enabled,
            active_first_observations: AtomicUsize::new(0),
            completed_revision: AtomicU64::new(0),
            active_snapshots: Mutex::new(BTreeMap::new()),
            lease_epoch: AtomicU64::new(0),
            earliest_deadline: AtomicU64::new(NO_DEADLINE),
            has_blocked_expiry: AtomicBool::new(false),
            blocked_lease_epoch: AtomicU64::new(0),
            registry,
            #[cfg(feature = "trace")]
            metrics: KeyVersionCacheMetrics::default(),
        }))
    }

    pub(crate) fn ptr_eq(&self, other: &Self) -> bool {
        Arc::ptr_eq(&self.0, &other.0)
    }

    /// 返回兼容期 LogWrite 的每表 publication 门。
    ///
    /// LogWrite 当前禁止外部使用且本轮不改变其提交协议；其它在用表不得重新调用此方法。
    /// 后续单独重构 LogWrite 时应连同本字段和方法一起重新评估或移除。
    pub(crate) fn publication(&self) -> &RwLock<()> {
        &self.0.publication
    }

    pub(crate) fn len(&self) -> usize {
        self.0.versions.len()
    }

    pub(crate) fn current(&self, key: &Binary) -> Option<VersionRecord> {
        self.0.versions.get(key).map(|record| record.clone())
    }

    pub(crate) fn current_version(&self, key: &Binary) -> Option<Version> {
        self.0
            .versions
            .get(key)
            .map(|record| record.version.clone())
    }

    /// 在一次版本缺失查询读取表数据前建立非阻塞的首次观察租约。
    ///
    /// TTL 关闭时返回无操作 guard；TTL 开启时只做一次 checked 原子增加。scanner 在取得到期
    /// 候选后、exact-remove 前观察该计数，非零时保留候选。guard 不可 clone，借用当前
    /// `KeyVersions`，作用域内不得 await；所有返回、`?` 和 unwind 都由 Drop 精确递减并推进
    /// `lease_epoch`，使被延迟的候选最迟在下一轮重新检查。
    ///
    /// 安全性不依赖“Acquire load 总能读取墙钟意义上的最新值”。计数增加严格先于查询的第二次
    /// 同 Key DashMap 点读；若该点读仍为缺失，则后来插入新候选的 commit 及读取该候选的
    /// scanner 必须依次经过同一分片锁，形成从计数增加到 scanner 检查的同步链。若 scanner
    /// 更早取得候选，则由其 exact-remove 与查询二次点读的分片锁先后关系保证查询看到删除前
    /// 版本或删除后的当前数据。改变 DashMap 或调整二次点读顺序时必须重新证明该关系。
    ///
    /// `usize::MAX` 个同时存活的 guard 超过进程地址空间可容纳的对象数量；checked 更新仍在
    /// 修改计数前 fail-fast，防止内部状态破坏时回绕为零并错误放行 TTL。
    pub(crate) fn lease_first_observation(&self) -> FirstObservationLease<'_> {
        if self.0.ttl_enabled {
            let mut observed = self.0.active_first_observations.load(Ordering::Acquire);
            loop {
                let next = observed
                    .checked_add(1)
                    .expect("active first-observation lease count overflow");
                match self.0.active_first_observations.compare_exchange_weak(
                    observed,
                    next,
                    Ordering::AcqRel,
                    Ordering::Acquire) {
                    Ok(_) => break,
                    Err(current) => observed = current,
                }
            }
            FirstObservationLease {
                versions: self,
                active: true,
            }
        } else {
            FirstObservationLease {
                versions: self,
                active: false,
            }
        }
    }

    /// scanner 的候选后检查点；Acquire 与租约取得/释放的读改写形成同一原子修改顺序。
    #[inline]
    fn has_active_first_observation(&self) -> bool {
        self.0.active_first_observations.load(Ordering::Acquire) != 0
    }

    /// 返回无锁、O(1) 的 trace 指标快照；不得把该观测值用于事务或 TTL 正确性判断。
    #[cfg(feature = "trace")]
    pub(crate) fn metrics_snapshot(&self) -> KeyVersionCacheMetricsSnapshot {
        KeyVersionCacheMetricsSnapshot {
            record_count: self.0.metrics.record_count.load(Ordering::Relaxed),
            estimated_memory_bytes: self
                .0
                .metrics
                .estimated_memory_bytes
                .load(Ordering::Relaxed),
        }
    }

    /// 只在持有 Vacant entry、确定即将新增结构记录时调用；已有 Key 替换不改变容量指标。
    /// 调用必须先于 entry guard 释放，使 TTL 不可能在指标建立前看到并删除该记录。
    #[cfg(feature = "trace")]
    fn record_inserted(&self, estimated_memory_bytes: u64) {
        self.0
            .metrics
            .estimated_memory_bytes
            .fetch_add(estimated_memory_bytes, Ordering::Relaxed);
        self.0.metrics.record_count.fetch_add(1, Ordering::Relaxed);
    }

    /// 只在 exact-remove 确实移除结构记录后调用，与首次插入严格一一配平。
    #[cfg(feature = "trace")]
    fn record_removed(&self, estimated_memory_bytes: u64) {
        let old_count = self.0.metrics.record_count.fetch_sub(1, Ordering::Relaxed);
        let old_memory = self
            .0
            .metrics
            .estimated_memory_bytes
            .fetch_sub(estimated_memory_bytes, Ordering::Relaxed);
        debug_assert!(old_count >= 1,
                      "key version metric record count underflow");
        debug_assert!(old_memory >= estimated_memory_bytes,
                      "key version metric estimated memory underflow");
    }

    /// 返回当前逻辑值对应的现有版本，或为首次观察原子创建一个版本。
    ///
    /// 调用方必须先读取一次逻辑值，再根据返回结果决定能否直接使用该值：Vacant 表示本调用
    /// 按该值的存在状态完成首次登记；Occupied 表示期间已有首次观察或提交，调用方必须重读
    /// 逻辑值，禁止把第一次旧值与竞争产生的新版本配对。TTL 随后删除版本只会使 prepare
    /// 保守冲突。
    ///
    /// 命中路径 O(1) 平均时间且不分配 Guid；缺失路径分配一个 Guid、一个记录和至多一个 FIFO
    /// token。方法不 await，不获取 prepare/root/cache 锁，也不执行 WAL 或数据文件 I/O。
    pub(crate) fn first_observation<F>(&self,
                                       key: Binary,
                                       exists: bool,
                                       alloc_uid: F) -> FirstObservation
        where F: FnOnce() -> Guid
    {
        match self.0.versions.entry(key) {
            Entry::Occupied(entry) => FirstObservation::Occupied(entry.get().version.clone()),
            Entry::Vacant(entry) => {
                let version = if exists {
                    Version::Upsert(alloc_uid())
                } else {
                    Version::Delete(alloc_uid())
                };
                let deadline = self.next_deadline();
                let record = VersionRecord {
                    version: version.clone(),
                    source: VersionSource::FirstObservation,
                    revision: self.0.completed_revision.load(Ordering::Acquire),
                    deadline_tick: deadline,
                    generation: 1,
                };
                let ttl_key = if deadline == NO_DEADLINE {
                    None
                } else {
                    Some(entry.key().clone())
                };
                #[cfg(feature = "trace")]
                let estimated_memory_bytes = estimated_record_memory_bytes(
                    entry.key(),
                    deadline != NO_DEADLINE);
                // 指标必须在记录对 TTL scanner 可见前建立，否则 scanner 可能先删除并递减尚未
                // 增加的计数。原子更新不取锁；FIFO 操作仍必须等 entry guard 释放后执行。
                #[cfg(feature = "trace")]
                self.record_inserted(estimated_memory_bytes);
                drop(entry.insert(record));
                if let Some(ttl_key) = ttl_key {
                    self.0.ttl_keys.push(ttl_key);
                }
                self.register_deadline(deadline);
                FirstObservation::Inserted(version)
            },
        }
    }

    /// 租用当前已完成 revision，防止 TTL 提前删除本事务识别后续提交所需的版本记录。
    ///
    /// 调用方必须同时持有该表的数据 root/cache guard：commit 在相同 guard 内先发布数据和版本，
    /// 再 Release-store 新 revision，因此“数据快照 + revision + 活跃租约”不会被一次 commit
    /// 撕裂。不得在 prepare guard 内调用本方法；内部仅短暂获取 `active_snapshots` 同步
    /// mutex，不 await、不执行 I/O。
    ///
    /// 返回 lease 可跨线程移动，显式 release 与 Drop 均幂等。活跃计数按 revision 聚合，创建和
    /// 释放平均 O(log r)，其中 r 是当前活跃 revision 数，而不是 Key 数。
    pub(crate) fn lease_current(&self) -> SnapshotLease {
        let revision = self.0.completed_revision.load(Ordering::Acquire);
        let mut active = self.0.active_snapshots.lock();
        let count = active.entry(revision).or_insert(0);
        *count = count.saturating_add(1);
        drop(active);
        SnapshotLease {
            versions: self.clone(),
            revision,
            released: AtomicBool::new(false),
        }
    }

    /// Acquire-load 最近一次完整发布的数据/版本 revision。
    pub(crate) fn completed_revision(&self) -> u64 {
        self.0.completed_revision.load(Ordering::Acquire)
    }

    /// 计算下一 revision；只能在本表数据 root/cache guard 内用于一次实际写提交。
    ///
    /// 返回 None 表示 u64 空间耗尽。调用方必须在修改数据前把它提升为不可 rollback 的 Fatal，
    /// 绝不能回绕或复用 revision。
    pub(crate) fn checked_next_revision(&self) -> Option<u64> {
        self.completed_revision().checked_add(1)
    }

    /// 在本表全部数据和写 Key 版本完成发布后推进 completed revision。
    ///
    /// 调用方必须仍持有对应数据 root/cache guard。Release-store 与事务创建侧的 Acquire-load
    /// 配对；本方法不校验单调性，因为同一 guard 串行化该表所有合法 managed writer。
    pub(crate) fn complete_revision(&self, revision: u64) {
        self.0.completed_revision.store(revision, Ordering::Release);
    }

    /// 判断该 Key 是否存在晚于事务快照的真实提交版本。
    ///
    /// FirstObservation 不表示写提交，不能制造普通事务冲突；CommittedWrite 的最新记录足以代表
    /// 此 Key 在快照后至少发生过一次写。调用方在 prepare 最终围栏中完成复核；活跃 lease
    /// 同时保证 TTL 不会删除 `revision > snapshot_revision` 的必要证据。
    pub(crate) fn has_committed_after(&self,
                                      key: &Binary,
                                      snapshot_revision: u64) -> bool {
        if let Some(record) = self.0.versions.get(key) {
            record.source == VersionSource::CommittedWrite
                && record.revision > snapshot_revision
        } else {
            false
        }
    }

    /// 发布本事务对一个 Key 的最终版本，并返回只描述本事务的回执项。
    ///
    /// 调用方必须持有本表数据 root/cache guard，并传入当前根事务的 TID 与本表本次提交唯一
    /// revision。`Some` 生成 Upsert，`None` 生成 Delete；不会读取提交后的“全局最新版本”，
    /// 所以随后其它事务覆盖该 Key 也不会改变已返回回执。
    ///
    /// TTL 可并发 exact-remove，但 DashMap 分片原子操作保证两种结果都保有有效 token：删除先发生
    /// 时本次 insert 视为首次并入队，更新先发生时 scanner 看到签名变化并归还原 token。方法不
    /// await、不获取 prepare 锁、不执行 I/O；平均 O(1)，首次记录额外创建一个 O(1) Binary owner。
    pub(crate) fn publish(&self,
                          table: Atom,
                          key: Binary,
                          value: Option<&Binary>,
                          transaction_uid: Guid,
                          revision: u64) -> TableKeyVersion {
        let version = if value.is_some() {
            Version::Upsert(transaction_uid)
        } else {
            Version::Delete(transaction_uid)
        };
        let deadline = self.next_deadline();
        // generation 读取和记录替换必须位于同一个 Key 的 entry guard 内。query 首次观察可与
        // 提交并发；若继续先 get 后 insert，竞争插入可能被覆盖且 token 判断
        // 使用过期结果。这里没有 await、表锁或 prepare 锁。
        let inserted = match self.0.versions.entry(key.clone()) {
            Entry::Occupied(mut entry) => {
                let generation = entry.get().generation.saturating_add(1);
                entry.insert(VersionRecord {
                    version: version.clone(),
                    source: VersionSource::CommittedWrite,
                    revision,
                    deadline_tick: deadline,
                    generation,
                });
                false
            },
            Entry::Vacant(entry) => {
                let record = VersionRecord {
                    version: version.clone(),
                    source: VersionSource::CommittedWrite,
                    revision,
                    deadline_tick: deadline,
                    generation: 1,
                };
                // trace 容量指标必须在记录对 TTL scanner 可见前建立；默认构建没有该原子成本。
                #[cfg(feature = "trace")]
                self.record_inserted(estimated_record_memory_bytes(
                    entry.key(),
                    deadline != NO_DEADLINE));
                drop(entry.insert(record));
                true
            },
        };
        // 已有 Key 的唯一 token 可能正在 scanner 本地批次中；只有首次插入才创建新 token。
        if inserted {
            if deadline != NO_DEADLINE {
                self.0.ttl_keys.push(key.clone());
            }
        }
        self.register_deadline(deadline);
        TableKeyVersion {
            table,
            key,
            version,
        }
    }

    pub(crate) fn clear_records(&self) {
        // 该入口只允许在 repair 完成、外部事务尚不可创建且 TTL task 尚未启动的静默启动期调用。
        // 先清索引再清 Map，随后把旁路指标归零；若未来放宽为并发调用，必须重新设计 Map/指标
        // 的原子清空协议，不能直接复用当前实现。
        debug_assert_eq!(self.0.active_first_observations.load(Ordering::Acquire), 0,
                         "repair-time version clear must not overlap a first-observation query");
        self.0.ttl_keys.clear();
        self.0.versions.clear();
        #[cfg(feature = "trace")]
        {
            self.0.metrics.record_count.store(0, Ordering::Relaxed);
            self.0
                .metrics
                .estimated_memory_bytes
                .store(0, Ordering::Relaxed);
        }
        self.0.earliest_deadline.store(NO_DEADLINE, Ordering::Release);
        self.0.has_blocked_expiry.store(false, Ordering::Release);
        self.0.blocked_lease_epoch.store(
            self.0.lease_epoch.load(Ordering::Acquire),
            Ordering::Release);
    }

    fn next_deadline(&self) -> u64 {
        let Some(registry) = self.0.registry.upgrade() else {
            return NO_DEADLINE;
        };
        let Some(ttl) = registry.ttl_ticks else {
            return NO_DEADLINE;
        };
        // scanner 的当前 tick 向下取整，因此 deadline 基准必须向上取整；否则版本可能比
        // 量化后的有效 TTL 提前不足 1ms 淘汰。这里只改变时间边界，不改变 TTL 的 1ms
        // 量化契约、扫描比较或版本并发控制。证据见 docs/KEY_VERSION_TTL_EARLY_EXPIRY_BUG.md。
        deadline_tick(registry.origin.elapsed(), ttl)
    }

    fn register_deadline(&self, deadline: u64) {
        if deadline != NO_DEADLINE {
            self.0.earliest_deadline.fetch_min(deadline, Ordering::AcqRel);
        }
    }

    async fn collect_expired(&self,
                             rt: &MultiTaskRuntime<()>,
                             now: u64,
                             statistics: &mut TtlCollectStatistics) {
        self.0.earliest_deadline.swap(NO_DEADLINE, Ordering::AcqRel);
        self.0.has_blocked_expiry.store(false, Ordering::Release);
        let scan_epoch = self.0.lease_epoch.load(Ordering::Acquire);
        // 固定本轮开始时的 token 数量。FIFO 中尚未处理的旧 token 始终位于并发新插入和本轮
        // 重新入队 token 之前，因此只消费该固定数量即可把后两类工作严格推迟到后续轮次。
        let mut remaining = self.0.ttl_keys.len();
        let mut blocked = false;

        while remaining > 0 {
            let batch = self
                .0
                .ttl_keys
                .take_batch(remaining.min(TTL_SCAN_BATCH_SIZE));
            if batch.is_empty() {
                break;
            }
            remaining -= batch.len();
            statistics.scanned_records += batch.len();
            statistics.batches += 1;
            let mut retained = Vec::with_capacity(batch.len());
            for key in batch {
                let Some(candidate) = self.current(&key) else {
                    continue;
                };
                if candidate.deadline_tick > now {
                    self.register_deadline(candidate.deadline_tick);
                    retained.push(key);
                    continue;
                }
                statistics.due_candidates += 1;

                // 必须先取得完整候选，再读取租约计数。若在轮次开始时缓存一次 0，查询可能随后
                // 取得租约、提交插入新版本，而 scanner 仍用过期的 0 删除它。查询的租约增加、
                // 二次同 Key 点读、后续 commit 插入和本次 candidate 点读通过同一 DashMap 分片
                // 锁建立同步顺序；这里不能移到 candidate 之前，也不能换成没有等价证明的容器。
                // 计数非零时整表到期项只保留不等待；租约 Drop 推进 lease_epoch，保证最后一个
                // 租约释放后重扫。
                if self.has_active_first_observation() {
                    statistics.first_observation_blocked += 1;
                    blocked = true;
                    retained.push(key);
                    continue;
                }

                if candidate.source == VersionSource::CommittedWrite {
                    let min_active = self
                        .0
                        .active_snapshots
                        .lock()
                        .keys()
                        .next()
                        .copied();
                    if min_active
                        .map(|revision| candidate.revision > revision)
                        .unwrap_or(false) {
                        statistics.snapshot_blocked += 1;
                        blocked = true;
                        retained.push(key);
                        continue;
                    }
                }

                let removed = self
                    .0
                    .versions
                    .remove_if(&key, |_key, current| current.exact_eq(&candidate));
                if removed.is_some() {
                    #[cfg(feature = "trace")]
                    if let Some((removed_key, removed_record)) = removed.as_ref() {
                        self.record_removed(estimated_record_memory_bytes(
                            removed_key,
                            removed_record.deadline_tick != NO_DEADLINE));
                    }
                    statistics.removed_records += 1;
                    match candidate.source {
                        VersionSource::FirstObservation => {
                            statistics.removed_first_observations += 1;
                        },
                        VersionSource::CommittedWrite => {
                            statistics.removed_committed_writes += 1;
                        },
                    }
                } else {
                    statistics.stale_candidates += 1;
                    if let Some(current) = self.current(&key) {
                        if current.deadline_tick > now
                            || current.source == VersionSource::FirstObservation {
                            self.register_deadline(current.deadline_tick);
                        } else {
                            let min_active = self
                                .0
                                .active_snapshots
                                .lock()
                                .keys()
                                .next()
                                .copied();
                            if min_active
                                .map(|revision| current.revision > revision)
                                .unwrap_or(false) {
                                statistics.snapshot_blocked += 1;
                                blocked = true;
                            } else {
                                self.register_deadline(current.deadline_tick);
                            }
                        }
                        retained.push(key);
                    }
                }
            }
            // 不允许在 token 临时离开索引时 await；先归还全部保留 Key，再主动让出 runtime。
            self.0.ttl_keys.push_batch(retained);
            rt.timeout(0).await;
            statistics.yields += 1;
        }

        if blocked {
            self.0.blocked_lease_epoch.store(scan_epoch, Ordering::Release);
            self.0.has_blocked_expiry.store(true, Ordering::Release);
        }

        // 插入方先入队再 fetch_min，已有记录更新也始终 fetch_min；因此 swap 前后的并发写要么
        // 由本轮 token 重新登记，要么在 swap 后直接登记，无需阻塞写入的 DashMap 全表复扫。
    }
}

/// 一个 TTL 开启表在版本缺失查询期间持有的首次观察租约。
///
/// 该 guard 不持有 Mutex、RwLock、DashMap entry 或表数据 guard，也不拥有 Arc；它只能活在被借用
/// 的 `KeyVersions` 内。`active=false` 是 TTL 关闭时的零原子成本路径。类型不可 clone，Drop 是
/// 唯一释放点，因此不存在重复释放或被遗忘的显式完成分支。
pub(crate) struct FirstObservationLease<'a> {
    versions: &'a KeyVersions,
    active: bool,
}

impl Drop for FirstObservationLease<'_> {
    fn drop(&mut self) {
        if !self.active {
            return;
        }
        let mut observed = self
            .versions
            .0
            .active_first_observations
            .load(Ordering::Acquire);
        loop {
            let next = observed
                .checked_sub(1)
                .expect("active first-observation lease count underflow");
            match self.versions.0.active_first_observations.compare_exchange_weak(
                observed,
                next,
                Ordering::AcqRel,
                Ordering::Acquire) {
                Ok(_) => break,
                Err(current) => observed = current,
            }
        }
        // 先让 active 递减对 scanner 可见，再推进代次。即使 scanner 与 Drop 交错，也只会多
        // 保留一个固定轮询周期，不会永久遗失到期 token。
        self.versions.0.lease_epoch.fetch_add(1, Ordering::AcqRel);
    }
}

/// 一个表事务的数据快照 revision 租约。
///
/// lease 不持有 root/cache guard，也不固定表数据本身；COW 数据根由表事务单独拥有。它只阻止
/// TTL 删除 `CommittedWrite.revision` 晚于本 revision 的记录。`released` 使 commit、rollback
/// 与 Drop 的重复清理收敛为一次，避免活跃计数下溢。
pub(crate) struct SnapshotLease {
    versions: KeyVersions,
    revision: u64,
    released: AtomicBool,
}

impl SnapshotLease {
    pub(crate) fn revision(&self) -> u64 {
        self.revision
    }

    /// 幂等释放活跃 revision，并推进 lease_epoch 以唤醒被长事务阻塞的后续 TTL 轮询。
    ///
    /// 方法短暂获取同步 mutex，不 await。先更新活跃集合再推进 epoch，保证 scanner 即使与释放
    /// 交错，也只会至多多等待一个固定轮询间隔，不会永久遗失到期记录。
    pub(crate) fn release(&self) {
        if self.released.swap(true, Ordering::AcqRel) {
            return;
        }
        let mut active = self.versions.0.active_snapshots.lock();
        if let Some(count) = active.get_mut(&self.revision) {
            if *count <= 1 {
                active.remove(&self.revision);
            } else {
                *count -= 1;
            }
        }
        drop(active);
        self.versions.0.lease_epoch.fetch_add(1, Ordering::AcqRel);
    }
}

impl Drop for SnapshotLease {
    fn drop(&mut self) {
        self.release();
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum PrepareMode {
    Ordinary,
    /// 公开建表 API 在根协议选择前登记的内部 Meta 动作。
    ///
    /// 该模式可随之后选择的 Ordinary 或 Versioned 根一起 prepare/commit，但自身不属于业务
    /// 协议，也不产生公开版本回执。它必须使用与 Versioned 相同的严格 Key 冲突规则，防止
    /// 并发 DDL 被 Ordinary DirtyWrite 的放宽语义穿透。
    SchemaCreate,
    Versioned,
}

pub(crate) struct PreparedActions {
    pub(crate) mode: PrepareMode,
    pub(crate) actions: XHashMap<Binary, KVActionLog>,
}

/// 在用四表的表级预提交占用 owner。
///
/// prepare map 和正在提交的子事务只共享一个不可变动作集合，不复制 Key/Value。提交必须保留
/// map 中的 owner，直至数据、版本和 completed revision 全部发布，再按 Arc 身份精确清理。
pub(crate) type SharedPreparedActions = Arc<PreparedActions>;

/// 表级 commit 取得 prepared 项时可能发现的结构不变量错误。
///
/// 该错误只描述同步 `prepare` map 的局部状态，不决定事务错误等级。调用方已经进入 commit，
/// 必须结合整棵事务树可能已有兄弟节点发布这一事实，把这些错误都转换为 Fatal。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum PreparedCommitError {
    Missing,
    ModeMismatch(PrepareMode),
}

/// 已发布数据后精确清理 prepared owner 时的结构不变量错误。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum PreparedCleanupError {
    Missing,
    IdentityMismatch,
}

/// 判断同一表的 prepare map 是否已经登记指定根 TID。
///
/// 调用方必须在同一个 prepare mutex guard 内完成本检查、全部 Key 冲突检查和最终 insert，
/// 才能保证第二个同 TID 子节点不会覆盖第一个节点已经冻结的动作。
#[inline]
pub(crate) fn has_prepared_transaction<P>(prepare: &XHashMap<Guid, P>,
                                       transaction_uid: &Guid) -> bool {
    prepare.contains_key(transaction_uid)
}

/// 在 commit 开始时取得不可变 prepared owner，但保留表级预提交占用。
///
/// 调用方只在现有 prepare mutex 内调用。匹配路径只 clone 一个 Arc；模式错配保持旧行为，
/// 立即移除错误项，因为事务已经进入根 WAL 后不可 rollback 的提交阶段。
#[inline]
pub(crate) fn retain_prepared_for_commit(
    prepare: &mut XHashMap<Guid, SharedPreparedActions>,
    transaction_uid: &Guid,
    expected_mode: PrepareMode,
    is_writable: bool,
) -> Result<Option<SharedPreparedActions>, PreparedCommitError> {
    match prepare.get(transaction_uid) {
        Some(prepared) if prepared.mode == expected_mode => Ok(Some(prepared.clone())),
        Some(prepared) => {
            let prepared_mode = prepared.mode;
            let _ = prepare.remove(transaction_uid);
            Err(PreparedCommitError::ModeMismatch(prepared_mode))
        },
        None if is_writable => Err(PreparedCommitError::Missing),
        None => Ok(None),
    }
}

/// 在数据、版本和 completed revision 发布后精确清理本次提交的预提交占用。
///
/// 只按 TID remove 无法识别非法重复调用是否替换了 owner；Arc 身份检查保证当前提交绝不删除
/// 其它动作。调用方只持有 prepare mutex，不得同时持有 root/cache 或 DashMap guard。
#[inline]
pub(crate) fn remove_retained_prepared(
    prepare: &mut XHashMap<Guid, SharedPreparedActions>,
    transaction_uid: &Guid,
    retained: &SharedPreparedActions,
) -> Result<(), PreparedCleanupError> {
    match prepare.get(transaction_uid) {
        Some(current) if Arc::ptr_eq(current, retained) => {
            let _ = prepare.remove(transaction_uid);
            Ok(())
        },
        Some(_) => Err(PreparedCleanupError::IdentityMismatch),
        None => Err(PreparedCleanupError::Missing),
    }
}

/// 版本化根事务的共享提交回执汇聚器。
///
/// 每个子表在完成数据、版本和 revision 发布并释放表数据锁后，追加本事务最终写集合；根事务
/// 在整棵树 commit 成功后一次性 take。内部 mutex 只保护该根事务私有 Vec，不与表锁形成嵌套，
/// 不执行 await 或 I/O。普通 commit 不安装本对象，因此不承担公开 Vec 的收集成本。
#[derive(Clone)]
pub(crate) struct VersionReceipt(Arc<Mutex<Vec<TableKeyVersion>>>);

impl VersionReceipt {
    pub(crate) fn new() -> Self {
        Self(Arc::new(Mutex::new(Vec::new())))
    }

    pub(crate) fn append(&self, mut versions: Vec<TableKeyVersion>) {
        if versions.is_empty() {
            return;
        }
        let mut receipt = self.0.lock();
        if receipt.is_empty() {
            *receipt = versions;
        } else {
            receipt.append(&mut versions);
        }
    }

    pub(crate) fn take(&self) -> Vec<TableKeyVersion> {
        mem::take(&mut *self.0.lock())
    }

    pub(crate) fn clear(&self) {
        self.0.lock().clear();
    }
}

/// 一个由 `KVDBManager` 装配的表事务所需的版本协议上下文。
///
/// `versions` 绑定精确表实例；`snapshot` 固定事务创建时 revision；Versioned 的 `expected` 是
/// 外部真实读缓存提交的显式版本，SchemaCreate 固定为空；`mode` 决定 prepare 预留兼容矩阵；
/// `receipt` 只在需要返回业务写版本的 Versioned 子事务中存在，内部 schema 固定为 None。
/// 上下文随表事务 Arc 存活，显式终结遗漏时 SnapshotLease 的 Drop 仍负责最终资源释放。
pub(crate) struct TableVersionContext {
    versions: KeyVersions,
    snapshot: SnapshotLease,
    mode: PrepareMode,
    expected: XHashMap<Binary, Version>,
    receipt: Option<VersionReceipt>,
}

impl TableVersionContext {
    pub(crate) fn new(versions: KeyVersions,
                      snapshot: SnapshotLease,
                      mode: PrepareMode,
                      expected: XHashMap<Binary, Version>,
                      receipt: Option<VersionReceipt>) -> Self {
        Self {
            versions,
            snapshot,
            mode,
            expected,
            receipt,
        }
    }

    pub(crate) fn versions(&self) -> &KeyVersions {
        &self.versions
    }

    pub(crate) fn snapshot_revision(&self) -> u64 {
        self.snapshot.revision()
    }

    pub(crate) fn mode(&self) -> PrepareMode {
        self.mode
    }

    pub(crate) fn expected(&self) -> &XHashMap<Binary, Version> {
        &self.expected
    }

    pub(crate) fn receipt(&self) -> Option<&VersionReceipt> {
        self.receipt.as_ref()
    }

    pub(crate) fn release_snapshot(&self) {
        self.snapshot.release();
    }
}

/// 判断两个已登记动作在指定 prepare 模式下是否互斥。
pub(crate) fn prepared_actions_conflict(existing_mode: PrepareMode,
                                         existing: &KVActionLog,
                                         current_mode: PrepareMode,
                                         current: &KVActionLog) -> bool {
    if existing_mode != PrepareMode::Ordinary || current_mode != PrepareMode::Ordinary {
        return !matches!((existing, current), (KVActionLog::Read, KVActionLog::Read));
    }

    match existing {
        KVActionLog::Read => matches!(current, KVActionLog::Write(_)),
        KVActionLog::DirtyWrite(_) => false,
        KVActionLog::Write(_) => !matches!(current, KVActionLog::DirtyWrite(_)),
    }
}

/// 严格比较两个逻辑值状态；Missing 与任何 value 都不相等。
pub(crate) fn binary_state_equal(left: Option<&Binary>, right: Option<&Binary>) -> bool {
    match (left, right) {
        (None, None) => true,
        (Some(left), Some(right)) => Binary::binary_equal(left, right),
        _ => false,
    }
}

/// 判断指定动作是否与任一已登记事务的同 Key 动作冲突。
pub(crate) fn has_prepared_conflict<P>(prepare: &XHashMap<Guid, P>,
                                       key: &Binary,
                                       mode: PrepareMode,
                                       action: &KVActionLog) -> bool
    where P: Borrow<PreparedActions>
{
    prepare.values().any(|prepared| {
        let prepared = prepared.borrow();
        prepared
            .actions
            .get(key)
            .map(|existing| prepared_actions_conflict(prepared.mode,
                                                      existing,
                                                      mode,
                                                      action))
            .unwrap_or(false)
    })
}

#[cfg(test)]
mod tests {
    use std::{collections::BTreeSet,
              io::ErrorKind,
              panic::AssertUnwindSafe,
              sync::{Arc, Barrier, atomic::Ordering},
              thread,
              time::Duration};

    use futures::executor::block_on;
    use pi_async_rt::rt::multi_thread::MultiTaskRuntimeBuilder;
    use pi_atom::Atom;
    use pi_bon::{Encode, WriteBuffer};
    use pi_guid::Guid;

    use crate::{Binary, KVActionLog, TableKeyVersion};

    use super::{KeyVersionConfig,
                KeyVersionRegistry,
                MAX_TICK,
                FirstObservation,
                PrepareMode,
                PreparedActions,
                PreparedCleanupError,
                PreparedCommitError,
                TableKeyConflict,
                TTL_SCAN_BATCH_SIZE,
                TtlKeyIndex,
                VersionReceipt,
                VersionConflictKind,
                deadline_tick,
                duration_to_ticks,
                has_prepared_transaction,
                normalize_conflicts,
                prepared_actions_conflict,
                remove_retained_prepared,
                retain_prepared_for_commit};
    #[cfg(feature = "trace")]
    use super::{KeyVersionApiMetricsSnapshot,
                KeyVersionApiOperation,
                NO_DEADLINE,
                estimated_record_memory_bytes};

    /// 完整冲突集合必须按 Table/Key 排序、去重，并让版本失配覆盖同 Key 的事务冲突。
    #[test]
    fn test_normalize_conflicts_orders_deduplicates_and_merges_kind() {
        let table_a = Atom::from("conflict_a");
        let table_b = Atom::from("conflict_b");
        let key_1 = binary_from_u32(1);
        let key_2 = binary_from_u32(2);
        let normalized = normalize_conflicts(vec![
            TableKeyConflict {
                table: table_b.clone(),
                key: key_2.clone(),
                kind: VersionConflictKind::TransactionConflict,
            },
            TableKeyConflict {
                table: table_a.clone(),
                key: key_2.clone(),
                kind: VersionConflictKind::TransactionConflict,
            },
            TableKeyConflict {
                table: table_a.clone(),
                key: key_1.clone(),
                kind: VersionConflictKind::TransactionConflict,
            },
            TableKeyConflict {
                table: table_a.clone(),
                key: key_2.clone(),
                kind: VersionConflictKind::ReadSetVersionMismatch,
            },
            TableKeyConflict {
                table: table_a.clone(),
                key: key_1.clone(),
                kind: VersionConflictKind::TransactionConflict,
            },
        ]);

        assert_eq!(normalized, vec![
            TableKeyConflict {
                table: table_a.clone(),
                key: key_1,
                kind: VersionConflictKind::TransactionConflict,
            },
            TableKeyConflict {
                table: table_a,
                key: key_2.clone(),
                kind: VersionConflictKind::ReadSetVersionMismatch,
            },
            TableKeyConflict {
                table: table_b,
                key: key_2,
                kind: VersionConflictKind::TransactionConflict,
            },
        ]);
    }

    /// SchemaCreate 和 Versioned 都必须使用严格矩阵；Ordinary 的既有 dirty 写放宽保持不变。
    #[test]
    fn test_schema_create_uses_strict_prepared_conflict_matrix() {
        let read = KVActionLog::Read;
        let write = KVActionLog::Write(Some(binary_from_u32(1)));
        let dirty_write = KVActionLog::DirtyWrite(Some(binary_from_u32(2)));

        for strict_mode in [PrepareMode::SchemaCreate, PrepareMode::Versioned] {
            assert!(!prepared_actions_conflict(strict_mode,
                                               &read,
                                               PrepareMode::Ordinary,
                                               &read));
            assert!(prepared_actions_conflict(strict_mode,
                                              &read,
                                              PrepareMode::Ordinary,
                                              &write));
            assert!(prepared_actions_conflict(PrepareMode::Ordinary,
                                              &dirty_write,
                                              strict_mode,
                                              &write));
            assert!(prepared_actions_conflict(strict_mode,
                                              &write,
                                              PrepareMode::Ordinary,
                                              &dirty_write));
        }

        assert!(!prepared_actions_conflict(PrepareMode::Ordinary,
                                           &dirty_write,
                                           PrepareMode::Ordinary,
                                           &write));
    }

    /// 重复根 TID 检查必须精确区分已登记和未登记项，且不得修改既有冻结动作。
    #[test]
    fn test_prepared_transaction_duplicate_guard_is_non_destructive() {
        let transaction_uid = Guid(11);
        let other_uid = Guid(12);
        let mut prepare = pi_hash::XHashMap::default();
        prepare.insert(transaction_uid.clone(), PreparedActions {
            mode: PrepareMode::Ordinary,
            actions: pi_hash::XHashMap::default(),
        });

        assert!(has_prepared_transaction(&prepare, &transaction_uid));
        assert!(!has_prepared_transaction(&prepare, &other_uid));
        assert_eq!(prepare.len(), 1);
        assert_eq!(prepare.get(&transaction_uid).map(|item| item.mode),
                   Some(PrepareMode::Ordinary));
    }

    /// commit 必须保留匹配 owner 到发布结束，并用 Arc 身份精确清理而不误删替换项。
    #[test]
    fn test_retain_prepared_for_commit_enforces_owner_lifecycle() {
        let ordinary_uid = Guid(21);
        let mismatch_uid = Guid(22);
        let missing_uid = Guid(23);
        let mut prepare = pi_hash::XHashMap::default();
        let mut ordinary_actions = pi_hash::XHashMap::default();
        ordinary_actions.insert(binary_from_u32(1), KVActionLog::Read);
        let ordinary = Arc::new(PreparedActions {
            mode: PrepareMode::Ordinary,
            actions: ordinary_actions,
        });
        prepare.insert(ordinary_uid.clone(), ordinary.clone());
        prepare.insert(mismatch_uid.clone(), Arc::new(PreparedActions {
            mode: PrepareMode::Versioned,
            actions: pi_hash::XHashMap::default(),
        }));

        let retained = match retain_prepared_for_commit(&mut prepare,
                                                        &ordinary_uid,
                                                        PrepareMode::Ordinary,
                                                        true) {
            Ok(Some(prepared)) => {
                assert!(Arc::ptr_eq(&prepared, &ordinary));
                assert_eq!(prepared.mode, PrepareMode::Ordinary);
                assert!(prepared
                    .actions
                    .get(&binary_from_u32(1))
                    .map(|action| matches!(action, KVActionLog::Read))
                    .unwrap_or(false));
                prepared
            },
            _ => panic!("matching writable prepared item must be returned"),
        };
        assert!(prepare.contains_key(&ordinary_uid),
                "retaining commit input must keep the prepare reservation visible");
        assert_eq!(remove_retained_prepared(&mut prepare, &ordinary_uid, &retained), Ok(()));
        assert!(!prepare.contains_key(&ordinary_uid));

        assert!(matches!(retain_prepared_for_commit(&mut prepare,
                                                    &mismatch_uid,
                                                    PrepareMode::Ordinary,
                                                    true),
                         Err(PreparedCommitError::ModeMismatch(PrepareMode::Versioned))));
        assert!(!prepare.contains_key(&mismatch_uid));
        assert!(matches!(retain_prepared_for_commit(&mut prepare,
                                                    &missing_uid,
                                                    PrepareMode::Ordinary,
                                                    true),
                         Err(PreparedCommitError::Missing)));
        assert!(matches!(retain_prepared_for_commit(&mut prepare,
                                                    &missing_uid,
                                                    PrepareMode::Ordinary,
                                                    false),
                         Ok(None)));

        let retained = Arc::new(PreparedActions {
            mode: PrepareMode::Ordinary,
            actions: pi_hash::XHashMap::default(),
        });
        let replacement = Arc::new(PreparedActions {
            mode: PrepareMode::Ordinary,
            actions: pi_hash::XHashMap::default(),
        });
        prepare.insert(ordinary_uid.clone(), retained.clone());
        prepare.insert(ordinary_uid.clone(), replacement.clone());
        assert_eq!(remove_retained_prepared(&mut prepare, &ordinary_uid, &retained),
                   Err(PreparedCleanupError::IdentityMismatch));
        assert!(Arc::ptr_eq(prepare.get(&ordinary_uid).unwrap(), &replacement),
                "identity mismatch must not remove another owner");
        assert_eq!(remove_retained_prepared(&mut prepare, &ordinary_uid, &replacement), Ok(()));
        assert_eq!(remove_retained_prepared(&mut prepare, &ordinary_uid, &replacement),
                   Err(PreparedCleanupError::Missing));
    }

    /// publish 必须在同一个 DashMap entry 临界区中读取并推进 generation，同时精确替换
    /// source、revision 和 Upsert/Delete 类型，不能重复增加结构记录。
    #[test]
    fn test_publish_atomically_advances_current_record() {
        let config = KeyVersionConfig::new(Duration::ZERO, Duration::ZERO).unwrap();
        let (registry, _shutdown) = KeyVersionRegistry::new(config);
        let versions = registry.create_table_versions();
        let table = Atom::from("publish-unit");
        let key = binary_from_u32(31);
        let value = binary_from_u32(310);

        assert_eq!(versions.first_observation(key.clone(), false, || Guid(1)),
                   FirstObservation::Inserted(super::Version::Delete(Guid(1))));
        let first = versions.current(&key).unwrap();
        assert_eq!(first.source, super::VersionSource::FirstObservation);
        assert_eq!(first.revision, 0);
        assert_eq!(first.generation, 1);

        let upsert = versions.publish(table.clone(),
                                      key.clone(),
                                      Some(&value),
                                      Guid(2),
                                      1);
        assert_table_key_version_fields(&upsert, &TableKeyVersion {
            table: table.clone(),
            key: key.clone(),
            version: super::Version::Upsert(Guid(2)),
        });
        let second = versions.current(&key).unwrap();
        assert_eq!(second.version, super::Version::Upsert(Guid(2)));
        assert_eq!(second.source, super::VersionSource::CommittedWrite);
        assert_eq!(second.revision, 1);
        assert_eq!(second.generation, 2);

        let delete = versions.publish(table.clone(), key.clone(), None, Guid(3), 2);
        assert_table_key_version_fields(&delete, &TableKeyVersion {
            table,
            key: key.clone(),
            version: super::Version::Delete(Guid(3)),
        });
        let third = versions.current(&key).unwrap();
        assert_eq!(third.version, super::Version::Delete(Guid(3)));
        assert_eq!(third.source, super::VersionSource::CommittedWrite);
        assert_eq!(third.revision, 2);
        assert_eq!(third.generation, 3);
        assert_eq!(versions.len(), 1);
    }

    /// 局部红线验收：query 第一次确认版本缺失并读到旧的 None 后，提交把数据变为 Some。
    /// scanner 在租约活跃时必须保留新版本，使第二次版本检查观察到 Upsert；否则旧的 None
    /// 会错误生成新的 Delete 首次观察版本。本测试只固定内部交错，完整生产可达性由独立真实
    /// TTL target 验证。
    #[test]
    fn test_first_observation_lease_closes_stale_value_redline() {
        let config = KeyVersionConfig::new(Duration::from_millis(1),
                                           Duration::from_millis(1)).unwrap();
        let (registry, _shutdown) = KeyVersionRegistry::new(config);
        let versions = registry.create_table_versions();
        let table = Atom::from("first-observation-redline");
        let key = binary_from_u32(32);
        let committed_value = binary_from_u32(320);

        let stale_value = None::<Binary>;
        let lease = versions.lease_first_observation();
        let _ = versions.publish(table,
                                 key.clone(),
                                 Some(&committed_value),
                                 Guid(2),
                                 1);
        let rt = MultiTaskRuntimeBuilder::default()
            .init_worker_size(1)
            .build();
        let mut blocked_statistics = super::TtlCollectStatistics::default();
        block_on(versions.collect_expired(&rt,
                                          MAX_TICK,
                                          &mut blocked_statistics));

        assert_eq!(versions.current_version(&key),
                   Some(super::Version::Upsert(Guid(2))),
                   "TTL must retain a committed version while a missing query owns a lease");
        assert_eq!(blocked_statistics.due_candidates, 1);
        assert_eq!(blocked_statistics.first_observation_blocked, 1);
        assert_eq!(blocked_statistics.removed_records, 0);
        assert_eq!(index_len(&versions), 1,
                   "a lease-blocked candidate must retain its TTL token");
        let observed = versions.first_observation(key, stale_value.is_some(), || Guid(3));
        assert_eq!(observed,
                   FirstObservation::Occupied(super::Version::Upsert(Guid(2))));
        drop(lease);
        assert!(!versions.has_active_first_observation());

        let mut released_statistics = super::TtlCollectStatistics::default();
        block_on(versions.collect_expired(&rt,
                                          MAX_TICK,
                                          &mut released_statistics));
        assert_eq!(released_statistics.removed_records, 1);
        assert_eq!(released_statistics.removed_committed_writes, 1);
        assert!(versions.current_version(&binary_from_u32(32)).is_none());
        assert_eq!(index_len(&versions), 0,
                   "the last lease release must make the retained token collectible");
    }

    /// TTL 关闭时租约必须是无操作：首次缺失查询不写活跃计数，也不推进 TTL 唤醒代次。
    #[test]
    fn test_first_observation_lease_is_noop_when_ttl_disabled() {
        let config = KeyVersionConfig::new(Duration::ZERO, Duration::ZERO).unwrap();
        let (registry, _shutdown) = KeyVersionRegistry::new(config);
        let versions = registry.create_table_versions();
        let epoch = versions.0.lease_epoch.load(Ordering::Acquire);

        let lease = versions.lease_first_observation();
        assert_eq!(versions.0.active_first_observations.load(Ordering::Acquire), 0);
        assert!(!versions.has_active_first_observation());
        drop(lease);

        assert_eq!(versions.0.active_first_observations.load(Ordering::Acquire), 0);
        assert_eq!(versions.0.lease_epoch.load(Ordering::Acquire), epoch);
    }

    /// TTL 开启时嵌套租约必须逐个计数；每个不可 clone guard 的 Drop 只释放自己的一项。
    #[test]
    fn test_first_observation_lease_balances_nested_guards() {
        let config = KeyVersionConfig::new(Duration::from_millis(1),
                                           Duration::from_millis(1)).unwrap();
        let (registry, _shutdown) = KeyVersionRegistry::new(config);
        let versions = registry.create_table_versions();

        let first = versions.lease_first_observation();
        let second = versions.lease_first_observation();
        assert_eq!(versions.0.active_first_observations.load(Ordering::Acquire), 2);
        assert!(versions.has_active_first_observation());

        drop(first);
        assert_eq!(versions.0.active_first_observations.load(Ordering::Acquire), 1);
        assert_eq!(versions.0.lease_epoch.load(Ordering::Acquire), 1);
        drop(second);
        assert_eq!(versions.0.active_first_observations.load(Ordering::Acquire), 0);
        assert_eq!(versions.0.lease_epoch.load(Ordering::Acquire), 2);
    }

    /// 多线程同时持有的租约必须全部可见并最终严格配平；该计数不能依赖 runtime owner 线程。
    #[test]
    fn test_first_observation_lease_balances_cross_thread_guards() {
        const THREADS: usize = 8;

        let config = KeyVersionConfig::new(Duration::from_millis(1),
                                           Duration::from_millis(1)).unwrap();
        let (registry, _shutdown) = KeyVersionRegistry::new(config);
        let versions = Arc::new(registry.create_table_versions());
        let acquired = Arc::new(Barrier::new(THREADS + 1));
        let release = Arc::new(Barrier::new(THREADS + 1));
        let mut threads = Vec::with_capacity(THREADS);
        for _ in 0..THREADS {
            let versions = versions.clone();
            let acquired = acquired.clone();
            let release = release.clone();
            threads.push(thread::spawn(move || {
                let lease = versions.lease_first_observation();
                acquired.wait();
                release.wait();
                drop(lease);
            }));
        }

        acquired.wait();
        assert_eq!(versions.0.active_first_observations.load(Ordering::Acquire), THREADS);
        release.wait();
        for thread in threads {
            thread.join().expect("first-observation lease worker must not panic");
        }
        assert_eq!(versions.0.active_first_observations.load(Ordering::Acquire), 0);
        assert_eq!(versions.0.lease_epoch.load(Ordering::Acquire), THREADS as u64);
    }

    /// unwind 必须执行 RAII Drop；否则一个 panic 就会让同表到期版本永久无法回收。
    #[test]
    fn test_first_observation_lease_releases_during_unwind() {
        let config = KeyVersionConfig::new(Duration::from_millis(1),
                                           Duration::from_millis(1)).unwrap();
        let (registry, _shutdown) = KeyVersionRegistry::new(config);
        let versions = registry.create_table_versions();

        let unwind = std::panic::catch_unwind(AssertUnwindSafe(|| {
            let _lease = versions.lease_first_observation();
            panic!("intentional first-observation lease unwind");
        }));
        assert!(unwind.is_err());
        assert_eq!(versions.0.active_first_observations.load(Ordering::Acquire), 0);
        assert_eq!(versions.0.lease_epoch.load(Ordering::Acquire), 1);
    }

    /// 被破坏的最大计数必须在增加前 fail-fast，不能回绕成 0 让 scanner 错误删除版本。
    #[test]
    fn test_first_observation_lease_count_never_wraps() {
        let config = KeyVersionConfig::new(Duration::from_millis(1),
                                           Duration::from_millis(1)).unwrap();
        let (registry, _shutdown) = KeyVersionRegistry::new(config);
        let versions = registry.create_table_versions();
        versions.0.active_first_observations.store(usize::MAX, Ordering::Release);

        let overflow = std::panic::catch_unwind(AssertUnwindSafe(|| {
            let _lease = versions.lease_first_observation();
        }));
        assert!(overflow.is_err());
        assert_eq!(versions.0.active_first_observations.load(Ordering::Acquire), usize::MAX);
        versions.0.active_first_observations.store(0, Ordering::Release);
    }

    /// 首批回执应直接接管调用方 Vec 的 allocation；空批无副作用，后续批次保持表项顺序，
    /// take 必须一次性转移全部元素并清空共享汇聚器。
    #[test]
    fn test_version_receipt_moves_first_batch_and_appends_in_order() {
        let receipt = VersionReceipt::new();
        let first_item = TableKeyVersion {
            table: Atom::from("receipt-a"),
            key: binary_from_u32(41),
            version: super::Version::Upsert(Guid(4)),
        };
        let second_item = TableKeyVersion {
            table: Atom::from("receipt-b"),
            key: binary_from_u32(42),
            version: super::Version::Delete(Guid(5)),
        };
        let third_item = TableKeyVersion {
            table: Atom::from("receipt-c"),
            key: binary_from_u32(43),
            version: super::Version::Upsert(Guid(6)),
        };
        let first_batch = vec![first_item.clone(), second_item.clone()];
        let first_ptr = first_batch.as_ptr();
        let first_capacity = first_batch.capacity();

        receipt.append(first_batch);
        {
            let stored = receipt.0.lock();
            assert_eq!(stored.as_ptr(), first_ptr,
                       "empty receipt must take ownership of the first Vec allocation");
            assert_eq!(stored.capacity(), first_capacity);
            assert_eq!(stored.len(), 2);
            assert_table_key_version_fields(&stored[0], &first_item);
            assert_table_key_version_fields(&stored[1], &second_item);
        }

        receipt.append(Vec::new());
        {
            let stored = receipt.0.lock();
            assert_eq!(stored.as_ptr(), first_ptr,
                       "an empty append must not replace the current allocation");
            assert_eq!(stored.len(), 2);
            assert_table_key_version_fields(&stored[0], &first_item);
            assert_table_key_version_fields(&stored[1], &second_item);
        }

        receipt.append(vec![third_item.clone()]);
        let taken = receipt.take();
        assert_eq!(taken.len(), 3);
        assert_table_key_version_fields(&taken[0], &first_item);
        assert_table_key_version_fields(&taken[1], &second_item);
        assert_table_key_version_fields(&taken[2], &third_item);
        assert!(receipt.take().is_empty());
    }

    /// 只有开启 TTL 时 ZERO 轮询间隔才非法；关闭 TTL 不得创建 task owner 通道。
    ///
    /// 真实公开 startup 的错误种类、目录、manager 和 WAL 零副作用由独立
    /// `manager_startup_configuration` target 验证，本单元只隔离配置到 registry 的直接映射。
    #[test]
    fn test_key_version_config_zero_poll_interval_depends_on_ttl_enablement() {
        let error = KeyVersionConfig::new(Duration::from_secs(1), Duration::ZERO)
            .expect_err("enabled TTL with zero poll interval must be rejected");
        assert_eq!(error.kind(), ErrorKind::InvalidInput);

        let disabled = KeyVersionConfig::new(Duration::ZERO, Duration::ZERO)
            .expect("disabled TTL must ignore a zero poll interval");
        assert_eq!(disabled.ttl_ticks(), None);
        assert_eq!(disabled.poll_interval_ticks(), 0);
        let (disabled_registry, disabled_receiver) = KeyVersionRegistry::new(disabled);
        assert!(disabled_receiver.is_none());
        assert!(disabled_registry.0.shutdown_tx.is_none());

        let enabled = KeyVersionConfig::new(Duration::from_secs(1), Duration::from_millis(1))
            .expect("enabled TTL with a positive poll interval must be accepted");
        assert_eq!(enabled.ttl_ticks(), Some(1000));
        assert_eq!(enabled.poll_interval_ticks(), 1);
        let (enabled_registry, enabled_receiver) = KeyVersionRegistry::new(enabled);
        assert!(enabled_receiver.is_some());
        assert!(enabled_registry.0.shutdown_tx.is_some());
    }

    /// TTL 使用 1ms 最小单位；非零 sub-ms 值提升到最小单位，其余小数直接忽略。
    ///
    /// 该测试固定配置量化契约，不是 `BUG-KV-TTL-001` 的红证据；提前淘汰由独立真实 target
    /// 按量化后的有效 TTL 验证。
    #[test]
    fn test_duration_to_ticks_uses_one_millisecond_granularity() {
        assert_eq!(duration_to_ticks(Duration::from_nanos(1)), 1);
        assert_eq!(duration_to_ticks(Duration::from_micros(999)), 1);
        assert_eq!(duration_to_ticks(Duration::from_millis(1)), 1);
        assert_eq!(duration_to_ticks(Duration::from_micros(1_999)), 1);
        assert_eq!(duration_to_ticks(Duration::from_millis(2)), 2);
    }

    /// deadline 基准只在存在 sub-ms 余数时向上取整，并对极大时间和 TTL 饱和。
    #[test]
    fn test_deadline_tick_never_shortens_effective_ttl() {
        assert_eq!(deadline_tick(Duration::from_millis(10), 20), 30);
        assert_eq!(deadline_tick(
            Duration::from_millis(10) + Duration::from_nanos(1), 20), 31);
        assert_eq!(deadline_tick(Duration::from_micros(10_999), 20), 31);
        assert_eq!(deadline_tick(Duration::from_millis(MAX_TICK - 1), 20), MAX_TICK);
        assert_eq!(deadline_tick(Duration::from_millis(MAX_TICK), MAX_TICK), MAX_TICK);
    }

    /// scanner 固定轮次起始 token 数后，本轮重入队和并发新增 token 必须留给下一轮。
    #[test]
    fn test_ttl_key_index_round_snapshot_defers_requeued_and_new_tokens() {
        let index = TtlKeyIndex::new();
        index.push(binary_from_u32(1));
        index.push(binary_from_u32(2));
        index.push(binary_from_u32(3));

        let mut remaining = index.len();
        let first = index.take_batch(2);
        remaining -= first.len();
        assert_eq!(binary_values(&first), vec![1, 2]);

        index.push(first[0].clone());
        index.push(binary_from_u32(4));
        let rest_of_round = index.take_batch(remaining);
        assert_eq!(binary_values(&rest_of_round), vec![3]);

        let deferred = index.take_batch(index.len());
        assert_eq!(binary_values(&deferred), vec![1, 4]);
        assert_eq!(index.len(), 0);
    }

    /// 每次取出至多一个 scanner 批次，不会因大表 token 数改变固定上限。
    #[test]
    fn test_ttl_key_index_honors_batch_limit() {
        let index = TtlKeyIndex::new();
        for value in 0..(TTL_SCAN_BATCH_SIZE + 44) {
            index.push(binary_from_u32(value as u32));
        }

        let first = index.take_batch(TTL_SCAN_BATCH_SIZE);
        assert_eq!(first.len(), TTL_SCAN_BATCH_SIZE);
        assert_eq!(index.len(), 44);
        let second = index.take_batch(TTL_SCAN_BATCH_SIZE);
        assert_eq!(second.len(), 44);
        assert_eq!(index.len(), 0);
    }

    /// 多线程生产者只能增加各自的唯一 token；单 scanner 必须无丢失、无重复地全部取得。
    #[test]
    fn test_ttl_key_index_accepts_concurrent_unique_producers() {
        const PRODUCERS: usize = 4;
        const TOKENS_PER_PRODUCER: usize = 128;

        let index = Arc::new(TtlKeyIndex::new());
        thread::scope(|scope| {
            for producer in 0..PRODUCERS {
                let index = index.clone();
                scope.spawn(move || {
                    for offset in 0..TOKENS_PER_PRODUCER {
                        let value = producer * TOKENS_PER_PRODUCER + offset;
                        index.push(binary_from_u32(value as u32));
                    }
                });
            }
        });

        let tokens = index.take_batch(index.len());
        assert_eq!(tokens.len(), PRODUCERS * TOKENS_PER_PRODUCER);
        let values: BTreeSet<u32> = binary_values(&tokens).into_iter().collect();
        assert_eq!(values.len(), PRODUCERS * TOKENS_PER_PRODUCER);
        assert_eq!(values.first(), Some(&0));
        assert_eq!(values.last(), Some(&((PRODUCERS * TOKENS_PER_PRODUCER - 1) as u32)));
        assert_eq!(index.len(), 0);
    }

    /// 首次观察创建唯一 token，cache hit 和已有记录更新不重复入队；TTL 精确删除后重新建立
    /// 记录时必须创建新 token。
    #[test]
    fn test_key_versions_maintains_one_token_per_current_record() {
        let config = KeyVersionConfig::new(Duration::from_secs(1),
                                           Duration::from_millis(10)).unwrap();
        let (registry, _shutdown) = KeyVersionRegistry::new(config);
        let versions = registry.create_table_versions();
        let key = encode_usize(7);
        let value = binary_from_u32(70);

        let first = versions.first_observation(key.clone(), false, || Guid(1));
        assert_eq!(first, FirstObservation::Inserted(super::Version::Delete(Guid(1))));
        assert_eq!(index_len(&versions), 1);
        let cached = versions.first_observation(key.clone(), false, || {
            panic!("cache hit must not allocate another first-observation Guid")
        });
        assert_eq!(cached, FirstObservation::Occupied(super::Version::Delete(Guid(1))));
        assert_eq!(index_len(&versions), 1);

        let _ = versions.publish(Atom::from("ttl-unit"),
                                 key.clone(),
                                 Some(&value),
                                 Guid(2),
                                 1);
        assert_eq!(index_len(&versions), 1);

        let token = versions.0.ttl_keys.take_batch(1);
        assert_eq!(token.len(), 1);
        assert_eq!(token[0].as_ref(), key.as_ref());
        let candidate = versions.current(&key).unwrap();
        assert!(versions
            .0
            .versions
            .remove_if(&key, |_key, current| current.exact_eq(&candidate))
            .is_some());
        assert_eq!(index_len(&versions), 0);

        let _ = versions.publish(Atom::from("ttl-unit"),
                                 key.clone(),
                                 None,
                                 Guid(3),
                                 2);
        assert_eq!(index_len(&versions), 1);
    }

    /// 清空索引必须释放 token 持有的 Binary payload owner，不能把表或 Key 生命周期永久延长。
    #[test]
    fn test_ttl_key_index_clear_releases_binary_owner() {
        let index = TtlKeyIndex::new();
        let key = binary_from_u32(99);
        let payload = Arc::downgrade(&key.0);
        index.push(key.clone());
        drop(key);
        assert!(payload.upgrade().is_some());

        index.clear();
        assert!(payload.upgrade().is_none());
        assert_eq!(index.len(), 0);
    }

    /// registry 精确移除表后，`KeyVersions` 与 FIFO/Map 共同持有的 Key payload 必须随最后
    /// 一个表 owner 一起释放；内部只有指向 registry 的 Weak，不得形成引用环。
    #[test]
    fn test_key_versions_drop_releases_table_and_binary_owners() {
        let config = KeyVersionConfig::new(Duration::from_secs(1),
                                           Duration::from_millis(10)).unwrap();
        let (registry, _shutdown) = KeyVersionRegistry::new(config);
        let versions = registry.create_table_versions();
        let versions_owner = Arc::downgrade(&versions.0);
        let table = Atom::from("ttl-drop-unit");
        let key = encode_usize(101);
        let payload = Arc::downgrade(&key.0);

        let _ = versions.first_observation(key.clone(), false, || Guid(1));
        assert_eq!(index_len(&versions), 1);
        registry.install(table.clone(), versions.clone());
        drop(key);
        assert!(versions_owner.upgrade().is_some());
        assert!(payload.upgrade().is_some());

        registry.remove_exact(&table, &versions);
        assert!(registry.0.tables.get(&table).is_none());
        drop(versions);

        assert!(versions_owner.upgrade().is_none());
        assert!(payload.upgrade().is_none());
    }

    /// 显式 commit/rollback 清理与 Drop 后备清理必须共享一次性门，不能重复递减活跃快照。
    #[test]
    fn test_snapshot_lease_release_is_idempotent() {
        let config = KeyVersionConfig::new(Duration::ZERO, Duration::ZERO).unwrap();
        let (registry, _shutdown) = KeyVersionRegistry::new(config);
        let versions = registry.create_table_versions();
        let lease = versions.lease_current();

        assert_eq!(lease.revision(), 0);
        assert_eq!(versions.0.active_snapshots.lock().get(&0), Some(&1));
        lease.release();
        assert!(versions.0.active_snapshots.lock().is_empty());
        assert_eq!(versions.0.lease_epoch.load(Ordering::Acquire), 1);

        lease.release();
        drop(lease);
        assert!(versions.0.active_snapshots.lock().is_empty());
        assert_eq!(versions.0.lease_epoch.load(Ordering::Acquire), 1);
    }

    /// revision 空间耗尽必须显式返回 None，不能回绕为 0 并破坏 ABA 判定。
    #[test]
    fn test_next_revision_never_wraps() {
        let config = KeyVersionConfig::new(Duration::ZERO, Duration::ZERO).unwrap();
        let (registry, _shutdown) = KeyVersionRegistry::new(config);
        let versions = registry.create_table_versions();

        versions.0.completed_revision.store(u64::MAX - 1, Ordering::Release);
        assert_eq!(versions.checked_next_revision(), Some(u64::MAX));
        versions.complete_revision(u64::MAX);
        assert_eq!(versions.completed_revision(), u64::MAX);
        assert_eq!(versions.checked_next_revision(), None);
    }

    /// 估值必须使用 payload capacity 而不是 len，并且 TTL 只增加一个 FIFO owner/slot。
    #[cfg(feature = "trace")]
    #[test]
    fn test_trace_record_memory_estimate_uses_capacity_and_ttl_owner() {
        let small = binary_with_capacity(1, 32);
        let large = binary_with_capacity(1, 256);
        let small_without_ttl = estimated_record_memory_bytes(&small, false);
        let large_without_ttl = estimated_record_memory_bytes(&large, false);
        let large_with_ttl = estimated_record_memory_bytes(&large, true);

        assert_eq!(large_without_ttl - small_without_ttl,
                   (large.0.capacity() - small.0.capacity()) as u64);
        assert_eq!(large_with_ttl - large_without_ttl,
                   std::mem::size_of::<(Binary, std::sync::atomic::AtomicUsize)>() as u64);
    }

    /// 只有 Map 结构新增/删除改变容量指标；命中和已有 Key 的版本替换不得重复累计。
    #[cfg(feature = "trace")]
    #[test]
    fn test_trace_cache_metrics_track_insert_replace_remove_and_clear() {
        let config = KeyVersionConfig::new(Duration::from_secs(1),
                                           Duration::from_millis(10)).unwrap();
        let (registry, _shutdown) = KeyVersionRegistry::new(config);
        let versions = registry.create_table_versions();
        let first_key = binary_with_capacity(11, 64);
        let second_key = binary_with_capacity(22, 128);
        let value = binary_from_u32(99);
        let first_bytes = estimated_record_memory_bytes(&first_key, true);
        let second_bytes = estimated_record_memory_bytes(&second_key, true);

        assert_eq!(versions.metrics_snapshot(), Default::default());
        let first_version = versions.first_observation(first_key.clone(), false, || Guid(1));
        assert_eq!(first_version,
                   FirstObservation::Inserted(super::Version::Delete(Guid(1))));
        assert_eq!(versions.metrics_snapshot(), super::KeyVersionCacheMetricsSnapshot {
            record_count: 1,
            estimated_memory_bytes: first_bytes,
        });
        assert_eq!(versions.first_observation(first_key.clone(), false, || {
            panic!("cache hit must not allocate a new Guid")
        }), FirstObservation::Occupied(super::Version::Delete(Guid(1))));
        assert_eq!(versions.metrics_snapshot().record_count, 1);

        let _ = versions.publish(Atom::from("trace-metrics"),
                                 first_key.clone(),
                                 Some(&value),
                                 Guid(2),
                                 1);
        assert_eq!(versions.metrics_snapshot(), super::KeyVersionCacheMetricsSnapshot {
            record_count: 1,
            estimated_memory_bytes: first_bytes,
        });
        let _ = versions.publish(Atom::from("trace-metrics"),
                                 second_key.clone(),
                                 None,
                                 Guid(3),
                                 2);
        assert_eq!(versions.metrics_snapshot(), super::KeyVersionCacheMetricsSnapshot {
            record_count: 2,
            estimated_memory_bytes: first_bytes + second_bytes,
        });

        let candidate = versions.current(&first_key).unwrap();
        let removed = versions
            .0
            .versions
            .remove_if(&first_key, |_key, current| current.exact_eq(&candidate))
            .expect("exact current record must be removable");
        versions.record_removed(estimated_record_memory_bytes(
            &removed.0,
            removed.1.deadline_tick != NO_DEADLINE));
        assert_eq!(versions.metrics_snapshot(), super::KeyVersionCacheMetricsSnapshot {
            record_count: 1,
            estimated_memory_bytes: second_bytes,
        });

        versions.clear_records();
        assert_eq!(versions.metrics_snapshot(), Default::default());
        assert_eq!(versions.len(), 0);
        assert_eq!(index_len(&versions), 0);
    }

    /// 并发首次观察的 entry 线性化必须让同 Key 只计一次，不同 Key 全部精确计入。
    #[cfg(feature = "trace")]
    #[test]
    fn test_trace_cache_metrics_match_quiescent_map_after_concurrent_insertions() {
        const THREADS: usize = 4;
        const KEYS_PER_THREAD: usize = 128;

        let config = KeyVersionConfig::new(Duration::ZERO, Duration::ZERO).unwrap();
        let (registry, _shutdown) = KeyVersionRegistry::new(config);
        let versions = registry.create_table_versions();
        let shared_key = binary_with_capacity(7, 96);
        thread::scope(|scope| {
            for worker in 0..THREADS {
                let versions = versions.clone();
                let shared_key = shared_key.clone();
                scope.spawn(move || {
                    let _ = versions.first_observation(shared_key, false, || {
                        Guid((worker + 1) as u128)
                    });
                    for offset in 0..KEYS_PER_THREAD {
                        let value = 1_000 + worker * KEYS_PER_THREAD + offset;
                        let key = encode_usize(value);
                        let _ = versions.first_observation(key, false, || {
                            Guid((10_000 + value) as u128)
                        });
                    }
                });
            }
        });

        let expected_count = 1 + THREADS * KEYS_PER_THREAD;
        let expected_memory = versions
            .0
            .versions
            .iter()
            .map(|entry| estimated_record_memory_bytes(
                entry.key(),
                entry.value().deadline_tick != NO_DEADLINE))
            .sum::<u64>();
        assert_eq!(versions.len(), expected_count);
        assert_eq!(versions.metrics_snapshot(), super::KeyVersionCacheMetricsSnapshot {
            record_count: expected_count as u64,
            estimated_memory_bytes: expected_memory,
        });
    }

    /// 每次合法调用只能归入一个 outcome；未 finish 的 guard 模拟 future 取消并计为失败。
    #[cfg(feature = "trace")]
    #[test]
    fn test_trace_api_call_guard_records_one_terminal_outcome() {
        let config = KeyVersionConfig::new(Duration::ZERO, Duration::ZERO).unwrap();
        let (registry, _shutdown) = KeyVersionRegistry::new(config);

        registry.begin_api_call(KeyVersionApiOperation::Query).finish(true);
        registry.begin_api_call(KeyVersionApiOperation::Query).finish(false);
        drop(registry.begin_api_call(KeyVersionApiOperation::Query));
        registry.begin_api_call(KeyVersionApiOperation::Prepare).finish(true);
        drop(registry.begin_api_call(KeyVersionApiOperation::Prepare));
        registry.begin_api_call(KeyVersionApiOperation::Commit).finish(false);

        assert_eq!(registry.api_metrics_snapshot(), KeyVersionApiMetricsSnapshot {
            query_success: 1,
            query_failure: 2,
            prepare_success: 1,
            prepare_failure: 1,
            commit_success: 0,
            commit_failure: 1,
        });
    }

    /// tracing loop 只上报累计快照差值；单次 u64 回绕必须按模运算得到正确 delta。
    #[cfg(feature = "trace")]
    #[test]
    fn test_trace_api_metrics_delta_handles_counter_wrap() {
        let previous = KeyVersionApiMetricsSnapshot {
            query_success: u64::MAX,
            query_failure: 7,
            prepare_success: 8,
            prepare_failure: 9,
            commit_success: 10,
            commit_failure: 11,
        };
        let current = KeyVersionApiMetricsSnapshot {
            query_success: 1,
            query_failure: 10,
            prepare_success: 12,
            prepare_failure: 14,
            commit_success: 16,
            commit_failure: 18,
        };
        assert_eq!(current.delta_since(previous), KeyVersionApiMetricsSnapshot {
            query_success: 2,
            query_failure: 3,
            prepare_success: 4,
            prepare_failure: 5,
            commit_success: 6,
            commit_failure: 7,
        });
    }

    fn index_len(versions: &super::KeyVersions) -> usize {
        versions.0.ttl_keys.len()
    }

    fn assert_table_key_version_fields(actual: &TableKeyVersion,
                                       expected: &TableKeyVersion) {
        assert_eq!(actual.table, expected.table);
        assert_eq!(actual.key.as_ref(), expected.key.as_ref());
        assert_eq!(actual.version, expected.version);
    }

    fn binary_from_u32(value: u32) -> Binary {
        Binary::new(value.to_le_bytes().to_vec())
    }

    #[cfg(feature = "trace")]
    fn binary_with_capacity(value: u32, capacity: usize) -> Binary {
        let mut encoded = WriteBuffer::new();
        (value as usize).encode(&mut encoded);
        let mut bytes = Vec::with_capacity(capacity.max(encoded.len()));
        bytes.extend_from_slice(&encoded.bytes);
        Binary::new(bytes)
    }

    fn encode_usize(value: usize) -> Binary {
        let mut buffer = WriteBuffer::new();
        value.encode(&mut buffer);
        Binary::new(buffer.bytes)
    }

    fn binary_values(tokens: &[Binary]) -> Vec<u32> {
        tokens
            .iter()
            .map(|token| u32::from_le_bytes(token.as_ref().try_into().unwrap()))
            .collect()
    }
}