sochdb_core/

version_chain.rs

// SPDX-License-Identifier: AGPL-3.0-or-later
// SochDB - LLM-Optimized Embedded Database
// Copyright (C) 2026 Sushanth Reddy Vanagala (https://github.com/sushanthpy)
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <https://www.gnu.org/licenses/>.

//! Unified MVCC Version Chain Interface
//!
//! This module defines the canonical interface for MVCC version chains across SochDB.
//! Multiple implementations exist for different subsystems, but they share these traits.
//!
//! ## Implementations
//!
//! | Implementation | Location | Use Case |
//! |---------------|----------|----------|
//! | `VersionChain` | `sochdb_core::epoch_gc` | Epoch-based GC with VecDeque |
//! | `VersionChain` | `sochdb_storage::mvcc_snapshot` | Snapshot-based visibility |
//! | `VersionChain` | `sochdb_storage::version_store` | Generic key-value MVCC |
//! | `VersionChain` | `sochdb_storage::durable_storage` | Binary-search optimized |
//!
//! ## Visibility Semantics
//!
//! All implementations follow these MVCC visibility rules:
//!
//! 1. **Read Committed**: A version is visible if its creating transaction has committed
//!    before the reader's start timestamp.
//!
//! 2. **Snapshot Isolation**: A version is visible if:
//!    - It was committed before the reader's snapshot timestamp
//!    - It was not deleted, or deleted after the snapshot timestamp
//!
//! 3. **Serializable (SSI)**: Adds read-write conflict detection on top of SI.

/// Transaction identifier type
pub type TxnId = u64;

/// Logical timestamp type
pub type Timestamp = u64;

/// Version visibility context
///
/// Provides the information needed to determine if a version is visible
/// to a particular reader/transaction.
#[derive(Debug, Clone)]
pub struct VisibilityContext {
    /// Reader's transaction ID
    pub reader_txn_id: TxnId,
    /// Reader's snapshot timestamp
    pub snapshot_ts: Timestamp,
    /// Set of transaction IDs that are still active (not committed)
    pub active_txn_ids: std::collections::HashSet<TxnId>,
}

impl VisibilityContext {
    /// Create a new visibility context
    pub fn new(reader_txn_id: TxnId, snapshot_ts: Timestamp) -> Self {
        Self {
            reader_txn_id,
            snapshot_ts,
            active_txn_ids: std::collections::HashSet::new(),
        }
    }

    /// Create with active transaction set
    pub fn with_active_txns(
        reader_txn_id: TxnId,
        snapshot_ts: Timestamp,
        active_txn_ids: std::collections::HashSet<TxnId>,
    ) -> Self {
        Self {
            reader_txn_id,
            snapshot_ts,
            active_txn_ids,
        }
    }

    /// Check if a transaction was committed before this snapshot
    pub fn is_committed_before(&self, txn_id: TxnId, commit_ts: Option<Timestamp>) -> bool {
        match commit_ts {
            Some(ts) => ts < self.snapshot_ts && !self.active_txn_ids.contains(&txn_id),
            None => false,
        }
    }
}

/// Version metadata
///
/// Common metadata for all version chain implementations.
#[derive(Debug, Clone)]
pub struct VersionMeta {
    /// Transaction that created this version
    pub created_by: TxnId,
    /// Timestamp when this version was created
    pub created_ts: Timestamp,
    /// Transaction that deleted this version (0 = not deleted)
    pub deleted_by: TxnId,
    /// Timestamp when this version was deleted (0 = not deleted)
    pub deleted_ts: Timestamp,
    /// Commit timestamp (0 = not yet committed)
    pub commit_ts: Timestamp,
}

impl VersionMeta {
    /// Create metadata for a new uncommitted version
    pub fn new_uncommitted(created_by: TxnId, created_ts: Timestamp) -> Self {
        Self {
            created_by,
            created_ts,
            deleted_by: 0,
            deleted_ts: 0,
            commit_ts: 0,
        }
    }

    /// Check if this version is visible according to the context
    pub fn is_visible(&self, ctx: &VisibilityContext) -> bool {
        // Must be committed before snapshot
        if self.commit_ts == 0 {
            // Uncommitted - only visible to creating transaction
            return self.created_by == ctx.reader_txn_id;
        }

        if self.commit_ts >= ctx.snapshot_ts {
            return false;
        }

        // Must not be deleted, or deleted after snapshot
        if self.deleted_by != 0 && self.deleted_ts < ctx.snapshot_ts {
            return false;
        }

        true
    }

    /// Mark as committed
    pub fn commit(&mut self, commit_ts: Timestamp) {
        self.commit_ts = commit_ts;
    }

    /// Mark as deleted
    pub fn delete(&mut self, deleted_by: TxnId, deleted_ts: Timestamp) {
        self.deleted_by = deleted_by;
        self.deleted_ts = deleted_ts;
    }

    /// Check if version is committed
    pub fn is_committed(&self) -> bool {
        self.commit_ts != 0
    }

    /// Check if version is deleted
    pub fn is_deleted(&self) -> bool {
        self.deleted_by != 0
    }
}

/// Trait for MVCC version chain implementations
///
/// Implementors store multiple versions of a value and provide
/// visibility-based access according to MVCC semantics.
pub trait MvccVersionChain {
    /// The value type stored in versions
    type Value;

    /// Get the visible version for the given context
    fn get_visible(&self, ctx: &VisibilityContext) -> Option<&Self::Value>;

    /// Get the latest version (regardless of visibility)
    fn get_latest(&self) -> Option<&Self::Value>;

    /// Number of versions in the chain
    fn version_count(&self) -> usize;

    /// Check if the chain is empty
    fn is_empty(&self) -> bool {
        self.version_count() == 0
    }
}

/// Trait for mutable version chain operations
pub trait MvccVersionChainMut: MvccVersionChain {
    /// Add a new uncommitted version
    fn add_uncommitted(&mut self, value: Self::Value, txn_id: TxnId);

    /// Commit a version
    fn commit_version(&mut self, txn_id: TxnId, commit_ts: Timestamp) -> bool;

    /// Mark the latest visible version as deleted
    fn delete_version(&mut self, txn_id: TxnId, delete_ts: Timestamp) -> bool;

    /// Garbage collect versions older than the given timestamp
    /// Returns (versions_removed, bytes_freed)
    fn gc(&mut self, min_visible_ts: Timestamp) -> (usize, usize);
}

/// Trait for detecting write conflicts
pub trait WriteConflictDetection {
    /// Check if there's a write-write conflict with another transaction
    fn has_write_conflict(&self, txn_id: TxnId) -> bool;
}

// =============================================================================
// Rec 6: Compile-Time Concurrency Policy Markers
// =============================================================================

/// Marker trait for version chain concurrency strategy.
///
/// Implementors tag their version chain with the concurrency mechanism used,
/// enabling generic code to select appropriate strategies at compile time.
pub trait ConcurrencyPolicy: Send + Sync {
    /// Human-readable name for diagnostics
    const NAME: &'static str;
}

/// No internal synchronization — caller must hold external lock (DashMap shard, RwLock, etc.)
pub struct ExternalLock;
impl ConcurrencyPolicy for ExternalLock {
    const NAME: &'static str = "ExternalLock";
}

/// Internal RwLock — chain can be shared across threads with &self methods
pub struct InternalRwLock;
impl ConcurrencyPolicy for InternalRwLock {
    const NAME: &'static str = "InternalRwLock";
}

/// Lock-free atomics — CAS-based, fully concurrent &self methods
pub struct LockFreeAtomic;
impl ConcurrencyPolicy for LockFreeAtomic {
    const NAME: &'static str = "LockFreeAtomic";
}

// Safety: marker types are stateless
unsafe impl Send for ExternalLock {}
unsafe impl Sync for ExternalLock {}
unsafe impl Send for InternalRwLock {}
unsafe impl Sync for InternalRwLock {}
unsafe impl Send for LockFreeAtomic {}
unsafe impl Sync for LockFreeAtomic {}

// =============================================================================
// Rec 11: Consolidated Binary-Search Version Chain
// =============================================================================

/// Trait for version entry types used in binary-search chains.
///
/// Implemented by `durable_storage::Version` and `mvcc_concurrent::VersionEntry`
/// to allow a single `BinarySearchChain<E>` to handle both.
pub trait ChainEntry: Sized + std::fmt::Debug {
    /// Get the commit timestamp (0 = uncommitted)
    fn commit_ts(&self) -> u64;
    /// Get the transaction ID that created this version
    fn txn_id(&self) -> u64;
    /// Set the commit timestamp (called during commit)
    fn set_commit_ts(&mut self, ts: u64);
}

/// A binary-search optimized version chain — the consolidated core.
///
/// Stores committed versions sorted descending by `commit_ts`, with at most
/// one uncommitted version slot. Uses `partition_point` for O(log V) lookups.
///
/// This struct captures the duplicated logic previously in:
/// - `durable_storage::VersionChain`
/// - `mvcc_concurrent::VersionChain`
///
/// Both modules now wrap `BinarySearchChain<E>` and delegate the core
/// binary-search / commit / abort / read / conflict-check operations to it.
#[derive(Debug)]
pub struct BinarySearchChain<E: ChainEntry> {
    /// Committed versions sorted by commit_ts DESCENDING (newest first)
    committed: Vec<E>,
    /// Single uncommitted version slot (at most one per transaction writing this key)
    uncommitted: Option<E>,
}

impl<E: ChainEntry> Default for BinarySearchChain<E> {
    fn default() -> Self {
        Self::new()
    }
}

impl<E: ChainEntry> BinarySearchChain<E> {
    /// Create a new empty chain.
    #[inline]
    pub fn new() -> Self {
        Self {
            committed: Vec::new(),
            uncommitted: None,
        }
    }

    // ---- Uncommitted slot management ----

    /// Replace the uncommitted slot. Returns the previous entry if any.
    #[inline]
    pub fn set_uncommitted(&mut self, entry: E) -> Option<E> {
        self.uncommitted.replace(entry)
    }

    /// Reference to the uncommitted entry.
    #[inline]
    pub fn uncommitted(&self) -> Option<&E> {
        self.uncommitted.as_ref()
    }

    /// Mutable reference to the uncommitted entry.
    #[inline]
    pub fn uncommitted_mut(&mut self) -> Option<&mut E> {
        self.uncommitted.as_mut()
    }

    // ---- Core operations ----

    /// Commit the uncommitted version with the given `commit_ts`.
    ///
    /// Moves the entry from the uncommitted slot into the sorted committed
    /// list at the correct position. Returns `true` on success.
    ///
    /// Complexity: O(log V) binary search + O(V) insert (amortised O(1) for newest).
    pub fn commit(&mut self, txn_id: u64, commit_ts: u64) -> bool {
        if let Some(ref mut v) = self.uncommitted {
            if v.txn_id() == txn_id && v.commit_ts() == 0 {
                v.set_commit_ts(commit_ts);
                let committed_version = self.uncommitted.take().unwrap();
                let insert_pos = self
                    .committed
                    .partition_point(|e| e.commit_ts() > commit_ts);
                self.committed.insert(insert_pos, committed_version);
                return true;
            }
        }
        false
    }

    /// Abort a transaction — clear the uncommitted slot if it matches `txn_id`.
    #[inline]
    pub fn abort(&mut self, txn_id: u64) {
        if let Some(ref v) = self.uncommitted {
            if v.txn_id() == txn_id {
                self.uncommitted = None;
            }
        }
    }

    /// Read the visible version at the given snapshot timestamp.
    ///
    /// If `current_txn_id` is provided and matches the uncommitted version's
    /// transaction, the uncommitted version is returned (read-own-writes).
    ///
    /// Otherwise performs O(log V) binary search for the newest committed
    /// version with `commit_ts < snapshot_ts`.
    #[inline]
    pub fn read_at(&self, snapshot_ts: u64, current_txn_id: Option<u64>) -> Option<&E> {
        if let Some(txn_id) = current_txn_id {
            if let Some(ref v) = self.uncommitted {
                if v.txn_id() == txn_id {
                    return Some(v);
                }
            }
        }
        let idx = self
            .committed
            .partition_point(|v| v.commit_ts() >= snapshot_ts);
        self.committed.get(idx)
    }

    /// Check if there's a write-write conflict with another transaction.
    #[inline]
    pub fn has_write_conflict(&self, my_txn_id: u64) -> bool {
        if let Some(ref v) = self.uncommitted {
            return v.txn_id() != my_txn_id;
        }
        false
    }

    /// GC versions older than `min_active_ts`.
    ///
    /// Retention must agree with [`read_at`](Self::read_at), whose visibility
    /// rule is **strict**: a reader at snapshot `S` observes the newest version
    /// with `commit_ts < S`. The smallest live snapshot is `min_active_ts`, so
    /// the oldest version any reader can still need is the newest one with
    /// `commit_ts < min_active_ts`. We therefore keep every version with
    /// `commit_ts >= min_active_ts` **plus one anchor** — the newest version
    /// strictly below `min_active_ts`.
    ///
    /// Using `>=` here (rather than `>`) is load-bearing for read-safety: with
    /// `>`, a boundary version whose `commit_ts == min_active_ts` would be
    /// chosen as the anchor and the genuinely-needed older version (the one a
    /// reader at `snapshot == min_active_ts` resolves, since that boundary
    /// version is *not* visible to it under the strict `<` rule) would be
    /// freed — causing the reader to observe a spurious `None`.
    pub fn gc_by_ts(&mut self, min_active_ts: u64) {
        if self.committed.len() <= 1 {
            return;
        }
        let split_idx = self
            .committed
            .partition_point(|v| v.commit_ts() >= min_active_ts);
        let keep_count = if split_idx < self.committed.len() {
            split_idx + 1
        } else {
            split_idx
        };
        self.committed.truncate(keep_count);
    }

    // ---- Accessors ----

    /// Total version count (committed + uncommitted).
    #[inline]
    pub fn version_count(&self) -> usize {
        self.committed.len() + usize::from(self.uncommitted.is_some())
    }

    /// Number of committed versions.
    #[inline]
    pub fn committed_count(&self) -> usize {
        self.committed.len()
    }

    /// Check if chain is empty.
    #[inline]
    pub fn is_empty(&self) -> bool {
        self.committed.is_empty() && self.uncommitted.is_none()
    }

    /// Slice of committed versions (newest first).
    #[inline]
    pub fn committed_versions(&self) -> &[E] {
        &self.committed
    }

    /// Mutable access to committed versions (for custom GC).
    #[inline]
    pub fn committed_versions_mut(&mut self) -> &mut Vec<E> {
        &mut self.committed
    }

    /// Latest version: uncommitted if present, else newest committed.
    #[inline]
    pub fn latest(&self) -> Option<&E> {
        self.uncommitted.as_ref().or_else(|| self.committed.first())
    }

    /// Newest committed version only.
    #[inline]
    pub fn latest_committed(&self) -> Option<&E> {
        self.committed.first()
    }
}

// =============================================================================
// Rec 11: Unified MVCC Store Trait
// =============================================================================

/// Error type for MVCC store operations.
#[derive(Debug)]
pub enum MvccStoreError {
    /// Another uncommitted write exists on this key
    WriteConflict,
}

impl std::fmt::Display for MvccStoreError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::WriteConflict => write!(f, "write-write conflict"),
        }
    }
}

impl std::error::Error for MvccStoreError {}

/// GC statistics returned by `MvccStore::mvcc_gc`.
#[derive(Debug, Default, Clone)]
pub struct MvccGcStats {
    pub versions_removed: usize,
    pub keys_scanned: usize,
}

/// Unified MVCC key-value store trait.
///
/// Provides a common interface for:
/// - `durable_storage::MvccMemTable`
/// - `mvcc_concurrent::VersionStore`
/// - `epoch_mvcc::EpochMvccStore`
///
/// Callers can program against this trait to be agnostic to the
/// underlying concurrency / storage strategy.
pub trait MvccStore: Send + Sync {
    /// Read the visible value at `snapshot_ts`, optionally seeing own writes
    /// from `txn_id`.
    fn mvcc_get(&self, key: &[u8], snapshot_ts: u64, txn_id: Option<u64>) -> Option<Vec<u8>>;

    /// Write a value (or tombstone `None`) as uncommitted for the given transaction.
    fn mvcc_put(
        &self,
        key: &[u8],
        value: Option<Vec<u8>>,
        txn_id: u64,
    ) -> Result<(), MvccStoreError>;

    /// Commit one key's uncommitted write. Returns `true` if found and committed.
    fn mvcc_commit_key(&self, key: &[u8], txn_id: u64, commit_ts: u64) -> bool;

    /// Abort one key's uncommitted write.
    fn mvcc_abort_key(&self, key: &[u8], txn_id: u64);

    /// Check if there's an uncommitted write conflict on a key.
    fn mvcc_has_conflict(&self, key: &[u8], txn_id: u64) -> bool;

    /// Run garbage collection. Returns statistics.
    fn mvcc_gc(&self, min_ts: u64) -> MvccGcStats;

    /// Number of distinct keys in the store.
    fn mvcc_key_count(&self) -> usize;
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_version_meta_visibility() {
        let mut meta = VersionMeta::new_uncommitted(1, 100);

        // Uncommitted - only visible to creator
        let ctx = VisibilityContext::new(1, 200);
        assert!(meta.is_visible(&ctx));

        let ctx2 = VisibilityContext::new(2, 200);
        assert!(!meta.is_visible(&ctx2));

        // After commit - visible to later snapshots
        meta.commit(150);
        assert!(meta.is_visible(&ctx2));

        // Not visible to earlier snapshots
        let ctx3 = VisibilityContext::new(3, 100);
        assert!(!meta.is_visible(&ctx3));
    }

    #[test]
    fn test_version_meta_deletion() {
        let mut meta = VersionMeta::new_uncommitted(1, 100);
        meta.commit(150);
        meta.delete(2, 200);

        // Visible before deletion
        let ctx = VisibilityContext::new(3, 180);
        assert!(meta.is_visible(&ctx));

        // Not visible after deletion
        let ctx2 = VisibilityContext::new(3, 250);
        assert!(!meta.is_visible(&ctx2));
    }

    #[test]
    fn test_visibility_context_committed_before() {
        let mut active = std::collections::HashSet::new();
        active.insert(5);

        let ctx = VisibilityContext::with_active_txns(1, 200, active);

        // Committed before snapshot
        assert!(ctx.is_committed_before(2, Some(100)));

        // Committed after snapshot
        assert!(!ctx.is_committed_before(3, Some(250)));

        // Active transaction - not committed
        assert!(!ctx.is_committed_before(5, Some(100)));

        // No commit timestamp
        assert!(!ctx.is_committed_before(6, None));
    }

    #[test]
    fn test_concurrency_policy_names() {
        assert_eq!(ExternalLock::NAME, "ExternalLock");
        assert_eq!(InternalRwLock::NAME, "InternalRwLock");
        assert_eq!(LockFreeAtomic::NAME, "LockFreeAtomic");
    }

    // ---- BinarySearchChain tests (Rec 11) ----

    /// Minimal entry type for testing the consolidated chain.
    #[derive(Debug, Clone)]
    struct TestEntry {
        commit_ts: u64,
        txn_id: u64,
        val: i32,
    }

    impl ChainEntry for TestEntry {
        fn commit_ts(&self) -> u64 {
            self.commit_ts
        }
        fn txn_id(&self) -> u64 {
            self.txn_id
        }
        fn set_commit_ts(&mut self, ts: u64) {
            self.commit_ts = ts;
        }
    }

    #[test]
    fn test_binary_search_chain_commit_and_read() {
        let mut chain = BinarySearchChain::<TestEntry>::new();
        assert!(chain.is_empty());

        // Add uncommitted, then commit
        chain.set_uncommitted(TestEntry {
            commit_ts: 0,
            txn_id: 1,
            val: 10,
        });
        assert_eq!(chain.version_count(), 1);

        // Read own writes
        let v = chain.read_at(100, Some(1)).unwrap();
        assert_eq!(v.val, 10);

        // Other txn can't see it
        assert!(chain.read_at(100, Some(2)).is_none());

        // Commit
        assert!(chain.commit(1, 50));
        assert_eq!(chain.committed_count(), 1);

        // Now visible to snapshot_ts > 50
        let v = chain.read_at(51, None).unwrap();
        assert_eq!(v.val, 10);

        // Not visible to snapshot_ts <= 50
        assert!(chain.read_at(50, None).is_none());
    }

    #[test]
    fn test_binary_search_chain_abort() {
        let mut chain = BinarySearchChain::<TestEntry>::new();
        chain.set_uncommitted(TestEntry {
            commit_ts: 0,
            txn_id: 1,
            val: 10,
        });
        chain.abort(1);
        assert!(chain.is_empty());
        // Abort wrong txn is a no-op
        chain.set_uncommitted(TestEntry {
            commit_ts: 0,
            txn_id: 2,
            val: 20,
        });
        chain.abort(1);
        assert_eq!(chain.version_count(), 1);
    }

    #[test]
    fn test_binary_search_chain_write_conflict() {
        let mut chain = BinarySearchChain::<TestEntry>::new();
        assert!(!chain.has_write_conflict(1));

        chain.set_uncommitted(TestEntry {
            commit_ts: 0,
            txn_id: 1,
            val: 10,
        });
        assert!(!chain.has_write_conflict(1)); // own txn
        assert!(chain.has_write_conflict(2)); // other txn
    }

    #[test]
    fn test_binary_search_chain_gc() {
        let mut chain = BinarySearchChain::<TestEntry>::new();

        // Commit 5 versions at ts 10, 20, 30, 40, 50
        for i in 1..=5u64 {
            chain.set_uncommitted(TestEntry {
                commit_ts: 0,
                txn_id: i,
                val: i as i32,
            });
            chain.commit(i, i * 10);
        }
        assert_eq!(chain.committed_count(), 5);

        // GC with min_active_ts = 25 → keep ts > 25 (30, 40, 50) + 1 anchor (20)
        chain.gc_by_ts(25);
        assert_eq!(chain.committed_count(), 4); // 50, 40, 30, 20

        // GC with min_active_ts = 45 → keep ts > 45 (50) + 1 anchor (40)
        chain.gc_by_ts(45);
        assert_eq!(chain.committed_count(), 2); // 50, 40
    }

    #[test]
    fn test_binary_search_chain_multiple_versions() {
        let mut chain = BinarySearchChain::<TestEntry>::new();

        // Commit in order: ts=100, ts=200, ts=300
        for (i, ts) in [100u64, 200, 300].iter().enumerate() {
            let txn = (i + 1) as u64;
            chain.set_uncommitted(TestEntry {
                commit_ts: 0,
                txn_id: txn,
                val: *ts as i32,
            });
            chain.commit(txn, *ts);
        }

        // Snapshot at 150 → sees ts=100
        assert_eq!(chain.read_at(150, None).unwrap().val, 100);
        // Snapshot at 250 → sees ts=200
        assert_eq!(chain.read_at(250, None).unwrap().val, 200);
        // Snapshot at 350 → sees ts=300
        assert_eq!(chain.read_at(350, None).unwrap().val, 300);
        // Snapshot at 50 → sees nothing
        assert!(chain.read_at(50, None).is_none());
    }
}

#[cfg(test)]
mod version_chain_properties {
    use super::*;
    use proptest::prelude::*;

    /// Minimal committed entry for property testing the chain's read/GC contract.
    #[derive(Debug, Clone)]
    struct PropEntry {
        commit_ts: u64,
        val: i32,
    }

    impl ChainEntry for PropEntry {
        fn commit_ts(&self) -> u64 {
            self.commit_ts
        }
        fn txn_id(&self) -> u64 {
            0
        }
        fn set_commit_ts(&mut self, ts: u64) {
            self.commit_ts = ts;
        }
    }

    /// Build a chain from a set of distinct commit timestamps (committed in
    /// ascending order, mirroring how `MvccManager` allocates commit_ts).
    fn build_chain(mut tss: Vec<u64>) -> BinarySearchChain<PropEntry> {
        tss.sort_unstable();
        tss.dedup();
        let mut chain = BinarySearchChain::<PropEntry>::new();
        for ts in tss {
            chain.set_uncommitted(PropEntry {
                commit_ts: 0,
                val: ts as i32,
            });
            // txn_id 0 is fine here: one writer at a time in this model.
            chain.commit(0, ts);
        }
        chain
    }

    /// Reference visibility oracle: the value a reader at `snapshot` must see is
    /// the largest commit_ts strictly less than `snapshot` (strict `<` rule).
    fn expected_visible(tss: &[u64], snapshot: u64) -> Option<i32> {
        tss.iter()
            .copied()
            .filter(|&ts| ts < snapshot)
            .max()
            .map(|ts| ts as i32)
    }

    proptest! {
        /// `read_at` matches the strict-visibility oracle for any snapshot.
        #[test]
        fn prop_read_at_matches_strict_visibility(
            tss in prop::collection::vec(1u64..1000, 0..32),
            snapshot in 0u64..1100,
        ) {
            let mut sorted = tss.clone();
            sorted.sort_unstable();
            sorted.dedup();
            let chain = build_chain(tss);

            let got = chain.read_at(snapshot, None).map(|e| e.val);
            prop_assert_eq!(got, expected_visible(&sorted, snapshot));
        }

        /// GC is read-safe: after `gc_by_ts(min_active_ts)`, every reader whose
        /// snapshot is `>= min_active_ts` still observes exactly the same value
        /// as before GC. This is the invariant the T8 anchor fix (`>=`) restores —
        /// a too-aggressive GC would make such a reader see a spurious `None`.
        #[test]
        fn prop_gc_preserves_reads_for_live_snapshots(
            tss in prop::collection::vec(1u64..1000, 1..32),
            min_active in 1u64..1000,
            // Several snapshots at or above the GC watermark.
            extra in prop::collection::vec(0u64..200, 1..6),
        ) {
            let mut sorted = tss.clone();
            sorted.sort_unstable();
            sorted.dedup();

            let mut chain = build_chain(tss);

            // Snapshots that remain valid after GC are those >= min_active_ts.
            let snapshots: Vec<u64> = extra.iter().map(|d| min_active + *d).collect();

            // Capture pre-GC observations.
            let before: Vec<Option<i32>> = snapshots
                .iter()
                .map(|&s| chain.read_at(s, None).map(|e| e.val))
                .collect();

            chain.gc_by_ts(min_active);

            // Post-GC observations must be identical for every live snapshot.
            for (i, &s) in snapshots.iter().enumerate() {
                let after = chain.read_at(s, None).map(|e| e.val);
                prop_assert_eq!(after, before[i],
                    "GC at {} changed read at snapshot {}", min_active, s);
                // And it must still equal the oracle.
                prop_assert_eq!(after, expected_visible(&sorted, s));
            }
        }
    }
}