commonware_storage/archive/prunable/

mod.rs

//! A prunable key-value store for ordered data.
//!
//! Data is stored across two backends: [crate::journal::segmented::fixed] for fixed-size index entries and
//! [crate::journal::segmented::glob::Glob] for values (managed by [crate::journal::segmented::oversized]).
//! The location of written data is stored in-memory by both index and key (via [crate::index::unordered::Index])
//! to enable efficient lookups (on average).
//!
//! _Notably, [Archive] does not make use of compaction nor on-disk indexes (and thus has no read
//! nor write amplification during normal operation).
//!
//! # Format
//!
//! [Archive] uses a two-journal structure for efficient page cache usage:
//!
//! **Index Journal (segmented/fixed)** - Fixed-size entries for fast startup replay:
//! ```text
//! +---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+
//! | 0 | 1 | 2 | 3 | 4 | 5 | 6 | 7 | 8 | 9 |10 |11 |12 |13 |14 |15 |16 |17 |18 |19 |20 |21 |22 |23 |
//! +---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+
//! |          Index(u64)           |Key(Fixed Size)|        val_offset(u64)        | val_size(u32) |
//! +---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+
//! ```
//!
//! **Value Blob** - Raw values with CRC32 checksums (direct reads, no page cache):
//! ```text
//! +---+---+---+---+---+---+---+---+---+---+---+---+
//! |     Compressed Data (variable)    |   CRC32   |
//! +---+---+---+---+---+---+---+---+---+---+---+---+
//! ```
//!
//! # Uniqueness
//!
//! Indices are unique for [Archive] and writing to an occupied index is a no-op. Duplicate
//! indices can be stored via [`crate::archive::MultiArchive::put_multi`].
//!
//! Keys may be stored at multiple indices with either put variant. A lookup by
//! [`crate::archive::Identifier::Key`] may return any of the values at that key. Entries
//! whose index has been pruned are never returned or reported as present, so a key matching
//! both a pruned and a non-pruned entry resolves to the non-pruned entry.
//!
//! ## Conflicts
//!
//! Because a translated representation of a key is only ever stored in memory, it is possible (and
//! expected) that two keys will eventually be represented by the same translated key. To handle
//! this case, [Archive] must check the persisted form of all conflicting keys to ensure data from
//! the correct key is returned. To support efficient checks, [Archive] (via
//! [crate::index::unordered::Index]) keeps a linked list of all keys with the same translated
//! prefix:
//!
//! ```rust
//! struct Record {
//!     index: u64,
//!
//!     next: Option<Box<Record>>,
//! }
//! ```
//!
//! _To avoid random memory reads in the common case, the in-memory index directly stores the first
//! item in the linked list instead of a pointer to the first item._
//!
//! `index` is the key to the map used to serve lookups by `index` that stores the position in the
//! index journal (selected by `section = index / items_per_section * items_per_section` to minimize
//! the number of open blobs):
//!
//! ```text
//! // Maps index -> position in index journal
//! indices: BTreeMap<u64, u64>
//! ```
//!
//! _If the [Translator] provided by the caller does not uniformly distribute keys across the key
//! space or uses a translated representation that means keys on average have many conflicts,
//! performance will degrade._
//!
//! ## Memory Overhead
//!
//! [Archive] uses two maps to enable lookups by both index and key. The memory used to track each
//! index item is `8 + 8` (where `8` is the index and `8` is the position in the index journal).
//! The memory used to track each key item is `~translated(key).len() + 16` bytes (where `16` is the
//! size of the `Record` struct). This means that an [Archive] employing a [Translator] that uses
//! the first `8` bytes of a key will use `~40` bytes to index each key.
//!
//! ### MultiArchive Overhead
//!
//! [Archive] stores index positions in a dual-map layout:
//! - `indices: BTreeMap<u64, u64>` tracks the first position for each index.
//! - `extra_indices: BTreeMap<u64, Vec<u64>>` tracks additional positions for indices written via
//!   [crate::archive::MultiArchive::put_multi].
//!
//! This means the baseline overhead above remains unchanged for the first item at an index. For
//! indices with duplicates, the additional in-memory payload is:
//! - one `Vec<u64>` header (`24` bytes), and
//! - `n * 8` bytes for `n` additional positions.
//!
//! Equivalently, this is `24 + (n * 8)` bytes per duplicated index, excluding `BTreeMap` node
//! overhead for `extra_indices`.
//!
//! # Pruning
//!
//! [Archive] supports pruning up to a minimum `index` using the `prune` method. After `prune` is
//! called on a `section`, all interaction with a `section` less than the pruned `section` will
//! return an error.
//!
//! ## Lazy Index Cleanup
//!
//! Instead of performing a full iteration of the in-memory index, storing an additional in-memory
//! index per `section`, or replaying a `section` of the value blob,
//! [Archive] lazily cleans up the [crate::index::unordered::Index] after pruning. When a new key is
//! stored that overlaps (same translated value) with a pruned key, the pruned key is removed from
//! the in-memory index.
//!
//! # Read Path
//!
//! All reads (by index or key) first read the index entry from the index journal to get the
//! value location (offset and size), then read the value from the value blob. The index journal
//! uses a page cache for caching, so hot entries are served from memory. Values are read directly
//! from disk without caching to avoid polluting the page cache with large values.
//!
//! # Compression
//!
//! [Archive] supports compressing data before storing it on disk. This can be enabled by setting
//! the `compression` field in the `Config` struct to a valid `zstd` compression level. This setting
//! can be changed between initializations of [Archive], however, it must remain populated if any
//! data was written with compression enabled.
//!
//! # Querying for Gaps
//!
//! [Archive] tracks gaps in the index space to enable the caller to efficiently fetch unknown keys
//! using `next_gap`. This is a very common pattern when syncing blocks in a blockchain.
//!
//! # Example
//!
//! ```rust
//! use commonware_runtime::{Spawner, Runner, deterministic, buffer::paged::CacheRef};
//! use commonware_cryptography::{Hasher as _, Sha256};
//! use commonware_storage::{
//!     translator::FourCap,
//!     archive::{
//!         Archive as _,
//!         prunable::{Archive, Config},
//!     },
//! };
//! use commonware_utils::{NZUsize, NZU16, NZU64};
//!
//! let executor = deterministic::Runner::default();
//! executor.start(|context| async move {
//!     // Create an archive
//!     let cfg = Config {
//!         translator: FourCap,
//!         key_partition: "demo-index".into(),
//!         key_page_cache: CacheRef::from_pooler(&context, NZU16!(1024), NZUsize!(10)),
//!         value_partition: "demo-value".into(),
//!         compression: Some(3),
//!         codec_config: (),
//!         items_per_section: NZU64!(1024),
//!         key_write_buffer: NZUsize!(1024 * 1024),
//!         value_write_buffer: NZUsize!(1024 * 1024),
//!         replay_buffer: NZUsize!(4096),
//!     };
//!     let mut archive = Archive::init(context, cfg).await.unwrap();
//!
//!     // Put a key
//!     archive.put(1, Sha256::hash(b"data"), 10).await.unwrap();
//!
//!     // Sync the archive
//!     archive.sync().await.unwrap();
//! });
//! ```

use crate::translator::Translator;
use commonware_runtime::buffer::paged::CacheRef;
use std::num::{NonZeroU64, NonZeroUsize};

mod storage;
pub use storage::Archive;

/// Configuration for [Archive] storage.
#[derive(Clone)]
pub struct Config<T: Translator, C> {
    /// Logic to transform keys into their index representation.
    ///
    /// [Archive] assumes that all internal keys are spread uniformly across the key space.
    /// If that is not the case, lookups may be O(n) instead of O(1).
    pub translator: T,

    /// The partition to use for the key journal (stores index+key metadata).
    pub key_partition: String,

    /// The page cache to use for the key journal.
    pub key_page_cache: CacheRef,

    /// The partition to use for the value blob (stores values).
    pub value_partition: String,

    /// The compression level to use for the value blob.
    pub compression: Option<u8>,

    /// The [commonware_codec::Codec] configuration to use for the value stored in the archive.
    pub codec_config: C,

    /// The number of items per section (the granularity of pruning).
    pub items_per_section: NonZeroU64,

    /// The amount of bytes that can be buffered for the key journal before being written to a
    /// [commonware_runtime::Blob].
    pub key_write_buffer: NonZeroUsize,

    /// The amount of bytes that can be buffered for the value journal before being written to a
    /// [commonware_runtime::Blob].
    pub value_write_buffer: NonZeroUsize,

    /// The buffer size to use when replaying a [commonware_runtime::Blob].
    pub replay_buffer: NonZeroUsize,
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{
        archive::{Archive as _, Error, Identifier, MultiArchive as _},
        journal::Error as JournalError,
        translator::{FourCap, TwoCap},
    };
    use commonware_codec::{DecodeExt, Error as CodecError};
    use commonware_macros::{test_group, test_traced};
    use commonware_runtime::{
        deterministic,
        mocks::{
            fail_pending_syncs, release_next_pending_syncs, release_pending_syncs,
            DelayedSyncContext, PendingSyncs,
        },
        telemetry::metrics::has_metric_value,
        BufferPooler, Error as RError, Metrics as _, Runner, Spawner as _, Supervisor as _,
    };
    use commonware_utils::{sequence::FixedBytes, NZUsize, NZU16, NZU64};
    use rand::RngExt as _;
    use std::{
        collections::BTreeMap,
        num::{NonZeroU16, NonZeroU64},
        sync::{
            atomic::{AtomicUsize, Ordering},
            Arc,
        },
    };

    fn test_key(key: &str) -> FixedBytes<64> {
        let mut buf = [0u8; 64];
        let key = key.as_bytes();
        assert!(key.len() <= buf.len());
        buf[..key.len()].copy_from_slice(key);
        FixedBytes::decode(buf.as_ref()).unwrap()
    }

    const DEFAULT_ITEMS_PER_SECTION: u64 = 65536;
    const DEFAULT_WRITE_BUFFER: usize = 1024;
    const DEFAULT_REPLAY_BUFFER: usize = 4096;
    const PAGE_SIZE: NonZeroU16 = NZU16!(1024);
    const PAGE_CACHE_SIZE: NonZeroUsize = NZUsize!(10);

    fn test_config<E: BufferPooler>(
        context: &E,
        items_per_section: NonZeroU64,
    ) -> Config<FourCap, ()> {
        Config {
            translator: FourCap,
            key_partition: "test-index".into(),
            key_page_cache: CacheRef::from_pooler(context, PAGE_SIZE, PAGE_CACHE_SIZE),
            value_partition: "test-value".into(),
            codec_config: (),
            compression: None,
            key_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
            value_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
            replay_buffer: NZUsize!(DEFAULT_REPLAY_BUFFER),
            items_per_section,
        }
    }

    #[test_traced]
    fn test_put_after_start_sync_is_accepted_before_handle_completes() {
        let executor = deterministic::Runner::default();
        let (_, checkpoint) = executor.start_and_recover(|context| async move {
            let pending = PendingSyncs::default();
            let context = DelayedSyncContext {
                inner: context,
                pending: pending.clone(),
            };
            let cfg = test_config(&context, NZU64!(DEFAULT_ITEMS_PER_SECTION));
            let mut archive = Archive::init(context.child("storage"), cfg)
                .await
                .expect("Failed to initialize archive");

            let handle = archive
                .put_start_sync(1, test_key("aaa"), 10)
                .await
                .expect("Failed to start sync");
            let pending_after_start = pending.lock().len();
            assert!(
                pending_after_start > 0,
                "put_start_sync should return while the sync handle is still pending"
            );

            archive
                .put(2, test_key("bbb"), 20)
                .await
                .expect("archive should remain usable before sync completion");
            assert_eq!(
                pending.lock().len(),
                pending_after_start,
                "put should not issue a new storage sync while accepting later data"
            );
            assert_eq!(archive.get(Identifier::Index(1)).await.unwrap(), Some(10));

            release_pending_syncs(&pending);
            handle.await.expect("sync handle should complete");

            let follow_up = archive
                .start_sync()
                .await
                .expect("Failed to start next sync");
            assert!(
                !pending.lock().is_empty(),
                "the later put must remain pending for a future sync"
            );
            release_pending_syncs(&pending);
            follow_up.await.expect("follow-up sync should complete");
        });

        deterministic::Runner::from(checkpoint).start(|context| async move {
            let cfg = test_config(&context, NZU64!(DEFAULT_ITEMS_PER_SECTION));
            let archive = Archive::<_, _, FixedBytes<64>, i32>::init(context.child("reopen"), cfg)
                .await
                .expect("Failed to reopen archive");

            assert_eq!(archive.get(Identifier::Index(1)).await.unwrap(), Some(10));
            assert_eq!(archive.get(Identifier::Index(2)).await.unwrap(), Some(20));
        });
    }

    #[test_traced]
    fn test_duplicate_put_start_sync_observes_in_flight_sync() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let pending = PendingSyncs::default();
            let context = DelayedSyncContext {
                inner: context,
                pending: pending.clone(),
            };
            let cfg = test_config(&context, NZU64!(DEFAULT_ITEMS_PER_SECTION));
            let mut archive = Archive::init(context.child("storage"), cfg)
                .await
                .expect("Failed to initialize archive");

            let first = archive
                .put_start_sync(1, test_key("aaa"), 10)
                .await
                .expect("Failed to start sync");
            assert_eq!(pending.lock().len(), 2);

            let second = archive
                .put_start_sync(1, test_key("duplicate"), 99)
                .await
                .expect("Failed to start duplicate sync");
            assert_eq!(
                pending.lock().len(),
                2,
                "duplicate put_start_sync must not issue a new storage sync"
            );

            let started = Arc::new(AtomicUsize::new(0));
            let completed = Arc::new(AtomicUsize::new(0));
            let started_clone = started.clone();
            let completed_clone = completed.clone();
            let waiter = context.inner.child("duplicate").spawn(|_| async move {
                started_clone.fetch_add(1, Ordering::Relaxed);
                second.await.expect("duplicate sync handle should complete");
                completed_clone.fetch_add(1, Ordering::Relaxed);
            });

            while started.load(Ordering::Relaxed) == 0 {
                commonware_runtime::reschedule().await;
            }
            commonware_runtime::reschedule().await;
            assert_eq!(
                completed.load(Ordering::Relaxed),
                0,
                "duplicate put_start_sync must observe the original in-flight sync"
            );

            release_pending_syncs(&pending);
            first.await.expect("first sync handle should complete");
            while completed.load(Ordering::Relaxed) == 0 {
                commonware_runtime::reschedule().await;
            }
            waiter.await.expect("duplicate waiter failed");

            assert_eq!(archive.get(Identifier::Index(1)).await.unwrap(), Some(10));
        });
    }

    #[test_traced]
    fn test_overlapping_put_start_sync_waits_for_in_flight_sync() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let pending = PendingSyncs::default();
            let context = DelayedSyncContext {
                inner: context,
                pending: pending.clone(),
            };
            let cfg = test_config(&context, NZU64!(DEFAULT_ITEMS_PER_SECTION));
            let mut archive = Archive::init(context.child("storage"), cfg)
                .await
                .expect("Failed to initialize archive");

            let first = archive
                .put_start_sync(1, test_key("aaa"), 10)
                .await
                .expect("Failed to start sync");
            let pending_after_first = pending.lock().len();
            assert!(pending_after_first > 0);

            let started = Arc::new(AtomicUsize::new(0));
            let completed = Arc::new(AtomicUsize::new(0));
            let started_clone = started.clone();
            let completed_clone = completed.clone();
            let waiter = context.inner.child("second").spawn(|_| async move {
                started_clone.fetch_add(1, Ordering::Relaxed);
                let second = archive
                    .put_start_sync(2, test_key("bbb"), 20)
                    .await
                    .expect("Failed to start second sync");
                completed_clone.fetch_add(1, Ordering::Relaxed);
                (archive, second)
            });

            while started.load(Ordering::Relaxed) == 0 {
                commonware_runtime::reschedule().await;
            }
            commonware_runtime::reschedule().await;
            assert_eq!(completed.load(Ordering::Relaxed), 0);
            assert_eq!(
                pending.lock().len(),
                pending_after_first,
                "second put_start_sync must not start new syncs before the first completes"
            );

            release_pending_syncs(&pending);
            first.await.expect("first sync handle should complete");
            while completed.load(Ordering::Relaxed) == 0 {
                commonware_runtime::reschedule().await;
            }
            let (archive, second) = waiter.await.expect("second put task failed");
            assert!(!pending.lock().is_empty());
            release_pending_syncs(&pending);
            second.await.expect("second sync handle should complete");

            assert_eq!(archive.get(Identifier::Index(1)).await.unwrap(), Some(10));
            assert_eq!(archive.get(Identifier::Index(2)).await.unwrap(), Some(20));
        });
    }

    #[test_traced]
    fn test_sync_after_put_start_sync_waits_for_in_flight_sync() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let pending = PendingSyncs::default();
            let context = DelayedSyncContext {
                inner: context,
                pending: pending.clone(),
            };
            let cfg = test_config(&context, NZU64!(DEFAULT_ITEMS_PER_SECTION));
            let mut archive = Archive::init(context.child("storage"), cfg)
                .await
                .expect("Failed to initialize archive");

            let first = archive
                .put_start_sync(1, test_key("aaa"), 10)
                .await
                .expect("Failed to start sync");
            assert!(!pending.lock().is_empty());

            let started = Arc::new(AtomicUsize::new(0));
            let completed = Arc::new(AtomicUsize::new(0));
            let started_clone = started.clone();
            let completed_clone = completed.clone();
            let waiter = context.inner.child("sync").spawn(|_| async move {
                started_clone.fetch_add(1, Ordering::Relaxed);
                archive.sync().await.expect("sync should complete");
                completed_clone.fetch_add(1, Ordering::Relaxed);
                archive
            });

            while started.load(Ordering::Relaxed) == 0 {
                commonware_runtime::reschedule().await;
            }
            commonware_runtime::reschedule().await;
            assert_eq!(
                completed.load(Ordering::Relaxed),
                0,
                "shutdown sync must wait for the in-flight put_start_sync handle"
            );

            release_pending_syncs(&pending);
            first.await.expect("first sync handle should complete");
            while completed.load(Ordering::Relaxed) == 0 {
                commonware_runtime::reschedule().await;
            }
            let archive = waiter.await.expect("sync task failed");
            assert_eq!(archive.get(Identifier::Index(1)).await.unwrap(), Some(10));
        });
    }

    #[test_traced]
    fn test_destroy_after_put_start_sync_waits_for_in_flight_sync() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let pending = PendingSyncs::default();
            let context = DelayedSyncContext {
                inner: context,
                pending: pending.clone(),
            };
            let cfg = test_config(&context, NZU64!(DEFAULT_ITEMS_PER_SECTION));
            let mut archive =
                Archive::<_, _, FixedBytes<64>, i32>::init(context.child("storage"), cfg)
                    .await
                    .expect("Failed to initialize archive");

            let first = archive
                .put_start_sync(1, test_key("aaa"), 10)
                .await
                .expect("Failed to start sync");
            assert!(!pending.lock().is_empty());

            let started = Arc::new(AtomicUsize::new(0));
            let completed = Arc::new(AtomicUsize::new(0));
            let started_clone = started.clone();
            let completed_clone = completed.clone();
            let waiter = context.inner.child("destroy").spawn(|_| async move {
                started_clone.fetch_add(1, Ordering::Relaxed);
                archive.destroy().await.expect("destroy should complete");
                completed_clone.fetch_add(1, Ordering::Relaxed);
            });

            while started.load(Ordering::Relaxed) == 0 {
                commonware_runtime::reschedule().await;
            }
            commonware_runtime::reschedule().await;
            assert_eq!(
                completed.load(Ordering::Relaxed),
                0,
                "destroy must wait for the in-flight put_start_sync handle"
            );

            release_pending_syncs(&pending);
            first.await.expect("first sync handle should complete");
            while completed.load(Ordering::Relaxed) == 0 {
                commonware_runtime::reschedule().await;
            }
            waiter.await.expect("destroy task failed");
        });
    }

    #[test_traced]
    fn test_prune_after_put_start_sync_waits_for_in_flight_sync() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let pending = PendingSyncs::default();
            let context = DelayedSyncContext {
                inner: context,
                pending: pending.clone(),
            };
            let cfg = test_config(&context, NZU64!(1));
            let mut archive =
                Archive::<_, _, FixedBytes<64>, i32>::init(context.child("storage"), cfg)
                    .await
                    .expect("Failed to initialize archive");

            let first = archive
                .put_start_sync(1, test_key("aaa"), 10)
                .await
                .expect("Failed to start sync");
            assert!(!pending.lock().is_empty());

            let started = Arc::new(AtomicUsize::new(0));
            let completed = Arc::new(AtomicUsize::new(0));
            let started_clone = started.clone();
            let completed_clone = completed.clone();
            let waiter = context.inner.child("prune").spawn(|_| async move {
                started_clone.fetch_add(1, Ordering::Relaxed);
                archive.prune(2).await.expect("prune should complete");
                completed_clone.fetch_add(1, Ordering::Relaxed);
                archive
            });

            while started.load(Ordering::Relaxed) == 0 {
                commonware_runtime::reschedule().await;
            }
            commonware_runtime::reschedule().await;
            assert_eq!(
                completed.load(Ordering::Relaxed),
                0,
                "prune must wait for in-flight syncs on pruned sections"
            );

            release_pending_syncs(&pending);
            first
                .await
                .expect("sync handle should complete despite pruning");
            while completed.load(Ordering::Relaxed) == 0 {
                commonware_runtime::reschedule().await;
            }
            let archive = waiter.await.expect("prune task failed");
            assert_eq!(archive.get(Identifier::Index(1)).await.unwrap(), None);
        });
    }

    #[test_traced]
    fn test_prune_surfaces_failed_in_flight_sync() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let pending = PendingSyncs::default();
            let context = DelayedSyncContext {
                inner: context,
                pending: pending.clone(),
            };
            let cfg = test_config(&context, NZU64!(1));
            let mut archive =
                Archive::<_, _, FixedBytes<64>, i32>::init(context.child("storage"), cfg)
                    .await
                    .expect("Failed to initialize archive");

            let first = archive
                .put_start_sync(1, test_key("aaa"), 10)
                .await
                .expect("Failed to start sync");
            fail_pending_syncs(&pending);

            let err = archive
                .prune(2)
                .await
                .expect_err("prune must surface a failed in-flight sync");
            assert!(matches!(
                err,
                Error::Journal(JournalError::Runtime(RError::Io(_)))
            ));

            let err = first.await.expect_err("first sync handle should fail");
            assert!(matches!(err, RError::Io(_)));
        });
    }

    #[test_traced]
    fn test_put_start_sync_after_prune_drops_pruned_sync_requests() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let pending = PendingSyncs::default();
            let context = DelayedSyncContext {
                inner: context,
                pending: pending.clone(),
            };
            let cfg = test_config(&context, NZU64!(1));
            let mut archive =
                Archive::<_, _, FixedBytes<64>, i32>::init(context.child("storage"), cfg)
                    .await
                    .expect("Failed to initialize archive");

            let first = archive
                .put_start_sync(1, test_key("aaa"), 10)
                .await
                .expect("Failed to start sync");
            release_pending_syncs(&pending);
            first.await.expect("first sync handle should complete");

            archive.prune(2).await.expect("Failed to prune");

            // If pruning left section 1 in the retained sync-request set, these calls would trip
            // the journal's prune guard.
            let second = archive
                .put_start_sync(2, test_key("bbb"), 20)
                .await
                .expect("put_start_sync after prune should succeed");
            release_pending_syncs(&pending);
            second.await.expect("second sync handle should complete");
            archive
                .sync()
                .await
                .expect("sync after prune should succeed");

            assert_eq!(archive.get(Identifier::Index(1)).await.unwrap(), None);
            assert_eq!(archive.get(Identifier::Index(2)).await.unwrap(), Some(20));
        });
    }

    #[test_traced]
    fn test_overlapping_put_start_sync_restarts_after_all_handles_complete() {
        let executor = deterministic::Runner::default();
        let (_, checkpoint) = executor.start_and_recover(|context| async move {
            let pending = PendingSyncs::default();
            let context = DelayedSyncContext {
                inner: context,
                pending: pending.clone(),
            };
            let cfg = test_config(&context, NZU64!(1));
            let mut archive =
                Archive::<_, _, FixedBytes<64>, i32>::init(context.child("storage"), cfg)
                    .await
                    .expect("Failed to initialize archive");

            let first = archive
                .put_start_sync(1, test_key("aaa"), 10)
                .await
                .expect("Failed to start first sync");
            assert_eq!(pending.lock().len(), 2);

            let second = archive
                .put_start_sync(2, test_key("bbb"), 20)
                .await
                .expect("Failed to start second sync");
            assert_eq!(
                pending.lock().len(),
                4,
                "different sections should be able to have independent in-flight syncs"
            );

            release_pending_syncs(&pending);
            first.await.expect("first sync handle should complete");
            second.await.expect("second sync handle should complete");
        });

        deterministic::Runner::from(checkpoint).start(|context| async move {
            let cfg = test_config(&context, NZU64!(1));
            let archive = Archive::<_, _, FixedBytes<64>, i32>::init(context.child("reopen"), cfg)
                .await
                .expect("Failed to reopen archive");

            assert_eq!(archive.get(Identifier::Index(1)).await.unwrap(), Some(10));
            assert_eq!(archive.get(Identifier::Index(2)).await.unwrap(), Some(20));
        });
    }

    #[test_traced]
    fn test_overlapping_put_start_sync_restarts_only_completed_handles() {
        let executor = deterministic::Runner::default();
        let (_, checkpoint) = executor.start_and_recover(|context| async move {
            let pending = PendingSyncs::default();
            let context = DelayedSyncContext {
                inner: context,
                pending: pending.clone(),
            };
            let cfg = test_config(&context, NZU64!(1));
            let mut archive =
                Archive::<_, _, FixedBytes<64>, i32>::init(context.child("storage"), cfg)
                    .await
                    .expect("Failed to initialize archive");

            let first = archive
                .put_start_sync(1, test_key("aaa"), 10)
                .await
                .expect("Failed to start first sync");
            let second = archive
                .put_start_sync(2, test_key("bbb"), 20)
                .await
                .expect("Failed to start second sync");
            assert_eq!(pending.lock().len(), 4);

            release_next_pending_syncs(&pending, 2);
            first.await.expect("first sync handle should complete");

            drop(second);
            drop(archive);
        });

        deterministic::Runner::from(checkpoint).start(|context| async move {
            let cfg = test_config(&context, NZU64!(1));
            let archive = Archive::<_, _, FixedBytes<64>, i32>::init(context.child("reopen"), cfg)
                .await
                .expect("Failed to reopen archive");

            assert_eq!(archive.get(Identifier::Index(1)).await.unwrap(), Some(10));
            assert_eq!(archive.get(Identifier::Index(2)).await.unwrap(), None);
        });
    }

    #[test_traced]
    fn test_failed_start_sync_is_returned_by_next_start_sync_handle() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let pending = PendingSyncs::default();
            let context = DelayedSyncContext {
                inner: context,
                pending: pending.clone(),
            };
            let cfg = test_config(&context, NZU64!(DEFAULT_ITEMS_PER_SECTION));
            let mut archive =
                Archive::<_, _, FixedBytes<64>, i32>::init(context.child("storage"), cfg)
                    .await
                    .expect("Failed to initialize archive");

            let first = archive
                .put_start_sync(1, test_key("aaa"), 10)
                .await
                .expect("Failed to start sync");
            assert_eq!(pending.lock().len(), 2);
            fail_pending_syncs(&pending);

            archive
                .put(2, test_key("bbb"), 20)
                .await
                .expect("write should be accepted before observing the failed sync");

            let second = archive
                .start_sync()
                .await
                .expect("start_sync should return a handle for the failed sync");
            let err = second
                .await
                .expect_err("next start_sync handle should observe failed in-flight sync");
            assert!(matches!(err, RError::Io(_)));

            let err = first.await.expect_err("first sync handle should fail");
            assert!(matches!(err, RError::Io(_)));
        });
    }

    #[test_traced]
    fn test_archive_compression_then_none() {
        // Initialize the deterministic context
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Initialize the archive
            let cfg = Config {
                translator: FourCap,
                key_partition: "test-index".into(),
                key_page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
                value_partition: "test-value".into(),
                codec_config: (),
                compression: Some(3),
                key_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
                value_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
                replay_buffer: NZUsize!(DEFAULT_REPLAY_BUFFER),
                items_per_section: NZU64!(DEFAULT_ITEMS_PER_SECTION),
            };
            let mut archive = Archive::init(context.child("first"), cfg.clone())
                .await
                .expect("Failed to initialize archive");

            // Put the key-data pair
            let index = 1u64;
            let key = test_key("testkey");
            let data = 1;
            archive
                .put(index, key.clone(), data)
                .await
                .expect("Failed to put data");

            // Sync and drop the archive
            archive.sync().await.expect("Failed to sync archive");
            drop(archive);

            // Initialize the archive again without compression.
            // Index journal replay succeeds (no compression), but value reads will fail.
            let cfg = Config {
                translator: FourCap,
                key_partition: "test-index".into(),
                key_page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
                value_partition: "test-value".into(),
                codec_config: (),
                compression: None,
                key_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
                value_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
                replay_buffer: NZUsize!(DEFAULT_REPLAY_BUFFER),
                items_per_section: NZU64!(DEFAULT_ITEMS_PER_SECTION),
            };
            let archive =
                Archive::<_, _, FixedBytes<64>, i32>::init(context.child("second"), cfg.clone())
                    .await
                    .unwrap();

            // Getting the value should fail because compression settings mismatch.
            // Without compression, the codec sees extra bytes after decoding the value
            // (because the compressed data doesn't match the expected format).
            let result: Result<Option<i32>, _> = archive.get(Identifier::Index(index)).await;
            assert!(matches!(
                result,
                Err(Error::Journal(JournalError::Codec(CodecError::ExtraData(
                    _
                ))))
            ));
        });
    }

    #[test_traced]
    fn test_archive_overlapping_key_basic() {
        // Initialize the deterministic context
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Initialize the archive
            let cfg = Config {
                translator: FourCap,
                key_partition: "test-index".into(),
                key_page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
                value_partition: "test-value".into(),
                codec_config: (),
                compression: None,
                key_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
                value_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
                replay_buffer: NZUsize!(DEFAULT_REPLAY_BUFFER),
                items_per_section: NZU64!(DEFAULT_ITEMS_PER_SECTION),
            };
            let mut archive = Archive::init(context.child("storage"), cfg.clone())
                .await
                .expect("Failed to initialize archive");

            let index1 = 1u64;
            let key1 = test_key("keys1");
            let data1 = 1;
            let index2 = 2u64;
            let key2 = test_key("keys2");
            let data2 = 2;

            // Put the key-data pair
            archive
                .put(index1, key1.clone(), data1)
                .await
                .expect("Failed to put data");

            // Put the key-data pair
            archive
                .put(index2, key2.clone(), data2)
                .await
                .expect("Failed to put data");

            // Get the data back
            let retrieved = archive
                .get(Identifier::Key(&key1))
                .await
                .expect("Failed to get data")
                .expect("Data not found");
            assert_eq!(retrieved, data1);

            // Get the data back
            let retrieved = archive
                .get(Identifier::Key(&key2))
                .await
                .expect("Failed to get data")
                .expect("Data not found");
            assert_eq!(retrieved, data2);

            // Check metrics
            let buffer = context.encode();
            assert!(has_metric_value(&buffer, "items_tracked", 2));
            assert!(buffer.contains("unnecessary_reads_total 1"));
            assert!(buffer.contains("gets_total 2"));
        });
    }

    #[test_traced]
    fn test_archive_overlapping_key_multiple_sections() {
        // Initialize the deterministic context
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Initialize the archive
            let cfg = Config {
                translator: FourCap,
                key_partition: "test-index".into(),
                key_page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
                value_partition: "test-value".into(),
                codec_config: (),
                compression: None,
                key_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
                value_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
                replay_buffer: NZUsize!(DEFAULT_REPLAY_BUFFER),
                items_per_section: NZU64!(DEFAULT_ITEMS_PER_SECTION),
            };
            let mut archive = Archive::init(context.child("storage"), cfg.clone())
                .await
                .expect("Failed to initialize archive");

            let index1 = 1u64;
            let key1 = test_key("keys1");
            let data1 = 1;
            let index2 = 2_000_000u64;
            let key2 = test_key("keys2");
            let data2 = 2;

            // Put the key-data pair
            archive
                .put(index1, key1.clone(), data1)
                .await
                .expect("Failed to put data");

            // Put the key-data pair
            archive
                .put(index2, key2.clone(), data2)
                .await
                .expect("Failed to put data");

            // Get the data back
            let retrieved = archive
                .get(Identifier::Key(&key1))
                .await
                .expect("Failed to get data")
                .expect("Data not found");
            assert_eq!(retrieved, data1);

            // Get the data back
            let retrieved = archive
                .get(Identifier::Key(&key2))
                .await
                .expect("Failed to get data")
                .expect("Data not found");
            assert_eq!(retrieved, data2);
        });
    }

    #[test_traced]
    fn test_archive_prune_keys() {
        // Initialize the deterministic context
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Initialize the archive
            let cfg = Config {
                translator: FourCap,
                key_partition: "test-index".into(),
                key_page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
                value_partition: "test-value".into(),
                codec_config: (),
                compression: None,
                key_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
                value_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
                replay_buffer: NZUsize!(DEFAULT_REPLAY_BUFFER),
                items_per_section: NZU64!(1), // no mask - each item is its own section
            };
            let mut archive = Archive::init(context.child("storage"), cfg.clone())
                .await
                .expect("Failed to initialize archive");

            // Insert multiple keys across different sections
            let keys = vec![
                (1u64, test_key("key1-blah"), 1),
                (2u64, test_key("key2-blah"), 2),
                (3u64, test_key("key3-blah"), 3),
                (4u64, test_key("key3-bleh"), 3),
                (5u64, test_key("key4-blah"), 4),
            ];

            for (index, key, data) in &keys {
                archive
                    .put(*index, key.clone(), *data)
                    .await
                    .expect("Failed to put data");
            }

            // Check metrics
            let buffer = context.encode();
            assert!(has_metric_value(&buffer, "items_tracked", 5));

            // Prune sections less than 3
            archive.prune(3).await.expect("Failed to prune");

            // Ensure keys 1 and 2 are no longer present
            for (index, key, data) in keys {
                let retrieved = archive
                    .get(Identifier::Key(&key))
                    .await
                    .expect("Failed to get data");
                if index < 3 {
                    assert!(retrieved.is_none());
                } else {
                    assert_eq!(retrieved.expect("Data not found"), data);
                }
            }

            // Check metrics
            let buffer = context.encode();
            assert!(has_metric_value(&buffer, "items_tracked", 3));
            assert!(has_metric_value(&buffer, "indices_pruned_total", 2));
            assert!(has_metric_value(&buffer, "pruned_total", 0)); // no lazy cleanup yet

            // Try to prune older section
            archive.prune(2).await.expect("Failed to prune");

            // Try to prune current section again
            archive.prune(3).await.expect("Failed to prune");

            // Try to put older index
            let result = archive.put(1, test_key("key1-blah"), 1).await;
            assert!(matches!(result, Err(Error::AlreadyPrunedTo(3))));

            // Trigger lazy removal of keys
            archive
                .put(6, test_key("key2-blfh"), 5)
                .await
                .expect("Failed to put data");

            // Check metrics
            let buffer = context.encode();
            assert!(has_metric_value(&buffer, "items_tracked", 4)); // lazily remove one, add one
            assert!(has_metric_value(&buffer, "indices_pruned_total", 2));
            assert!(has_metric_value(&buffer, "pruned_total", 1));
        });
    }

    fn test_archive_keys_and_restart(num_keys: usize) -> String {
        // Initialize the deterministic context
        let executor = deterministic::Runner::default();
        executor.start(|mut context| async move {
            // Initialize the archive
            let items_per_section = 256u64;
            let cfg = Config {
                translator: TwoCap,
                key_partition: "test-index".into(),
                key_page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
                value_partition: "test-value".into(),
                codec_config: (),
                compression: None,
                key_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
                value_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
                replay_buffer: NZUsize!(DEFAULT_REPLAY_BUFFER),
                items_per_section: NZU64!(items_per_section),
            };
            let mut archive = Archive::init(
                context.child("init").with_attribute("index", 1),
                cfg.clone(),
            )
            .await
            .expect("Failed to initialize archive");

            // Insert multiple keys across different sections
            let mut keys = BTreeMap::new();
            while keys.len() < num_keys {
                let index = keys.len() as u64;
                let mut key = [0u8; 64];
                context.fill(&mut key);
                let key = FixedBytes::<64>::decode(key.as_ref()).unwrap();
                let mut data = [0u8; 1024];
                context.fill(&mut data);
                let data = FixedBytes::<1024>::decode(data.as_ref()).unwrap();

                archive
                    .put(index, key.clone(), data.clone())
                    .await
                    .expect("Failed to put data");
                keys.insert(key, (index, data));
            }

            // Ensure all keys can be retrieved
            for (key, (index, data)) in &keys {
                let retrieved = archive
                    .get(Identifier::Index(*index))
                    .await
                    .expect("Failed to get data")
                    .expect("Data not found");
                assert_eq!(&retrieved, data);
                let retrieved = archive
                    .get(Identifier::Key(key))
                    .await
                    .expect("Failed to get data")
                    .expect("Data not found");
                assert_eq!(&retrieved, data);
            }

            // Check metrics
            let buffer = context.encode();
            assert!(has_metric_value(&buffer, "items_tracked", num_keys));
            assert!(has_metric_value(&buffer, "pruned_total", 0));

            // Sync and drop the archive
            archive.sync().await.expect("Failed to sync archive");
            drop(archive);

            // Reinitialize the archive
            let cfg = Config {
                translator: TwoCap,
                key_partition: "test-index".into(),
                key_page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
                value_partition: "test-value".into(),
                codec_config: (),
                compression: None,
                key_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
                value_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
                replay_buffer: NZUsize!(DEFAULT_REPLAY_BUFFER),
                items_per_section: NZU64!(items_per_section),
            };
            let mut archive = Archive::<_, _, _, FixedBytes<1024>>::init(
                context.child("init").with_attribute("index", 2),
                cfg.clone(),
            )
            .await
            .expect("Failed to initialize archive");

            // Ensure all keys can be retrieved
            for (key, (index, data)) in &keys {
                let retrieved = archive
                    .get(Identifier::Index(*index))
                    .await
                    .expect("Failed to get data")
                    .expect("Data not found");
                assert_eq!(&retrieved, data);
                let retrieved = archive
                    .get(Identifier::Key(key))
                    .await
                    .expect("Failed to get data")
                    .expect("Data not found");
                assert_eq!(&retrieved, data);
            }

            // Prune first half
            let min = (keys.len() / 2) as u64;
            archive.prune(min).await.expect("Failed to prune");

            // Ensure all keys can be retrieved that haven't been pruned
            let min = (min / items_per_section) * items_per_section;
            let mut removed = 0;
            for (key, (index, data)) in keys {
                if index >= min {
                    let retrieved = archive
                        .get(Identifier::Key(&key))
                        .await
                        .expect("Failed to get data")
                        .expect("Data not found");
                    assert_eq!(retrieved, data);

                    // Check range
                    let (current_end, start_next) = archive.next_gap(index);
                    assert_eq!(current_end.unwrap(), num_keys as u64 - 1);
                    assert!(start_next.is_none());
                } else {
                    let retrieved = archive
                        .get(Identifier::Key(&key))
                        .await
                        .expect("Failed to get data");
                    assert!(retrieved.is_none());
                    removed += 1;

                    // Check range
                    let (current_end, start_next) = archive.next_gap(index);
                    assert!(current_end.is_none());
                    assert_eq!(start_next.unwrap(), min);
                }
            }

            // Check metrics
            let buffer = context.encode();
            assert!(has_metric_value(
                &buffer,
                "items_tracked",
                num_keys - removed
            ));
            assert!(has_metric_value(&buffer, "indices_pruned_total", removed));
            assert!(has_metric_value(&buffer, "pruned_total", 0)); // have not lazily removed keys yet

            context.auditor().state()
        })
    }

    #[test_group("slow")]
    #[test_traced]
    fn test_archive_many_keys_and_restart() {
        test_archive_keys_and_restart(100_000);
    }

    #[test_group("slow")]
    #[test_traced]
    fn test_determinism() {
        let state1 = test_archive_keys_and_restart(5_000);
        let state2 = test_archive_keys_and_restart(5_000);
        assert_eq!(state1, state2);
    }

    /// Regression: when the same key is stored at multiple indices and the
    /// earlier index is pruned, a subsequent `get`/`has` by key must resolve
    /// to the surviving, non-pruned entry rather than report the pruned one.
    /// Callers such as consensus's marshal cache rely on this to retain a
    /// reproposal of the same block at a later index even after the
    /// earlier index's retention window closes.
    #[test_traced]
    fn test_archive_key_lookup_skips_pruned_duplicates() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                translator: FourCap,
                key_partition: "test-index".into(),
                key_page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
                value_partition: "test-value".into(),
                codec_config: (),
                compression: None,
                key_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
                value_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
                replay_buffer: NZUsize!(DEFAULT_REPLAY_BUFFER),
                items_per_section: NZU64!(1),
            };
            let mut archive = Archive::init(context.child("storage"), cfg)
                .await
                .expect("Failed to initialize archive");

            // Same key stored at two different indices. Distinct values only
            // to make it observable which entry wins; a real caller would
            // store the same value (e.g. the same block) at both indices.
            let key = test_key("dupe-key");
            archive.put(2, key.clone(), 20).await.unwrap();
            archive.put(5, key.clone(), 50).await.unwrap();

            // Before pruning, either entry is a permitted answer per the
            // trait contract. The implementation happens to return the
            // earlier index, but we only assert a value is present.
            assert!(archive.get(Identifier::Key(&key)).await.unwrap().is_some());
            assert!(archive.has(Identifier::Key(&key)).await.unwrap());

            // Prune the earlier index (section 2). The later index must be
            // the sole surviving answer.
            archive.prune(3).await.unwrap();
            let got = archive.get(Identifier::Key(&key)).await.unwrap();
            assert_eq!(
                got,
                Some(50),
                "key lookup must skip the pruned entry and return the surviving one"
            );
            assert!(archive.has(Identifier::Key(&key)).await.unwrap());

            // Prune past the later index too — now nothing survives.
            archive.prune(6).await.unwrap();
            assert_eq!(archive.get(Identifier::Key(&key)).await.unwrap(), None);
            assert!(!archive.has(Identifier::Key(&key)).await.unwrap());
        });
    }

    #[test_traced]
    fn test_get_all_after_prune() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                translator: FourCap,
                key_partition: "test-index".into(),
                key_page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
                value_partition: "test-value".into(),
                codec_config: (),
                compression: None,
                key_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
                value_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
                replay_buffer: NZUsize!(DEFAULT_REPLAY_BUFFER),
                items_per_section: NZU64!(1),
            };
            let mut archive = Archive::init(context.child("storage"), cfg)
                .await
                .expect("Failed to initialize archive");

            archive.put_multi(1, test_key("aaa"), 10).await.unwrap();
            archive.put_multi(1, test_key("bbb"), 20).await.unwrap();
            archive.put_multi(3, test_key("ccc"), 30).await.unwrap();

            // Prune below index 3
            archive.prune(3).await.unwrap();

            // Pruned index returns None
            let all = archive.get_all(1).await.unwrap();
            assert_eq!(all, None);

            // Surviving index still works
            let all = archive.get_all(3).await.unwrap();
            assert_eq!(all, Some(vec![30]));
        });
    }

    #[test_traced]
    fn test_has_at() {
        let executor = deterministic::Runner::default();
        let (_, checkpoint) = executor.start_and_recover(|context| async move {
            let cfg = test_config(&context, NZU64!(2));
            let mut archive = Archive::init(context.child("storage"), cfg)
                .await
                .expect("Failed to initialize archive");

            // Vacant index
            assert!(!archive.has_at(1, &test_key("aaaa1")).await.unwrap());

            // Exact key at the index
            archive.put_multi(1, test_key("aaaa1"), 10).await.unwrap();
            assert!(archive.has_at(1, &test_key("aaaa1")).await.unwrap());

            // Same key is not reported at other indices
            assert!(!archive.has_at(2, &test_key("aaaa1")).await.unwrap());

            // A translated-key collision (FourCap shares the "aaaa" prefix)
            // must not produce a false positive
            assert!(!archive.has_at(1, &test_key("aaaa2")).await.unwrap());

            // A second entry at the same index is visible alongside the first
            archive.put_multi(1, test_key("aaaa2"), 20).await.unwrap();
            assert!(archive.has_at(1, &test_key("aaaa1")).await.unwrap());
            assert!(archive.has_at(1, &test_key("aaaa2")).await.unwrap());

            // A different key at an occupied index is absent
            assert!(!archive.has_at(1, &test_key("bbbb")).await.unwrap());

            archive.put_multi(3, test_key("cccc"), 30).await.unwrap();
            archive.sync().await.unwrap();
        });

        deterministic::Runner::from(checkpoint).start(|context| async move {
            let cfg = test_config(&context, NZU64!(2));
            let mut archive =
                Archive::<_, _, FixedBytes<64>, i32>::init(context.child("reopen"), cfg)
                    .await
                    .expect("Failed to reopen archive");

            // Replay rebuilds both entries at the shared index
            assert!(archive.has_at(1, &test_key("aaaa1")).await.unwrap());
            assert!(archive.has_at(1, &test_key("aaaa2")).await.unwrap());
            assert!(!archive.has_at(1, &test_key("bbbb")).await.unwrap());

            // Pruned indices report absent
            archive.prune(2).await.unwrap();
            assert!(!archive.has_at(1, &test_key("aaaa1")).await.unwrap());
            assert!(!archive.has_at(1, &test_key("aaaa2")).await.unwrap());
            assert!(archive.has_at(3, &test_key("cccc")).await.unwrap());

            archive.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_has_key() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_config(&context, NZU64!(2));
            let mut archive = Archive::init(context.child("storage"), cfg)
                .await
                .expect("Failed to initialize archive");

            // Absent key
            let key = test_key("aaaa1");
            assert!(!archive.has(Identifier::Key(&key)).await.unwrap());

            // Exact key
            archive.put(1, key.clone(), 10).await.unwrap();
            assert!(archive.has(Identifier::Key(&key)).await.unwrap());

            // A translated-key collision (FourCap shares the "aaaa" prefix)
            // must not produce a false positive
            let collision = test_key("aaaa2");
            assert!(!archive.has(Identifier::Key(&collision)).await.unwrap());
            archive.put(2, collision.clone(), 20).await.unwrap();
            assert!(archive.has(Identifier::Key(&collision)).await.unwrap());

            // Pruned keys report absent. Pruning is section-granular
            // (items_per_section = 2), so prune at a section boundary that
            // drops indices 1 and 2 while retaining index 4.
            archive.put(4, test_key("cccc"), 30).await.unwrap();
            archive.prune(4).await.unwrap();
            assert!(!archive.has(Identifier::Key(&key)).await.unwrap());
            assert!(!archive.has(Identifier::Key(&collision)).await.unwrap());
            assert!(archive
                .has(Identifier::Key(&test_key("cccc")))
                .await
                .unwrap());

            archive.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_put_multi_prune() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                translator: FourCap,
                key_partition: "test-index".into(),
                key_page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
                value_partition: "test-value".into(),
                codec_config: (),
                compression: None,
                key_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
                value_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
                replay_buffer: NZUsize!(DEFAULT_REPLAY_BUFFER),
                items_per_section: NZU64!(1),
            };
            let mut archive = Archive::init(context.child("storage"), cfg)
                .await
                .expect("Failed to initialize archive");

            // Two items at index 1, one at index 3
            archive.put_multi(1, test_key("aaa"), 10).await.unwrap();
            archive.put_multi(1, test_key("bbb"), 20).await.unwrap();
            archive.put_multi(3, test_key("ccc"), 30).await.unwrap();

            let buffer = context.encode();
            assert!(has_metric_value(&buffer, "items_tracked", 2));

            // Prune below index 3
            archive.prune(3).await.unwrap();

            // Both items at index 1 are gone
            assert_eq!(
                archive
                    .get(Identifier::Key(&test_key("aaa")))
                    .await
                    .unwrap(),
                None
            );
            assert_eq!(
                archive
                    .get(Identifier::Key(&test_key("bbb")))
                    .await
                    .unwrap(),
                None
            );

            // Item at index 3 survives
            assert_eq!(
                archive
                    .get(Identifier::Key(&test_key("ccc")))
                    .await
                    .unwrap(),
                Some(30)
            );

            let buffer = context.encode();
            assert!(has_metric_value(&buffer, "items_tracked", 1));
            assert!(has_metric_value(&buffer, "indices_pruned_total", 1));

            // put_multi below pruned index is rejected
            let result = archive.put_multi(2, test_key("ddd"), 40).await;
            assert!(matches!(result, Err(Error::AlreadyPrunedTo(3))));
        });
    }
}