commonware-storage 2026.9.0

use super::{Config, Error, Identifier};
use crate::{
    Context,
    journal::segmented::oversized::{
        Config as OversizedConfig, Oversized, Record as OversizedRecord,
    },
};
use commonware_codec::{CodecShared, FixedArray, FixedSize, Read, ReadExt, Write as CodecWrite};
use commonware_cryptography::{Crc32, Hasher, crc32};
use commonware_runtime::{
    Blob, Buf, BufMut, BufferPooler, IoBuf, ReadOptions, WriteOptions, buffer,
    iobuf::EncodeExt,
    telemetry::metrics::{Counter, MetricsExt as _},
};
use commonware_utils::{Array, Span};
use futures::future::try_join;
use std::{cmp::Ordering, collections::BTreeSet, num::NonZeroUsize, ops::Deref};
use tracing::debug;

/// The percentage of table entries that must reach `table_resize_frequency`
/// before a resize is triggered.
const RESIZE_THRESHOLD: u64 = 50;

/// Location of an item in the [Freezer].
///
/// This can be used to directly access the data for a given
/// key-value pair (rather than walking the journal chain).
#[derive(Hash, PartialEq, Eq, PartialOrd, Ord, Clone, Copy, FixedArray)]
#[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))]
#[repr(transparent)]
pub struct Cursor([u8; u64::SIZE + u64::SIZE + u32::SIZE]);

impl Cursor {
    /// Create a new [Cursor].
    fn new(section: u64, offset: u64, size: u32) -> Self {
        let mut buf = [0u8; u64::SIZE + u64::SIZE + u32::SIZE];
        buf[..u64::SIZE].copy_from_slice(&section.to_be_bytes());
        buf[u64::SIZE..u64::SIZE + u64::SIZE].copy_from_slice(&offset.to_be_bytes());
        buf[u64::SIZE + u64::SIZE..].copy_from_slice(&size.to_be_bytes());
        Self(buf)
    }

    /// Get the section of the cursor.
    fn section(&self) -> u64 {
        u64::from_be_bytes(self.0[..u64::SIZE].try_into().unwrap())
    }

    /// Get the offset of the cursor.
    fn offset(&self) -> u64 {
        u64::from_be_bytes(self.0[u64::SIZE..u64::SIZE + u64::SIZE].try_into().unwrap())
    }

    /// Get the size of the value.
    fn size(&self) -> u32 {
        u32::from_be_bytes(self.0[u64::SIZE + u64::SIZE..].try_into().unwrap())
    }
}

impl Read for Cursor {
    type Cfg = ();

    fn read_cfg(buf: &mut impl Buf, _: &Self::Cfg) -> Result<Self, commonware_codec::Error> {
        <[u8; u64::SIZE + u64::SIZE + u32::SIZE]>::read(buf).map(Self)
    }
}

impl CodecWrite for Cursor {
    fn write(&self, buf: &mut impl BufMut) {
        self.0.write(buf);
    }
}

impl FixedSize for Cursor {
    const SIZE: usize = u64::SIZE + u64::SIZE + u32::SIZE;
}

impl Span for Cursor {}

impl Array for Cursor {}

impl Deref for Cursor {
    type Target = [u8];
    fn deref(&self) -> &Self::Target {
        &self.0
    }
}

impl AsRef<[u8]> for Cursor {
    fn as_ref(&self) -> &[u8] {
        &self.0
    }
}

impl std::fmt::Debug for Cursor {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "Cursor(section={}, offset={}, size={})",
            self.section(),
            self.offset(),
            self.size()
        )
    }
}

impl std::fmt::Display for Cursor {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "Cursor(section={}, offset={}, size={})",
            self.section(),
            self.offset(),
            self.size()
        )
    }
}

/// Marker of [Freezer] progress.
///
/// This can be used to restore the [Freezer] to a consistent
/// state after shutdown.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Copy)]
#[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))]
pub struct Checkpoint {
    /// The epoch of the last committed operation.
    epoch: u64,
    /// The section of the last committed operation.
    section: u64,
    /// The size of the oversized index journal in the last committed section.
    oversized_size: u64,
    /// The size of the table.
    table_size: u32,
}

impl Checkpoint {
    /// Initialize a new [Checkpoint].
    const fn init(table_size: u32) -> Self {
        Self {
            table_size,
            epoch: 0,
            section: 0,
            oversized_size: 0,
        }
    }

    /// Return true if this checkpoint represents a fresh [Freezer].
    const fn is_empty(&self) -> bool {
        self.epoch == 0 && self.section == 0 && self.oversized_size == 0 && self.table_size == 0
    }
}

impl Read for Checkpoint {
    type Cfg = ();
    fn read_cfg(buf: &mut impl Buf, _: &()) -> Result<Self, commonware_codec::Error> {
        let epoch = u64::read(buf)?;
        let section = u64::read(buf)?;
        let oversized_size = u64::read(buf)?;
        let table_size = u32::read(buf)?;
        Ok(Self {
            epoch,
            section,
            oversized_size,
            table_size,
        })
    }
}

impl CodecWrite for Checkpoint {
    fn write(&self, buf: &mut impl BufMut) {
        self.epoch.write(buf);
        self.section.write(buf);
        self.oversized_size.write(buf);
        self.table_size.write(buf);
    }
}

impl FixedSize for Checkpoint {
    const SIZE: usize = u64::SIZE + u64::SIZE + u64::SIZE + u32::SIZE;
}

/// Name of the table blob.
const TABLE_BLOB_NAME: &[u8] = b"table";

/// Single table entry stored in the table blob.
#[derive(Debug, Clone, PartialEq)]
#[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))]
struct Entry {
    // Epoch in which this slot was written
    epoch: u64,
    // Section in which this slot was written
    section: u64,
    // Position in the key index for this section
    position: u64,
    // Number of items added to this entry since last resize
    added: u8,
    // CRC of (epoch | section | position | added)
    crc: u32,
}

impl Entry {
    /// The full size of a table entry (2 slots).
    const FULL_SIZE: usize = Self::SIZE * 2;

    /// Compute a checksum for [Entry].
    fn compute_crc(epoch: u64, section: u64, position: u64, added: u8) -> u32 {
        Crc32::hash(&[
            &epoch.to_be_bytes(),
            &section.to_be_bytes(),
            &position.to_be_bytes(),
            &added.to_be_bytes(),
        ])
        .as_u32()
    }

    /// Create a new [Entry].
    fn new(epoch: u64, section: u64, position: u64, added: u8) -> Self {
        Self {
            epoch,
            section,
            position,
            added,
            crc: Self::compute_crc(epoch, section, position, added),
        }
    }

    /// Create a new empty [Entry].
    const fn new_empty() -> Self {
        Self {
            epoch: 0,
            section: 0,
            position: 0,
            added: 0,
            crc: 0,
        }
    }

    /// Check if this entry is empty (all zeros).
    const fn is_empty(&self) -> bool {
        self.epoch == 0
            && self.section == 0
            && self.position == 0
            && self.added == 0
            && self.crc == 0
    }

    /// Check if this entry is valid.
    ///
    /// An empty entry does not have a valid checksum and is treated as invalid by this function.
    fn is_valid(&self) -> bool {
        Self::compute_crc(self.epoch, self.section, self.position, self.added) == self.crc
    }
}

impl FixedSize for Entry {
    const SIZE: usize = u64::SIZE + u64::SIZE + u64::SIZE + u8::SIZE + crc32::Digest::SIZE;
}

impl CodecWrite for Entry {
    fn write(&self, buf: &mut impl BufMut) {
        self.epoch.write(buf);
        self.section.write(buf);
        self.position.write(buf);
        self.added.write(buf);
        self.crc.write(buf);
    }
}

impl Read for Entry {
    type Cfg = ();
    fn read_cfg(buf: &mut impl Buf, _: &Self::Cfg) -> Result<Self, commonware_codec::Error> {
        let epoch = u64::read(buf)?;
        let section = u64::read(buf)?;
        let position = u64::read(buf)?;
        let added = u8::read(buf)?;
        let crc = u32::read(buf)?;

        Ok(Self {
            epoch,
            section,
            position,
            added,
            crc,
        })
    }
}

/// Sentinel value indicating no next entry in the collision chain.
const NO_NEXT_SECTION: u64 = u64::MAX;
const NO_NEXT_POSITION: u64 = u64::MAX;

/// Key entry stored in the segmented/fixed key index journal.
///
/// All fields are fixed size, enabling efficient collision chain traversal
/// without reading large values.
///
/// The `next` pointer uses sentinel values (u64::MAX, u64::MAX) to indicate
/// "no next entry" instead of Option, ensuring fixed-size encoding.
#[derive(Debug, Clone, PartialEq)]
struct Record<K: Array> {
    /// The key for this entry.
    key: K,
    /// Pointer to next entry in collision chain (section, position in key index).
    /// Uses (u64::MAX, u64::MAX) as sentinel for "no next".
    next_section: u64,
    next_position: u64,
    /// Byte offset in value journal (same section).
    value_offset: u64,
    /// Size of value data in the value journal.
    value_size: u32,
}

impl<K: Array> Record<K> {
    /// Create a new [Record].
    fn new(key: K, next: Option<(u64, u64)>, value_offset: u64, value_size: u32) -> Self {
        let (next_section, next_position) = next.unwrap_or((NO_NEXT_SECTION, NO_NEXT_POSITION));
        Self {
            key,
            next_section,
            next_position,
            value_offset,
            value_size,
        }
    }

    /// Get the next entry in the collision chain, if any.
    const fn next(&self) -> Option<(u64, u64)> {
        if self.next_section == NO_NEXT_SECTION && self.next_position == NO_NEXT_POSITION {
            None
        } else {
            Some((self.next_section, self.next_position))
        }
    }
}

impl<K: Array> CodecWrite for Record<K> {
    fn write(&self, buf: &mut impl BufMut) {
        self.key.write(buf);
        self.next_section.write(buf);
        self.next_position.write(buf);
        self.value_offset.write(buf);
        self.value_size.write(buf);
    }
}

impl<K: Array> Read for Record<K> {
    type Cfg = ();
    fn read_cfg(buf: &mut impl Buf, _: &Self::Cfg) -> Result<Self, commonware_codec::Error> {
        let key = K::read(buf)?;
        let next_section = u64::read(buf)?;
        let next_position = u64::read(buf)?;
        let value_offset = u64::read(buf)?;
        let value_size = u32::read(buf)?;

        Ok(Self {
            key,
            next_section,
            next_position,
            value_offset,
            value_size,
        })
    }
}

impl<K: Array> FixedSize for Record<K> {
    // key + next_section + next_position + value_offset + value_size
    const SIZE: usize = K::SIZE + u64::SIZE + u64::SIZE + u64::SIZE + u32::SIZE;
}

impl<K: Array> OversizedRecord for Record<K> {
    fn value_location(&self) -> (u64, u32) {
        (self.value_offset, self.value_size)
    }

    fn with_location(mut self, offset: u64, size: u32) -> Self {
        self.value_offset = offset;
        self.value_size = size;
        self
    }
}

#[cfg(feature = "arbitrary")]
impl<K: Array> arbitrary::Arbitrary<'_> for Record<K>
where
    K: for<'a> arbitrary::Arbitrary<'a>,
{
    fn arbitrary(u: &mut arbitrary::Unstructured<'_>) -> arbitrary::Result<Self> {
        Ok(Self {
            key: K::arbitrary(u)?,
            next_section: u64::arbitrary(u)?,
            next_position: u64::arbitrary(u)?,
            value_offset: u64::arbitrary(u)?,
            value_size: u32::arbitrary(u)?,
        })
    }
}

/// The freezer's state, boxed so the public [Freezer] handle stays pointer-sized.
struct Inner<E: Context, K: Array, V: CodecShared> {
    // Context for storage operations
    context: E,

    // Table configuration
    table_partition: String,
    table_size: u32,
    table_resize_threshold: u64,
    table_resize_frequency: u8,
    table_resize_chunk_size: u32,

    // Table blob that maps slots to key index chain heads
    table: E::Blob,

    // Combined key index + value storage with crash recovery
    oversized: Oversized<E, Record<K>, V>,

    // Target size for value blob sections
    blob_target_size: u64,

    // Current section for new writes
    current_section: u64,
    next_epoch: u64,

    // Sections with pending table updates to be synced
    modified_sections: BTreeSet<u64>,
    resizable: u32,
    resize_progress: Option<u32>,

    // Metrics
    puts: Counter,
    gets: Counter,
    has: Counter,
    unnecessary_reads: Counter,
    unnecessary_writes: Counter,
    resizes: Counter,
}

impl<E: Context, K: Array, V: CodecShared> Inner<E, K, V> {
    /// Calculate the byte offset for a table index.
    #[inline]
    const fn table_offset(table_index: u32) -> u64 {
        table_index as u64 * Entry::FULL_SIZE as u64
    }

    /// Parse table entries from a buffer.
    fn parse_entries(mut buf: impl Buf) -> Result<(Entry, Entry), Error> {
        let entry1 = Entry::read(&mut buf)?;
        let entry2 = Entry::read(&mut buf)?;
        Ok((entry1, entry2))
    }

    /// Read entries from the table blob.
    async fn read_table(blob: &E::Blob, table_index: u32) -> Result<(Entry, Entry), Error> {
        let offset = Self::table_offset(table_index);
        let read_buf = blob
            .read_at(offset, Entry::FULL_SIZE, ReadOptions::default())
            .await?;

        Self::parse_entries(read_buf)
    }

    /// Recover a single table entry and update tracking.
    async fn recover_entry(
        blob: &E::Blob,
        entry: &mut Entry,
        entry_offset: u64,
        max_valid_epoch: Option<u64>,
        max_epoch: &mut u64,
        max_section: &mut u64,
    ) -> Result<bool, Error> {
        if entry.is_empty() {
            return Ok(false);
        }

        if !entry.is_valid()
            || (max_valid_epoch.is_some() && entry.epoch > max_valid_epoch.unwrap())
        {
            debug!(
                valid_epoch = max_valid_epoch,
                entry_epoch = entry.epoch,
                "found invalid table entry"
            );
            *entry = Entry::new_empty();
            let zero_buf = IoBuf::from(&[0u8; Entry::SIZE]);
            blob.write_at(entry_offset, zero_buf, WriteOptions::default())
                .await?;
            Ok(true)
        } else if max_valid_epoch.is_none() && entry.epoch > *max_epoch {
            // Only track max epoch if we're discovering it (not validating against a known epoch)
            *max_epoch = entry.epoch;
            *max_section = entry.section;
            Ok(false)
        } else {
            Ok(false)
        }
    }

    /// Validate and clean invalid table entries for a given epoch.
    ///
    /// Returns (modified, max_epoch, max_section, resizable) where:
    /// - modified: whether any entries were cleaned
    /// - max_epoch: the maximum valid epoch found
    /// - max_section: the section corresponding to `max_epoch`
    /// - resizable: the number of entries that can be resized
    async fn recover_table(
        pooler: &impl BufferPooler,
        blob: &E::Blob,
        table_size: u32,
        table_resize_frequency: u8,
        max_valid_epoch: Option<u64>,
        table_replay_buffer: NonZeroUsize,
    ) -> Result<(bool, u64, u64, u32), Error> {
        // Create a buffered reader for efficient scanning
        let blob_size = Self::table_offset(table_size);
        let mut reader =
            buffer::Read::from_pooler(pooler, blob.clone(), blob_size, table_replay_buffer);

        // Iterate over all table entries and overwrite invalid ones
        let mut modified = false;
        let mut max_epoch = 0u64;
        let mut max_section = 0u64;
        let mut resizable = 0u32;
        for table_index in 0..table_size {
            let offset = Self::table_offset(table_index);

            // Read both entries from the buffer.
            let entry_buf = reader.read(Entry::FULL_SIZE).await?;
            let (mut entry1, mut entry2) = Self::parse_entries(entry_buf)?;

            // Check both entries
            let entry1_cleared = Self::recover_entry(
                blob,
                &mut entry1,
                offset,
                max_valid_epoch,
                &mut max_epoch,
                &mut max_section,
            )
            .await?;
            let entry2_cleared = Self::recover_entry(
                blob,
                &mut entry2,
                offset + Entry::SIZE as u64,
                max_valid_epoch,
                &mut max_epoch,
                &mut max_section,
            )
            .await?;
            modified |= entry1_cleared || entry2_cleared;

            // If the latest entry has reached the resize frequency, increment the resizable entries
            if let Some((_, _, added)) = Self::read_latest_entry(&entry1, &entry2)
                && added >= table_resize_frequency
            {
                resizable += 1;
            }
        }

        Ok((modified, max_epoch, max_section, resizable))
    }

    /// Determine the write offset for a table entry based on current entries and epoch.
    const fn compute_write_offset(entry1: &Entry, entry2: &Entry, epoch: u64) -> u64 {
        // If either entry matches the current epoch, overwrite it
        if !entry1.is_empty() && entry1.epoch == epoch {
            return 0;
        }
        if !entry2.is_empty() && entry2.epoch == epoch {
            return Entry::SIZE as u64;
        }

        // Otherwise, write to the older slot (or empty slot)
        match (entry1.is_empty(), entry2.is_empty()) {
            (true, _) => 0,                  // First slot is empty
            (_, true) => Entry::SIZE as u64, // Second slot is empty
            (false, false) => {
                if entry1.epoch < entry2.epoch {
                    0
                } else {
                    Entry::SIZE as u64
                }
            }
        }
    }

    /// Read the latest valid entry from two table slots.
    fn read_latest_entry(entry1: &Entry, entry2: &Entry) -> Option<(u64, u64, u8)> {
        match (
            !entry1.is_empty() && entry1.is_valid(),
            !entry2.is_empty() && entry2.is_valid(),
        ) {
            (true, true) => match entry1.epoch.cmp(&entry2.epoch) {
                Ordering::Greater => Some((entry1.section, entry1.position, entry1.added)),
                Ordering::Less => Some((entry2.section, entry2.position, entry2.added)),
                Ordering::Equal => {
                    unreachable!("two valid entries with the same epoch")
                }
            },
            (true, false) => Some((entry1.section, entry1.position, entry1.added)),
            (false, true) => Some((entry2.section, entry2.position, entry2.added)),
            (false, false) => None,
        }
    }

    /// Write a table entry to the appropriate slot based on epoch.
    async fn update_head(
        pooler: &impl BufferPooler,
        table: &E::Blob,
        table_index: u32,
        entry1: &Entry,
        entry2: &Entry,
        update: Entry,
    ) -> Result<(), Error> {
        // Calculate the base offset for this table index
        let table_offset = Self::table_offset(table_index);

        // Determine which slot to write to based on the provided entries
        let start = Self::compute_write_offset(entry1, entry2, update.epoch);

        // Write the new entry
        table
            .write_at(
                table_offset + start,
                update.encode_with_pool_mut(pooler.storage_buffer_pool()),
                WriteOptions::default(),
            )
            .await
            .map_err(Error::Runtime)
    }

    /// Initialize table with given size and sync.
    async fn init_table(blob: &E::Blob, table_size: u32) -> Result<(), Error> {
        let table_len = Self::table_offset(table_size);
        blob.resize(table_len).await?;
        blob.sync().await?;
        Ok(())
    }

    /// See [Freezer::init].
    async fn init(
        context: E,
        config: Config<V::Cfg>,
        checkpoint: Option<Checkpoint>,
    ) -> Result<Self, Error> {
        // Validate that initial_table_size is a power of 2
        assert!(
            config.table_initial_size > 0 && config.table_initial_size.is_power_of_two(),
            "table_initial_size must be a power of 2"
        );

        // A missing or empty checkpoint starts fresh: delete all existing freezer data
        let reset = checkpoint.is_none_or(|checkpoint| checkpoint.is_empty());
        if reset {
            for partition in [
                &config.key_partition,
                &config.value_partition,
                &config.table_partition,
            ] {
                match context.remove(partition, None).await {
                    Ok(()) | Err(commonware_runtime::Error::PartitionMissing(_)) => {}
                    Err(err) => return Err(Error::Runtime(err)),
                }
            }
        }

        // Initialize oversized journal. A checkpoint is only published after the
        // oversized journal is durably synced (see Self::sync), so recovery restores
        // exactly the checkpointed state: committed data it covers cannot be silently
        // repaired away, and anything beyond it is discarded.
        let oversized_cfg = OversizedConfig {
            index_partition: config.key_partition.clone(),
            value_partition: config.value_partition.clone(),
            index_page_cache: config.key_page_cache.clone(),
            index_write_buffer: config.key_write_buffer,
            value_write_buffer: config.value_write_buffer,
            replay_buffer: config.table_replay_buffer,
            compression: config.value_compression,
            codec_config: config.codec_config,
        };
        let oversized_context = context.child("oversized");
        let oversized: Oversized<E, Record<K>, V> = match checkpoint
            .filter(|checkpoint| !checkpoint.is_empty())
            .map(|checkpoint| (checkpoint.section, checkpoint.oversized_size))
        {
            Some(checkpoint) => {
                Oversized::init_with_checkpoint(oversized_context, oversized_cfg, checkpoint)
                    .await?
            }
            None => Oversized::init(oversized_context, oversized_cfg).await?,
        };

        // Open table blob
        let (table, table_len) = context
            .open(&config.table_partition, TABLE_BLOB_NAME)
            .await?;

        // Determine checkpoint based on initialization scenario
        let (checkpoint, resizable) = match checkpoint {
            // Non-empty checkpoint: align existing data to it
            Some(checkpoint) if !checkpoint.is_empty() => {
                // A non-empty checkpoint against an empty table references data that does not exist
                if table_len == 0 {
                    return Err(Error::CheckpointMismatch);
                }
                assert!(
                    checkpoint.table_size > 0 && checkpoint.table_size.is_power_of_two(),
                    "table_size must be a power of 2"
                );

                // Resize the table if needed. Growing is never valid: the checkpoint publishes
                // only after the table sync completes, so a shorter table cannot back the
                // checkpointed entries, and zero-extending would fabricate empty heads.
                let expected_table_len = Self::table_offset(checkpoint.table_size);
                if table_len < expected_table_len {
                    return Err(Error::CheckpointMismatch);
                }
                let mut modified = if table_len != expected_table_len {
                    table.resize(expected_table_len).await?;
                    true
                } else {
                    false
                };

                // Validate and clean invalid entries
                let (table_modified, _, _, resizable) = Self::recover_table(
                    &context,
                    &table,
                    checkpoint.table_size,
                    config.table_resize_frequency,
                    Some(checkpoint.epoch),
                    config.table_replay_buffer,
                )
                .await?;
                if table_modified {
                    modified = true;
                }

                // Sync table if needed
                if modified {
                    table.sync().await?;
                }

                (checkpoint, resizable)
            }

            // Missing or empty checkpoint: reset wiped any existing data, so initialize a new table
            _ => {
                Self::init_table(&table, config.table_initial_size).await?;
                (Checkpoint::init(config.table_initial_size), 0)
            }
        };

        // Create metrics
        let puts = context.counter("puts", "number of put operations");
        let gets = context.counter("gets", "number of get operations");
        let has = context.counter("has", "number of has operations");
        let unnecessary_reads = context.counter(
            "unnecessary_reads",
            "number of unnecessary reads performed during key lookups",
        );
        let unnecessary_writes = context.counter(
            "unnecessary_writes",
            "number of unnecessary writes performed during resize",
        );
        let resizes = context.counter("resizes", "number of table resizing operations");

        Ok(Self {
            context,
            table_partition: config.table_partition,
            table_size: checkpoint.table_size,
            table_resize_threshold: checkpoint.table_size as u64 * RESIZE_THRESHOLD / 100,
            table_resize_frequency: config.table_resize_frequency,
            table_resize_chunk_size: config.table_resize_chunk_size,
            table,
            oversized,
            blob_target_size: config.value_target_size,
            current_section: checkpoint.section,
            next_epoch: checkpoint.epoch.checked_add(1).expect("epoch overflow"),
            modified_sections: BTreeSet::new(),
            resizable,
            resize_progress: None,
            puts,
            gets,
            has,
            unnecessary_reads,
            unnecessary_writes,
            resizes,
        })
    }

    /// Compute the table index for a given key.
    ///
    /// As the table doubles in size during a resize, each existing entry splits into two:
    /// one at the original index and another at a new index (original index + previous table size).
    ///
    /// For example, with an initial table size of 4 (2^2):
    /// - Initially: uses 2 bits of the hash, mapping to entries 0, 1, 2, 3.
    /// - After resizing to 8: uses 3 bits, entry 0 splits into indices 0 and 4.
    /// - After resizing to 16: uses 4 bits, entry 0 splits into indices 0 and 8, and so on.
    ///
    /// To determine the appropriate entry, we AND the key's hash with the current table size.
    fn table_index(&self, key: &K) -> u32 {
        let hash = Crc32::checksum(key.as_ref());
        hash & (self.table_size - 1)
    }

    /// Determine if the table should be resized.
    const fn should_resize(&self) -> bool {
        self.resizable as u64 >= self.table_resize_threshold
    }

    /// Determine which blob section to write to based on current blob size.
    async fn update_section(&mut self) -> Result<(), Error> {
        // Get the current value blob section size
        let value_size = self.oversized.value_size(self.current_section).await?;

        // If the current section has reached the target size, create a new section
        if value_size >= self.blob_target_size {
            self.current_section += 1;
            debug!(
                size = value_size,
                section = self.current_section,
                "updated section"
            );
        }

        Ok(())
    }

    /// See [Freezer::put].
    async fn put(mut self: Box<Self>, key: K, value: V) -> Result<(Box<Self>, Cursor), Error> {
        self.puts.inc();

        // Update the section if needed
        self.update_section().await?;

        // Get head of the chain from table
        let table_index = self.table_index(&key);
        let (entry1, entry2) = Self::read_table(&self.table, table_index).await?;
        let head = Self::read_latest_entry(&entry1, &entry2);

        // Create key entry with pointer to previous head (value location set by oversized.append)
        let key_entry = Record::new(
            key,
            head.map(|(section, position, _)| (section, position)),
            0,
            0,
        );

        // Write value and key entry (glob first, then index)
        let (position, value_offset, value_size);
        (self.oversized, position, value_offset, value_size) = self
            .oversized
            .append(self.current_section, key_entry, &value)
            .await?;

        // Update the number of items added to the entry.
        //
        // We use `saturating_add` to handle overflow (when the table is at max size) gracefully.
        let mut added = head.map(|(_, _, added)| added).unwrap_or(0);
        added = added.saturating_add(1);

        // If we've reached the threshold for resizing, increment the resizable entries
        if added == self.table_resize_frequency {
            self.resizable += 1;
        }

        // Update the old position
        self.modified_sections.insert(self.current_section);
        let new_entry = Entry::new(self.next_epoch, self.current_section, position, added);
        Self::update_head(
            &self.context,
            &self.table,
            table_index,
            &entry1,
            &entry2,
            new_entry,
        )
        .await?;

        // If we're mid-resize and this entry has already been processed, update the new position too
        if let Some(resize_progress) = self.resize_progress
            && table_index < resize_progress
        {
            self.unnecessary_writes.inc();

            // If the previous entry crossed the threshold, so did this one
            if added == self.table_resize_frequency {
                self.resizable += 1;
            }

            // This entry has been processed, so we need to update the new position as well.
            //
            // The entries are still identical to the old ones, so we don't need to read them again.
            let new_table_index = self.table_size + table_index;
            let new_entry = Entry::new(self.next_epoch, self.current_section, position, added);
            Self::update_head(
                &self.context,
                &self.table,
                new_table_index,
                &entry1,
                &entry2,
                new_entry,
            )
            .await?;
        }

        let cursor = Cursor::new(self.current_section, value_offset, value_size);
        Ok((self, cursor))
    }

    /// Get the value for a given [Cursor].
    async fn get_cursor(&self, cursor: Cursor) -> Result<V, Error> {
        let value = self
            .oversized
            .get_value(cursor.section(), cursor.offset(), cursor.size())
            .await?;

        Ok(value)
    }

    /// Find the first key entry matching `key`, returning it with its section.
    ///
    /// Reads key entries only, never values.
    async fn find_key(&self, key: &K) -> Result<Option<(u64, Record<K>)>, Error> {
        // Get head of the chain from table
        let table_index = self.table_index(key);
        let (entry1, entry2) = Self::read_table(&self.table, table_index).await?;
        let Some((mut section, mut position, _)) = Self::read_latest_entry(&entry1, &entry2) else {
            return Ok(None);
        };

        // Follow the linked list chain to find the first matching key
        loop {
            // Get the key entry from the fixed key index (efficient, good cache locality)
            let key_entry = self.oversized.get(section, position).await?;

            // Check if this key matches
            if key_entry.key.as_ref() == key.as_ref() {
                return Ok(Some((section, key_entry)));
            }

            // Increment unnecessary reads
            self.unnecessary_reads.inc();

            // Follow the chain
            let Some(next) = key_entry.next() else {
                break; // End of chain
            };
            section = next.0;
            position = next.1;
        }

        Ok(None)
    }

    /// Get the first value for a given key.
    async fn get_key(&self, key: &K) -> Result<Option<V>, Error> {
        self.gets.inc();

        let Some((section, key_entry)) = self.find_key(key).await? else {
            return Ok(None);
        };
        let value = self
            .oversized
            .get_value(section, key_entry.value_offset, key_entry.value_size)
            .await?;
        Ok(Some(value))
    }

    /// See [Freezer::get].
    async fn get<'a>(&'a self, identifier: Identifier<'a, K>) -> Result<Option<V>, Error> {
        match identifier {
            Identifier::Cursor(cursor) => self.get_cursor(cursor).await.map(Some),
            Identifier::Key(key) => self.get_key(key).await,
        }
    }

    /// See [Freezer::has].
    async fn has(&self, key: &K) -> Result<bool, Error> {
        self.has.inc();

        Ok(self.find_key(key).await?.is_some())
    }

    /// Resize the table by doubling its size and split each entry into two.
    async fn start_resize(&mut self) -> Result<(), Error> {
        self.resizes.inc();

        // Double the table size (if not already at the max size)
        let old_size = self.table_size;
        let Some(new_size) = old_size.checked_mul(2) else {
            return Ok(());
        };
        self.table.resize(Self::table_offset(new_size)).await?;

        // Start the resize
        self.resize_progress = Some(0);
        debug!(old = old_size, new = new_size, "table resize started");

        Ok(())
    }

    /// Write a pair of entries to a buffer, replacing one slot with the new entry.
    fn rewrite_entries(buf: &mut impl BufMut, entry1: &Entry, entry2: &Entry, new_entry: &Entry) {
        if Self::compute_write_offset(entry1, entry2, new_entry.epoch) == 0 {
            new_entry.write(buf);
            entry2.write(buf);
        } else {
            entry1.write(buf);
            new_entry.write(buf);
        }
    }

    /// Continue a resize operation by processing the next chunk of entries.
    ///
    /// This function processes `table_resize_chunk_size` entries at a time, allowing the resize to
    /// be spread across multiple sync operations to avoid latency spikes.
    async fn advance_resize(&mut self) -> Result<(), Error> {
        // Compute the range to update
        let current_index = self.resize_progress.unwrap();
        let old_size = self.table_size;
        let chunk_end = (current_index + self.table_resize_chunk_size).min(old_size);
        let chunk_size = chunk_end - current_index;

        // Read the entire chunk
        let chunk_bytes = chunk_size as usize * Entry::FULL_SIZE;
        let read_offset = Self::table_offset(current_index);
        let mut read_buf = self
            .table
            .read_at(read_offset, chunk_bytes, ReadOptions::default())
            .await?;

        // Process each entry in the chunk
        let mut writes = self.context.storage_buffer_pool().alloc(chunk_bytes);
        for _ in 0..chunk_size {
            // Parse the next two slots directly from the read stream.
            let (entry1, entry2) = Self::parse_entries(&mut read_buf)?;

            // Get the current head
            let head = Self::read_latest_entry(&entry1, &entry2);

            // Get the reset entry (may be empty)
            let reset_entry = match head {
                Some((section, position, added)) => {
                    // If the entry was at or over the threshold, decrement the resizable entries.
                    if added >= self.table_resize_frequency {
                        self.resizable -= 1;
                    }
                    Entry::new(self.next_epoch, section, position, 0)
                }
                None => Entry::new_empty(),
            };

            // Rewrite the entries
            Self::rewrite_entries(&mut writes, &entry1, &entry2, &reset_entry);
        }

        // Put the writes into the table.
        let writes = writes.freeze();
        let old_write = self
            .table
            .write_at(read_offset, writes.clone(), WriteOptions::default());
        let new_offset = (old_size as usize * Entry::FULL_SIZE) as u64 + read_offset;
        let new_write = self
            .table
            .write_at(new_offset, writes, WriteOptions::default());
        try_join(old_write, new_write).await?;

        // Update progress
        if chunk_end >= old_size {
            // Resize complete
            self.table_size = old_size * 2;
            self.table_resize_threshold = self.table_size as u64 * RESIZE_THRESHOLD / 100;
            self.resize_progress = None;
            debug!(
                old = old_size,
                new = self.table_size,
                "table resize completed"
            );
        } else {
            // More chunks to process
            self.resize_progress = Some(chunk_end);
            debug!(current = current_index, chunk_end, "table resize progress");
        }

        Ok(())
    }

    /// See [Freezer::sync].
    async fn sync(mut self: Box<Self>) -> Result<(Box<Self>, Checkpoint), Error> {
        // Sync all modified sections for oversized journal
        self.oversized = self.oversized.sync(&self.modified_sections).await?;
        self.modified_sections.clear();

        // Start a resize (if needed)
        if self.should_resize() && self.resize_progress.is_none() {
            self.start_resize().await?;
        }

        // Continue a resize (if ongoing)
        if self.resize_progress.is_some() {
            self.advance_resize().await?;
        }

        // Sync updated table entries
        self.table.sync().await?;
        let stored_epoch = self.next_epoch;
        self.next_epoch = self.next_epoch.checked_add(1).expect("epoch overflow");

        // Get size from oversized
        let oversized_size = self.oversized.size(self.current_section)?;

        let checkpoint = Checkpoint {
            epoch: stored_epoch,
            section: self.current_section,
            oversized_size,
            table_size: self.table_size,
        };
        Ok((self, checkpoint))
    }

    /// See [Freezer::close].
    async fn close(mut self: Box<Self>) -> Result<Checkpoint, Error> {
        // If we're mid-resize, complete it
        while self.resize_progress.is_some() {
            self.advance_resize().await?;
        }

        // Sync any pending updates before closing
        let (_, checkpoint) = self.sync().await?;

        Ok(checkpoint)
    }

    /// See [Freezer::destroy].
    async fn destroy(self) -> Result<(), Error> {
        // Destroy oversized journal
        self.oversized.destroy().await?;

        // Destroy the table
        drop(self.table);
        self.context
            .remove(&self.table_partition, Some(TABLE_BLOB_NAME))
            .await?;
        self.context.remove(&self.table_partition, None).await?;

        Ok(())
    }
}

/// Implementation of [Freezer].
///
/// Mutating functions consume the freezer and return it only on success: an error (or a dropped
/// future) destroys the handle.
pub struct Freezer<E: Context, K: Array, V: CodecShared>(Box<Inner<E, K, V>>);

impl<E: Context, K: Array, V: CodecShared> std::fmt::Debug for Freezer<E, K, V> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Freezer")
            .field("current_section", &self.0.current_section)
            .field("next_epoch", &self.0.next_epoch)
            .finish_non_exhaustive()
    }
}

impl<E: Context, K: Array, V: CodecShared> Freezer<E, K, V> {
    /// Initialize a [Freezer] instance, aligning existing data to a [Checkpoint] when provided.
    ///
    /// Passing `None` or an empty [Checkpoint] deletes any existing freezer data and starts empty.
    pub async fn init(
        context: E,
        config: Config<V::Cfg>,
        checkpoint: Option<Checkpoint>,
    ) -> Result<Self, Error> {
        Ok(Self(Box::new(
            Inner::init(context, config, checkpoint).await?,
        )))
    }

    /// Put a key-value pair into the [Freezer].
    /// If the key already exists, the value is updated.
    pub async fn put(mut self, key: K, value: V) -> Result<(Self, Cursor), Error> {
        let cursor;
        (self.0, cursor) = self.0.put(key, value).await?;
        Ok((self, cursor))
    }

    /// Get the value for a given [Identifier].
    ///
    /// If a [Cursor] is known for the required key, it
    /// is much faster to use it than searching for a `key`.
    pub async fn get<'a>(&'a self, identifier: Identifier<'a, K>) -> Result<Option<V>, Error> {
        self.0.get(identifier).await
    }

    /// Check whether a value exists for a given key.
    ///
    /// Walks the same key index chain as [`Self::get`] with [`Identifier::Key`]
    /// but never reads values.
    pub async fn has(&self, key: &K) -> Result<bool, Error> {
        self.0.has(key).await
    }

    /// Sync all pending data in [Freezer].
    ///
    /// If the table needs to be resized, the resize will begin during this sync.
    /// The resize operation is performed incrementally across multiple sync calls
    /// to avoid a large latency spike (or unexpected long latency for [Freezer::put]).
    /// Each sync will process up to `table_resize_chunk_size` entries until the resize
    /// is complete.
    pub async fn sync(mut self) -> Result<(Self, Checkpoint), Error> {
        let checkpoint;
        (self.0, checkpoint) = self.0.sync().await?;
        Ok((self, checkpoint))
    }

    /// Close the [Freezer] and return a [Checkpoint] for recovery.
    pub async fn close(self) -> Result<Checkpoint, Error> {
        self.0.close().await
    }

    /// Close and remove any underlying blobs created by the [Freezer].
    pub async fn destroy(self) -> Result<(), Error> {
        self.0.destroy().await
    }

    /// Get the current progress of the resize operation.
    ///
    /// Returns `None` if the [Freezer] is not resizing.
    #[cfg(test)]
    pub fn resizing(&self) -> Option<u32> {
        self.0.resize_progress
    }

    /// Get the number of resizable entries.
    #[cfg(test)]
    pub fn resizable(&self) -> u32 {
        self.0.resizable
    }

    /// Get the current size of the table.
    #[cfg(test)]
    pub fn table_size(&self) -> u32 {
        self.0.table_size
    }
}

#[cfg(all(test, feature = "arbitrary"))]
mod conformance {
    use super::*;
    use commonware_codec::conformance::CodecConformance;
    use commonware_utils::sequence::U64;

    commonware_conformance::conformance_tests! {
        CodecConformance<Cursor>,
        CodecConformance<Checkpoint>,
        CodecConformance<Entry>,
        CodecConformance<Record<U64>>
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use commonware_codec::DecodeExt;
    use commonware_macros::test_traced;
    use commonware_runtime::{
        Runner, Storage, Supervisor as _, WriteOptions, buffer::paged::CacheRef, deterministic,
        deterministic::Context,
    };
    use commonware_utils::{
        NZU16, NZUsize,
        sequence::{FixedBytes, U64},
    };

    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()
    }

    fn test_key_at_index(table_size: u32, table_index: u32) -> FixedBytes<64> {
        assert!(table_size.is_power_of_two());
        assert!(table_index < table_size);

        for value in 0u64.. {
            let mut buf = [0u8; 64];
            let bytes = value.to_be_bytes();
            buf[..bytes.len()].copy_from_slice(&bytes);
            let key = FixedBytes::new(buf);
            if Crc32::checksum(key.as_ref()) & (table_size - 1) == table_index {
                return key;
            }
        }

        unreachable!("u64 key space exhausted");
    }

    type TestFreezer = Freezer<Context, U64, u64>;

    fn is_send<T: Send>(_: T) {}

    #[allow(dead_code)]
    fn assert_freezer_futures_are_send(freezer: TestFreezer, key: U64) {
        is_send(freezer.get(Identifier::Key(&key)));
        is_send(freezer.put(key, 0u64));
    }

    #[allow(dead_code)]
    fn assert_freezer_destroy_is_send(freezer: TestFreezer) {
        is_send(freezer.destroy());
    }

    #[test_traced]
    fn issue_2966_regression() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = super::super::Config {
                key_partition: "test-key-index".into(),
                key_write_buffer: NZUsize!(1024),
                key_page_cache: CacheRef::from_pooler(&context, NZU16!(1024), NZUsize!(10)),
                value_partition: "test-value-journal".into(),
                value_compression: None,
                value_write_buffer: NZUsize!(1024),
                value_target_size: 10 * 1024 * 1024,
                table_partition: "test-table".into(),
                // Use 4 entries but only insert to 2, leaving 2 empty
                table_initial_size: 4,
                table_resize_frequency: 1,
                table_resize_chunk_size: 4,
                table_replay_buffer: NZUsize!(64 * 1024),
                codec_config: (),
            };
            let freezer =
                Freezer::<_, FixedBytes<64>, i32>::init(context.child("first"), cfg.clone(), None)
                    .await
                    .unwrap();

            // Insert only 2 keys to different entries. With table_size=4, entries 2 and 3
            // should remain empty.
            let (freezer, _) = freezer.put(test_key("key0"), 0).await.unwrap();
            let (freezer, _) = freezer.put(test_key("key2"), 1).await.unwrap();
            freezer.close().await.unwrap();

            let (blob, size) = context.open(&cfg.table_partition, b"table").await.unwrap();
            let table_data = blob
                .read_at(0, size as usize, ReadOptions::default())
                .await
                .unwrap()
                .coalesce();

            // Verify resize happened (table doubled from 4 to 8)
            let num_entries = size as usize / Entry::FULL_SIZE;
            assert_eq!(num_entries, 8);

            // Count entries where both slots are truly empty. The bug would cause empty
            // entries to have one slot with epoch != 0 and valid CRC.
            let mut both_empty_count = 0;
            for entry_idx in 0..num_entries {
                let offset = entry_idx * Entry::FULL_SIZE;
                let buf = &table_data.as_ref()[offset..offset + Entry::FULL_SIZE];
                let (slot0, slot1) =
                    Inner::<Context, FixedBytes<64>, i32>::parse_entries(buf).unwrap();
                if slot0.is_empty() && slot1.is_empty() {
                    both_empty_count += 1;
                }
            }
            // 2 keys in 4 entries = 2 empty. After resize to 8, those become 4 empty.
            assert_eq!(both_empty_count, 4);
        });
    }

    #[test_traced]
    fn issue_2955_regression() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = super::super::Config {
                key_partition: "test-key-index".into(),
                key_write_buffer: NZUsize!(1024),
                key_page_cache: CacheRef::from_pooler(&context, NZU16!(1024), NZUsize!(10)),
                value_partition: "test-value-journal".into(),
                value_compression: None,
                value_write_buffer: NZUsize!(1024),
                value_target_size: 10 * 1024 * 1024,
                table_partition: "test-table".into(),
                table_initial_size: 4,
                table_resize_frequency: 1,
                table_resize_chunk_size: 4,
                table_replay_buffer: NZUsize!(64 * 1024),
                codec_config: (),
            };

            // Create freezer with data
            let checkpoint = {
                let freezer = Freezer::<_, FixedBytes<64>, i32>::init(
                    context.child("first"),
                    cfg.clone(),
                    None,
                )
                .await
                .unwrap();
                let (freezer, _) = freezer.put(test_key("key0"), 42).await.unwrap();
                let (freezer, _) = freezer.sync().await.unwrap();
                freezer.close().await.unwrap()
            };

            // Corrupt the CRC in both slots of the table entry
            {
                let (blob, _) = context.open(&cfg.table_partition, b"table").await.unwrap();
                let entry_data = blob
                    .read_at(0, Entry::FULL_SIZE, ReadOptions::default())
                    .await
                    .unwrap();
                let mut corrupted = entry_data.coalesce();
                // Corrupt CRC of first slot (last 4 bytes of first slot)
                corrupted.as_mut()[Entry::SIZE - 4] ^= 0xFF;
                // Corrupt CRC of second slot (last 4 bytes of second slot)
                corrupted.as_mut()[Entry::FULL_SIZE - 4] ^= 0xFF;
                blob.write_at(0, corrupted, WriteOptions::SYNC)
                    .await
                    .unwrap();
            }

            // Reopen to trigger recovery. The bug would set both cleared entries to
            // Entry::new(0,0,0,0) which has is_empty()=false and is_valid()=true.
            // read_latest_entry would then see two "valid" entries with epoch=0 and
            // panic on unreachable!().
            let freezer = Freezer::<_, FixedBytes<64>, i32>::init(
                context.child("second"),
                cfg.clone(),
                Some(checkpoint),
            )
            .await
            .unwrap();
            drop(freezer);
        });
    }

    #[test_traced]
    fn no_checkpoint_deletes_partial_sync_resize() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = super::super::Config {
                key_partition: "test-key-index".into(),
                key_write_buffer: NZUsize!(1024),
                key_page_cache: CacheRef::from_pooler(&context, NZU16!(1024), NZUsize!(10)),
                value_partition: "test-value-journal".into(),
                value_compression: None,
                value_write_buffer: NZUsize!(1024),
                value_target_size: 10 * 1024 * 1024,
                table_partition: "test-table".into(),
                table_initial_size: 2,
                table_resize_frequency: 1,
                table_resize_chunk_size: 1,
                table_replay_buffer: NZUsize!(64 * 1024),
                codec_config: (),
            };
            let key = test_key_at_index(4, 3);

            {
                let freezer = Freezer::<_, FixedBytes<64>, i32>::init(
                    context.child("first"),
                    cfg.clone(),
                    None,
                )
                .await
                .unwrap();
                let (freezer, _) = freezer.put(key.clone(), 42).await.unwrap();
                let (freezer, _) = freezer.sync().await.unwrap();

                assert_eq!(freezer.resizing(), Some(1));
                assert_eq!(freezer.get(Identifier::Key(&key)).await.unwrap(), Some(42));
            }

            let freezer =
                Freezer::<_, FixedBytes<64>, i32>::init(context.child("second"), cfg.clone(), None)
                    .await
                    .unwrap();
            assert_eq!(freezer.table_size(), 2);
            assert_eq!(freezer.resizing(), None);
            assert_eq!(freezer.get(Identifier::Key(&key)).await.unwrap(), None);
        });
    }

    #[test_traced]
    fn empty_checkpoint_deletes_existing_data() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = super::super::Config {
                key_partition: "test-key-index".into(),
                key_write_buffer: NZUsize!(1024),
                key_page_cache: CacheRef::from_pooler(&context, NZU16!(1024), NZUsize!(10)),
                value_partition: "test-value-journal".into(),
                value_compression: None,
                value_write_buffer: NZUsize!(1024),
                value_target_size: 10 * 1024 * 1024,
                table_partition: "test-table".into(),
                table_initial_size: 2,
                table_resize_frequency: 1,
                table_resize_chunk_size: 1,
                table_replay_buffer: NZUsize!(64 * 1024),
                codec_config: (),
            };
            let key = test_key_at_index(4, 3);

            {
                let freezer = Freezer::<_, FixedBytes<64>, i32>::init(
                    context.child("first"),
                    cfg.clone(),
                    None,
                )
                .await
                .unwrap();
                let (freezer, _) = freezer.put(key.clone(), 42).await.unwrap();
                let (freezer, _) = freezer.sync().await.unwrap();
                assert_eq!(freezer.get(Identifier::Key(&key)).await.unwrap(), Some(42));
            }

            let checkpoint = Checkpoint {
                epoch: 0,
                section: 0,
                oversized_size: 0,
                table_size: 0,
            };
            let freezer = Freezer::<_, FixedBytes<64>, i32>::init(
                context.child("second"),
                cfg.clone(),
                Some(checkpoint),
            )
            .await
            .unwrap();
            assert_eq!(freezer.table_size(), 2);
            assert_eq!(freezer.get(Identifier::Key(&key)).await.unwrap(), None);
        });
    }

    #[test_traced]
    fn no_checkpoint_deletes_close_started_partial_resize() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = super::super::Config {
                key_partition: "test-key-index".into(),
                key_write_buffer: NZUsize!(1024),
                key_page_cache: CacheRef::from_pooler(&context, NZU16!(1024), NZUsize!(10)),
                value_partition: "test-value-journal".into(),
                value_compression: None,
                value_write_buffer: NZUsize!(1024),
                value_target_size: 10 * 1024 * 1024,
                table_partition: "test-table".into(),
                table_initial_size: 2,
                table_resize_frequency: 1,
                table_resize_chunk_size: 1,
                table_replay_buffer: NZUsize!(64 * 1024),
                codec_config: (),
            };
            let key = test_key_at_index(4, 3);

            {
                let freezer = Freezer::<_, FixedBytes<64>, i32>::init(
                    context.child("first"),
                    cfg.clone(),
                    None,
                )
                .await
                .unwrap();
                let (freezer, _) = freezer.put(key.clone(), 42).await.unwrap();
                let checkpoint = freezer.close().await.unwrap();
                assert_eq!(checkpoint.table_size, 2);
            }

            let freezer =
                Freezer::<_, FixedBytes<64>, i32>::init(context.child("second"), cfg.clone(), None)
                    .await
                    .unwrap();
            assert_eq!(freezer.table_size(), 2);
            assert_eq!(freezer.resizing(), None);
            assert_eq!(freezer.get(Identifier::Key(&key)).await.unwrap(), None);
        });
    }

    #[test_traced]
    fn no_checkpoint_deletes_completed_resize() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = super::super::Config {
                key_partition: "test-key-index".into(),
                key_write_buffer: NZUsize!(1024),
                key_page_cache: CacheRef::from_pooler(&context, NZU16!(1024), NZUsize!(10)),
                value_partition: "test-value-journal".into(),
                value_compression: None,
                value_write_buffer: NZUsize!(1024),
                value_target_size: 10 * 1024 * 1024,
                table_partition: "test-table".into(),
                table_initial_size: 2,
                table_resize_frequency: 1,
                table_resize_chunk_size: 2,
                table_replay_buffer: NZUsize!(64 * 1024),
                codec_config: (),
            };
            let key = test_key_at_index(4, 3);

            {
                let freezer = Freezer::<_, FixedBytes<64>, i32>::init(
                    context.child("first"),
                    cfg.clone(),
                    None,
                )
                .await
                .unwrap();
                let (freezer, _) = freezer.put(key.clone(), 42).await.unwrap();
                let (freezer, checkpoint) = freezer.sync().await.unwrap();

                assert_eq!(checkpoint.table_size, 4);
                assert_eq!(freezer.resizing(), None);
                assert_eq!(freezer.get(Identifier::Key(&key)).await.unwrap(), Some(42));
            }

            let freezer =
                Freezer::<_, FixedBytes<64>, i32>::init(context.child("second"), cfg.clone(), None)
                    .await
                    .unwrap();
            assert_eq!(freezer.table_size(), 2);
            assert_eq!(freezer.get(Identifier::Key(&key)).await.unwrap(), None);
        });
    }

    #[test_traced]
    fn checkpoint_rewinds_completed_resize() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = super::super::Config {
                key_partition: "test-key-index".into(),
                key_write_buffer: NZUsize!(1024),
                key_page_cache: CacheRef::from_pooler(&context, NZU16!(1024), NZUsize!(10)),
                value_partition: "test-value-journal".into(),
                value_compression: None,
                value_write_buffer: NZUsize!(1024),
                value_target_size: 10 * 1024 * 1024,
                table_partition: "test-table".into(),
                table_initial_size: 2,
                table_resize_frequency: 1,
                table_resize_chunk_size: 2,
                table_replay_buffer: NZUsize!(64 * 1024),
                codec_config: (),
            };
            let key = test_key_at_index(4, 3);

            let stale_checkpoint = {
                let freezer = Freezer::<_, FixedBytes<64>, i32>::init(
                    context.child("first"),
                    cfg.clone(),
                    None,
                )
                .await
                .unwrap();
                let (freezer, stale_checkpoint) = freezer.sync().await.unwrap();
                assert_eq!(stale_checkpoint.table_size, 2);

                let (freezer, _) = freezer.put(key.clone(), 42).await.unwrap();
                let (freezer, checkpoint) = freezer.sync().await.unwrap();
                assert_eq!(checkpoint.table_size, 4);
                assert_eq!(freezer.resizing(), None);
                assert_eq!(freezer.get(Identifier::Key(&key)).await.unwrap(), Some(42));

                stale_checkpoint
            };

            let freezer = Freezer::<_, FixedBytes<64>, i32>::init(
                context.child("second"),
                cfg.clone(),
                Some(stale_checkpoint),
            )
            .await
            .unwrap();
            assert_eq!(freezer.table_size(), 2);
            assert_eq!(freezer.get(Identifier::Key(&key)).await.unwrap(), None);
        });
    }

    #[test_traced]
    fn non_empty_checkpoint_against_empty_table_errors() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = super::super::Config {
                key_partition: "test-key-index".into(),
                key_write_buffer: NZUsize!(1024),
                key_page_cache: CacheRef::from_pooler(&context, NZU16!(1024), NZUsize!(10)),
                value_partition: "test-value-journal".into(),
                value_compression: None,
                value_write_buffer: NZUsize!(1024),
                value_target_size: 10 * 1024 * 1024,
                table_partition: "test-table".into(),
                table_initial_size: 2,
                table_resize_frequency: 1,
                table_resize_chunk_size: 1,
                table_replay_buffer: NZUsize!(64 * 1024),
                codec_config: (),
            };

            let checkpoint = Checkpoint {
                epoch: 1,
                section: 0,
                oversized_size: 0,
                table_size: 2,
            };
            let result = Freezer::<_, FixedBytes<64>, i32>::init(
                context.child("storage"),
                cfg.clone(),
                Some(checkpoint),
            )
            .await;
            assert!(matches!(result, Err(Error::CheckpointMismatch)));
        });
    }

    /// A durable checkpoint's table fsync completed at the checkpointed size, so a shorter
    /// (but non-empty) table is corruption. Initialization must reject it rather than grow the
    /// table with fabricated empty entries, and retries must fail identically.
    #[test_traced]
    fn non_empty_checkpoint_against_short_table_errors() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = super::super::Config {
                key_partition: "test-key-index".into(),
                key_write_buffer: NZUsize!(1024),
                key_page_cache: CacheRef::from_pooler(&context, NZU16!(1024), NZUsize!(10)),
                value_partition: "test-value-journal".into(),
                value_compression: None,
                value_write_buffer: NZUsize!(1024),
                value_target_size: 10 * 1024 * 1024,
                table_partition: "test-table".into(),
                table_initial_size: 2,
                table_resize_frequency: 1,
                table_resize_chunk_size: 1,
                table_replay_buffer: NZUsize!(64 * 1024),
                codec_config: (),
            };

            let short_len = Entry::FULL_SIZE as u64;
            let (table, _) = context
                .open(&cfg.table_partition, TABLE_BLOB_NAME)
                .await
                .unwrap();
            table.resize(short_len).await.unwrap();
            table.sync().await.unwrap();
            drop(table);

            let checkpoint = Checkpoint {
                epoch: 1,
                section: 0,
                oversized_size: 0,
                table_size: 2,
            };
            for child in ["first", "retry"] {
                let result = Freezer::<_, FixedBytes<64>, i32>::init(
                    context.child(child),
                    cfg.clone(),
                    Some(checkpoint),
                )
                .await;
                assert!(matches!(result, Err(Error::CheckpointMismatch)));
            }

            // The rejection must not resize the table.
            let (_, size) = context
                .open(&cfg.table_partition, TABLE_BLOB_NAME)
                .await
                .unwrap();
            assert_eq!(size, short_len);
        });
    }

    #[test_traced]
    fn corrupted_committed_value_surfaces_at_read() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = super::super::Config {
                key_partition: "test-key-index".into(),
                key_write_buffer: NZUsize!(1024),
                key_page_cache: CacheRef::from_pooler(&context, NZU16!(1024), NZUsize!(10)),
                value_partition: "test-value-journal".into(),
                value_compression: None,
                value_write_buffer: NZUsize!(1024),
                value_target_size: 10 * 1024 * 1024,
                table_partition: "test-table".into(),
                table_initial_size: 4,
                table_resize_frequency: 1,
                table_resize_chunk_size: 4,
                table_replay_buffer: NZUsize!(64 * 1024),
                codec_config: (),
            };

            // Create freezer with committed data
            let checkpoint = {
                let freezer = Freezer::<_, FixedBytes<64>, i32>::init(
                    context.child("first"),
                    cfg.clone(),
                    None,
                )
                .await
                .unwrap();
                let (freezer, _) = freezer.put(test_key("key0"), 42).await.unwrap();
                let (freezer, _) = freezer.put(test_key("key1"), 43).await.unwrap();
                let (freezer, _) = freezer.sync().await.unwrap();
                freezer.close().await.unwrap()
            };
            assert!(checkpoint.oversized_size > 0);

            // Corrupt the last committed value's checksum in the value journal
            {
                let (blob, len) = context
                    .open(&cfg.value_partition, &checkpoint.section.to_be_bytes())
                    .await
                    .unwrap();
                let byte = blob
                    .read_at(len - 1, 1, ReadOptions::default())
                    .await
                    .unwrap();
                let mut corrupted = byte.coalesce();
                corrupted.as_mut()[0] ^= 0xFF;
                blob.write_at(len - 1, corrupted, WriteOptions::SYNC)
                    .await
                    .unwrap();
            }

            // Recovery restores the checkpointed state without probing committed
            // values, so init succeeds and the corruption surfaces at read on
            // exactly the affected key.
            let freezer = Freezer::<_, FixedBytes<64>, i32>::init(
                context.child("second"),
                cfg.clone(),
                Some(checkpoint),
            )
            .await
            .unwrap();
            assert!(matches!(
                freezer.get(Identifier::Key(&test_key("key1"))).await,
                Err(Error::Journal(crate::journal::Error::ChecksumMismatch(
                    _,
                    _
                )))
            ));
            assert_eq!(
                freezer
                    .get(Identifier::Key(&test_key("key0")))
                    .await
                    .unwrap(),
                Some(42)
            );

            // The freezer remains usable
            let (freezer, _) = freezer.put(test_key("key2"), 44).await.unwrap();
            let (freezer, _) = freezer.sync().await.unwrap();
            assert_eq!(
                freezer
                    .get(Identifier::Key(&test_key("key2")))
                    .await
                    .unwrap(),
                Some(44)
            );
        });
    }

    #[test_traced]
    fn incomplete_committed_section_fails_init() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // A tiny value target so every put seals a section
            let cfg = super::super::Config {
                key_partition: "test-key-index".into(),
                key_write_buffer: NZUsize!(1024),
                key_page_cache: CacheRef::from_pooler(&context, NZU16!(1024), NZUsize!(10)),
                value_partition: "test-value-journal".into(),
                value_compression: None,
                value_write_buffer: NZUsize!(1024),
                value_target_size: 8,
                table_partition: "test-table".into(),
                table_initial_size: 4,
                table_resize_frequency: 1,
                table_resize_chunk_size: 4,
                table_replay_buffer: NZUsize!(64 * 1024),
                codec_config: (),
            };

            // Create freezer with committed data across multiple sections
            let checkpoint = {
                let freezer = Freezer::<_, FixedBytes<64>, i32>::init(
                    context.child("first"),
                    cfg.clone(),
                    None,
                )
                .await
                .unwrap();
                let (freezer, _) = freezer.put(test_key("key0"), 42).await.unwrap();
                let (freezer, _) = freezer.put(test_key("key1"), 43).await.unwrap();
                let (freezer, _) = freezer.sync().await.unwrap();
                freezer.close().await.unwrap()
            };
            assert!(checkpoint.section > 0);

            // Truncate the first committed section's values, simulating lost durable
            // state below the checkpoint
            {
                let (blob, len) = context
                    .open(&cfg.value_partition, &0u64.to_be_bytes())
                    .await
                    .unwrap();
                assert!(len > 0);
                blob.resize(len - 1).await.unwrap();
                blob.sync().await.unwrap();
            }

            // The checkpoint covers the damaged section, so init must fail rather than
            // silently absorb the loss. Nothing is repaired, so the failure persists
            // across restarts.
            for instance in ["second", "third"] {
                let result = Freezer::<_, FixedBytes<64>, i32>::init(
                    context.child(instance),
                    cfg.clone(),
                    Some(checkpoint),
                )
                .await;
                assert!(matches!(
                    result,
                    Err(Error::Journal(crate::journal::Error::Corruption(_)))
                ));
            }
        });
    }
}