commonware_storage/journal/segmented/

variable.rs

//! An append-only log for storing arbitrary variable length items.
//!
//! `segmented::Journal` is an append-only log for storing arbitrary variable length data on disk. In
//! addition to replay, stored items can be directly retrieved given their section number and offset
//! within the section.
//!
//! # Format
//!
//! Data stored in `Journal` is persisted in one of many Blobs within a caller-provided `partition`.
//! The particular [Blob] in which data is stored is identified by a `section` number (`u64`).
//! Within a `section`, data is appended as an `item` with the following format:
//!
//! ```text
//! +---+---+---+---+---+---+---+---+
//! |       0 ~ 4       |    ...    |
//! +---+---+---+---+---+---+---+---+
//! | Size (varint u32) |   Data    |
//! +---+---+---+---+---+---+---+---+
//! ```
//!
//! # Open Blobs
//!
//! `Journal` uses 1 `commonware-storage::Blob` per `section` to store data. All `Blobs` in a given
//! `partition` are kept open during the lifetime of `Journal`. If the caller wishes to bound the
//! number of open `Blobs`, they can group data into fewer `sections` and/or prune unused
//! `sections`.
//!
//! # Sync
//!
//! Data written to `Journal` may not be immediately persisted to `Storage`. It is up to the caller
//! to determine when to force pending data to be written to `Storage` using the `sync` (or
//! `sync_all`) method.
//!
//! # Pruning
//!
//! All data appended to `Journal` must be assigned to some `section` (`u64`). This assignment
//! allows the caller to prune data from `Journal` by specifying a minimum `section` number. This
//! could be used, for example, by some blockchain application to prune old blocks.
//!
//! # Replay
//!
//! During application initialization, it is very common to replay data from `Journal` to recover
//! some in-memory state. `Journal` is heavily optimized for this pattern and provides a `replay`
//! method that consumes the journal into an owned [Replay] reader yielding all items in order
//! of their `section` and `offset`. [Replay::finish] returns the journal once exhausted.
//!
//! # Compression
//!
//! [Journal] supports optional zstd compression through [Config::compression]. Keep the choice
//! between `None` and `Some(_)` fixed while stored items are retained. Only the compression level
//! may change between initializations when compression is enabled.
//!
//! # Example
//!
//! ```rust
//! use commonware_runtime::{Spawner, Runner, deterministic, buffer::paged::CacheRef};
//! use commonware_storage::journal::segmented::variable::{Journal, Config};
//! use commonware_utils::{NZUsize, NZU16};
//!
//! let executor = deterministic::Runner::default();
//! executor.start(|context| async move {
//!     // Create a page cache
//!     let page_cache = CacheRef::new(NZU16!(1024), NZUsize!(10));
//!
//!     // Create a journal
//!     let journal = Journal::init(context, Config {
//!         partition: "partition".into(),
//!         compression: None,
//!         codec_config: (),
//!         page_cache,
//!         write_buffer: NZUsize!(1024 * 1024),
//!     }).await.unwrap();
//!
//!     // Append data to the journal
//!     let (journal, _, _) = journal.append(1, &128).await.unwrap();
//!
//!     // Sync the journal
//!     journal.sync_all().await.unwrap();
//! });
//! ```

use super::manager::{AppendFactory, Config as ManagerConfig, Manager};
use crate::journal::{
    Error,
    frame::{
        FrameInfo, Limited, UncompressedFrame, decode_item, decode_length_prefix,
        encode_compressed_frame_into, find_frame, read_frame_at,
    },
};
use bytes::Bytes;
use commonware_codec::{Codec, CodecShared, Copying, Encode as _, varint::MAX_U32_VARINT_SIZE};
use commonware_runtime::{
    Blob, Buf, Error as RError, Handle, IoBuf, Metrics, ReadOptions, Storage,
    buffer::paged::{CacheRef, Recovery as PagedRecovery, Replay as BlobReplay},
};
use std::{
    collections::{BTreeSet, VecDeque},
    num::NonZeroUsize,
};
use tracing::{trace, warn};

/// Configuration for `Journal` storage.
#[derive(Clone)]
pub struct Config<C> {
    /// The `commonware-runtime::Storage` partition to use
    /// for storing journal blobs.
    pub partition: String,

    /// Optional compression level (using `zstd`) to apply to data before storing.
    pub compression: Option<u8>,

    /// The codec configuration to use for encoding and decoding items.
    pub codec_config: C,

    /// The page cache to use for caching data.
    pub page_cache: CacheRef,

    /// The size of the write buffer to use for each blob.
    pub write_buffer: NonZeroUsize,
}

/// State for replaying a single section's blob.
struct SectionReplay<B: Blob> {
    section: u64,
    reader: BlobReplay<B>,
    skip_bytes: u64,
    offset: u64,
    valid_offset: u64,
    pending: Option<(usize, usize)>,
}

/// The journal's state, boxed so the public [Journal] handle stays pointer-sized.
struct Inner<E: Storage + Metrics, V: Codec> {
    manager: Manager<E, AppendFactory>,

    /// Nonempty sections opened at initialization that have not been replayed from offset zero.
    /// Public replay may repair them, and appends are blocked until their full replay succeeds.
    unrecovered: BTreeSet<u64>,

    /// Compression level (if enabled).
    compression: Option<u8>,

    /// Codec configuration.
    codec_config: V::Cfg,
}

impl<E: Storage + Metrics, V: Codec> Inner<E, V> {
    /// The section's writer. Only a failed or cancelled repair removes a replayed section, and
    /// either ends the replay.
    fn writer(&self, section: u64) -> Result<&PagedRecovery<E::Blob>, Error> {
        Ok(self
            .manager
            .get(section)?
            .expect("replayed section is present"))
    }
}

impl<E: Storage + Metrics, V: CodecShared> Inner<E, V> {
    /// See [Journal::init].
    async fn init(context: E, cfg: Config<V::Cfg>) -> Result<Self, Error> {
        let manager_cfg = ManagerConfig {
            partition: cfg.partition,
            factory: AppendFactory {
                write_buffer: cfg.write_buffer,
                page_cache_ref: cfg.page_cache,
            },
        };
        let manager = Manager::init(context, manager_cfg).await?;
        let mut unrecovered = BTreeSet::new();
        for section in manager.sections() {
            if manager.size(section)? != 0 {
                unrecovered.insert(section);
            }
        }

        Ok(Self {
            manager,
            unrecovered,
            compression: cfg.compression,
            codec_config: cfg.codec_config,
        })
    }

    /// Reads an item from the blob at the given offset.
    async fn read(
        compressed: bool,
        cfg: &V::Cfg,
        blob: &PagedRecovery<E::Blob>,
        offset: u64,
    ) -> Result<(u64, u32, V), Error> {
        read_frame_at(blob, offset, cfg, compressed).await
    }

    /// See [Journal::append].
    async fn append(&mut self, section: u64, item: &V) -> Result<(u64, u32), Error> {
        assert!(
            !self.unrecovered.contains(&section),
            "section {section} must be replayed before append"
        );

        // Size and validate the frame before creating the section so a rejected item leaves no
        // empty section.
        let (offset, item_len) = if let Some(level) = self.compression {
            // Buffer compressed output to determine its length before appending the frame.
            let mut buf = Vec::new();
            let item_len = encode_compressed_frame_into(level, item, &mut buf)?;
            let blob = self.manager.get_or_create(section).await?;
            let offset = match blob.try_append(&buf) {
                Some(offset) => offset,
                None => {
                    // Return the writer to the manager only after the owned append succeeds.
                    let blob = self.manager.take(section).await?;
                    let (blob, offset) = blob.append_owned(IoBuf::from(buf)).await?;
                    self.manager.put(section, blob);
                    offset
                }
            };
            (offset, item_len)
        } else {
            // Encode directly into the write buffer when the frame fits.
            let frame = UncompressedFrame::new(item)?;
            let blob = self.manager.get_or_create(section).await?;
            let offset = match blob.try_append_value(&frame) {
                Some(offset) => offset,
                None => {
                    // Return the writer to the manager only after the owned append succeeds.
                    let blob = self.manager.take(section).await?;
                    let (blob, offset) = blob.append_owned(frame.encode_mut().into()).await?;
                    self.manager.put(section, blob);
                    offset
                }
            };
            (offset, frame.item_len)
        };
        trace!(blob = section, offset, "appended item");
        Ok((offset, item_len))
    }

    /// See [Journal::get].
    async fn get(&self, section: u64, offset: u64) -> Result<V, Error> {
        let blob = self
            .manager
            .get(section)?
            .ok_or(Error::SectionOutOfRange(section))?;

        // Perform a multi-op read.
        let (_, _, item) =
            Self::read(self.compression.is_some(), &self.codec_config, blob, offset).await?;
        Ok(item)
    }

    /// See [Journal::get_many].
    async fn get_many(&self, section: u64, offsets: &[u64]) -> Result<Vec<V>, Error> {
        if offsets.is_empty() {
            return Ok(Vec::new());
        }
        let blob = self
            .manager
            .get(section)?
            .ok_or(Error::SectionOutOfRange(section))?;

        let compressed = self.compression.is_some();
        let cfg = &self.codec_config;
        let mut items = Vec::with_capacity(offsets.len());
        for &offset in offsets {
            let (_, _, item) = Self::read(compressed, cfg, blob, offset).await?;
            items.push(item);
        }
        Ok(items)
    }

    /// See [Journal::try_get_sync].
    fn try_get_sync(&self, section: u64, offset: u64) -> Option<V> {
        let blob = self.manager.get(section).ok()??;
        let remaining = blob.size().checked_sub(offset)?;
        let header_len = usize::try_from(remaining.min(MAX_U32_VARINT_SIZE as u64)).ok()?;
        if header_len == 0 {
            return None;
        }

        // Read the varint header to determine item size.
        let mut header = [0u8; MAX_U32_VARINT_SIZE];
        if !blob.try_read_sync_into(&mut header[..header_len], offset) {
            return None;
        }
        let mut cursor = Copying(&header[..header_len]);
        let (_, frame_info) = find_frame(&mut cursor, offset).ok()?;
        let (varint_len, data_len) = match frame_info {
            FrameInfo::Complete {
                varint_len,
                data_len,
            } => (varint_len, data_len),
            FrameInfo::Incomplete {
                varint_len,
                total_len,
                ..
            } => (varint_len, total_len),
        };
        let item_len = varint_len.checked_add(data_len)?;
        if item_len > usize::try_from(remaining).ok()? {
            return None;
        }

        // If the full item fits in the header read, decode directly.
        let compressed = self.compression.is_some();
        if item_len <= header_len {
            return decode_item::<V>(
                Copying(&header[varint_len..varint_len + data_len]),
                &self.codec_config,
                compressed,
            )
            .ok();
        }

        // Otherwise try reading the full item from cache. The buffer holds exactly the frame, so
        // skipping the varint leaves the item.
        let mut buf = vec![0u8; item_len];
        if !blob.try_read_sync_into(&mut buf, offset) {
            return None;
        }
        let mut buf = Bytes::from(buf);
        buf.advance(varint_len);
        decode_item::<V>(buf, &self.codec_config, compressed).ok()
    }

    /// See [Journal::size].
    fn size(&self, section: u64) -> Result<u64, Error> {
        self.manager.size(section)
    }

    /// See [Journal::sync].
    async fn sync(mut self: Box<Self>, sections: impl crate::Sections) -> Result<Box<Self>, Error> {
        self.manager = self.manager.sync(sections).await?;
        Ok(self)
    }

    /// See [Journal::start_sync].
    async fn start_sync(
        mut self: Box<Self>,
        sections: impl crate::Sections,
    ) -> Result<(Box<Self>, Handle<()>), Error> {
        let (manager, handle) = self.manager.start_sync(sections).await?;
        self.manager = manager;
        Ok((self, handle))
    }

    /// See [Journal::sync_all].
    async fn sync_all(mut self: Box<Self>) -> Result<Box<Self>, Error> {
        self.manager = self.manager.sync_all().await?;
        Ok(self)
    }

    /// See [Journal::prune].
    async fn prune(mut self: Box<Self>, min: u64) -> Result<(Box<Self>, bool), Error> {
        let (manager, pruned) = self.manager.prune(min).await?;
        self.manager = manager;
        if pruned {
            self.unrecovered.retain(|section| *section >= min);
        }
        Ok((self, pruned))
    }

    /// See [Journal::pruned].
    const fn pruned(&self, section: u64) -> bool {
        self.manager.pruned(section)
    }

    /// See [Journal::oldest_section].
    fn oldest_section(&self) -> Option<u64> {
        self.manager.oldest_section()
    }

    /// See [Journal::newest_section].
    fn newest_section(&self) -> Option<u64> {
        self.manager.newest_section()
    }

    /// See [Journal::is_empty].
    fn is_empty(&self) -> bool {
        self.manager.is_empty()
    }

    /// See [Journal::num_sections].
    fn num_sections(&self) -> usize {
        self.manager.num_sections()
    }

    /// See [Journal::destroy].
    async fn destroy(self) -> Result<(), Error> {
        self.manager.destroy().await
    }

    /// See [Journal::clear].
    async fn clear(mut self: Box<Self>) -> Result<Box<Self>, Error> {
        self.manager = self.manager.clear().await?;
        self.unrecovered.clear();
        Ok(self)
    }
}

/// A segmented journal with variable-size entries.
///
/// Each section is stored in a separate blob. Items are length-prefixed with a varint.
///
/// # Repair
///
/// Like
/// [sqlite](https://github.com/sqlite/sqlite/blob/8658a8df59f00ec8fcfea336a2a6a4b5ef79d2ee/src/wal.c#L1504-L1505)
/// and
/// [rocksdb](https://github.com/facebook/rocksdb/blob/0c533e61bc6d89fdf1295e8e0bcee4edb3aef401/include/rocksdb/options.h#L441-L445),
/// the first invalid data read will be considered the new end of the journal (and the
/// underlying [Blob] will be truncated to the last valid item). Repair occurs during
/// replay (not init) because any blob could have trailing bytes.
/// A nonempty section opened during initialization must be replayed from offset zero before it
/// accepts new appends. Sections created during the current execution can be appended immediately.
///
/// Mutating functions consume the journal and return it only on success: an error (or a dropped
/// future) destroys the handle. [Journal::replay] consumes the journal into an owned [Replay]
/// reader, which returns it via [Replay::finish] once exhausted. Mutations on pruned sections
/// fail with [Error::AlreadyPrunedToSection] without mutating. Check [Journal::pruned] first to
/// keep the handle.
pub struct Journal<E: Storage + Metrics, V: Codec>(Box<Inner<E, V>>);

impl<E: Storage + Metrics, V: CodecShared> std::fmt::Debug for Journal<E, V> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Journal")
            .field("oldest_section", &self.oldest_section())
            .field("newest_section", &self.newest_section())
            .finish_non_exhaustive()
    }
}

impl<E: Storage + Metrics, V: CodecShared> Journal<E, V> {
    /// Open through `(section, logical byte end)` and remove later sections.
    ///
    /// An end inside an item rounds down. All retained data is validated before returning. Payload
    /// reads stop at the bound, but every section name is validated, including names above it.
    pub async fn init_at_most(
        context: E,
        cfg: Config<V::Cfg>,
        section: u64,
        end: u64,
    ) -> Result<Self, Error> {
        let replay_buffer = cfg.write_buffer;
        let mut end = end;
        super::manager::truncate_paged_tail(
            &context,
            &cfg.partition,
            cfg.page_cache.page_size(),
            section,
            end,
        )
        .await?;
        let mut journal = Self::init(context, cfg).await?;

        // Resolve a torn page before truncation tries to rewrite its partial checksum.
        if let Some(blob) = journal.0.manager.get(section)?
            && end > 0
            && end < blob.size()
        {
            end = end.min(
                blob.recoverable_prefix_len_at_most(0, end, replay_buffer, ReadOptions::default())
                    .await?,
            );
        }
        journal.0.manager = journal.0.manager.truncate_pending(section, end).await?;
        let mut replay = journal
            .replay(0, 0, replay_buffer, ReadOptions::default())
            .await?;
        while let Some(item) = replay.next().await {
            item?;
        }
        replay.finish()
    }

    /// Initialize a new `Journal` instance.
    ///
    /// All backing blobs are opened but not read during
    /// initialization. The `replay` method can be used
    /// to iterate over all items in the `Journal`.
    pub async fn init(context: E, cfg: Config<V::Cfg>) -> Result<Self, Error> {
        Ok(Self(Box::new(Inner::init(context, cfg).await?)))
    }

    /// Consumes the journal and returns an owned [Replay] reader over all items starting
    /// with the item at the given `start_section` and `start_offset` into that section.
    ///
    /// Setup flushes buffered pages so the reader observes every accepted write. It
    /// validates the requested start bound and copies each replayed section's partial tail
    /// page, but does not allocate `buffer` bytes per blob. Read buffers are allocated
    /// lazily as the reader advances. Every backing blob read performed by the returned
    /// replay uses `read_options`, including reads after advancing to another section.
    ///
    /// A nonzero start must be a boundary already validated by a prior replay or a durable
    /// marker: torn-page repair treats everything below it as proven.
    pub async fn replay(
        mut self,
        start_section: u64,
        start_offset: u64,
        buffer: NonZeroUsize,
        read_options: ReadOptions,
    ) -> Result<Replay<E, V>, Error> {
        // Capture independent replay views while returning each writer to the manager.
        let mut sections = VecDeque::new();
        let replayed: Vec<_> = self.0.manager.sections_from(start_section).collect();
        for section in replayed {
            let blob = self.0.manager.take(section).await?;
            let (blob, reader) = blob.replay(buffer, read_options).await?;
            self.0.manager.put(section, blob);

            // Only the first section skips bytes preceding the validated starting boundary.
            let skip_bytes = if section == start_section {
                start_offset
            } else {
                0
            };
            sections.push_back(SectionReplay {
                section,
                reader,
                skip_bytes,
                offset: 0,
                valid_offset: skip_bytes,
                pending: None,
            });
        }
        let finished = sections.is_empty();
        let replay = Replay {
            journal: self,
            sections,
            recovered_from: if start_offset == 0 {
                Some(start_section)
            } else {
                start_section.checked_add(1)
            },
            buffer,
            read_options,
            finished,
            errored: false,
            repairing: false,
        };

        // A start offset beyond the front section's apparent tail can never resolve to an
        // item boundary. Reject it up front rather than yielding a silently empty replay:
        // the offset is caller-supplied and unvalidated, so it must never be adopted.
        if let Some(current) = replay.sections.front()
            && current.section == start_section
            && start_offset > current.reader.blob_size()
        {
            return Err(Error::ItemOutOfRange(start_offset));
        }
        Ok(replay)
    }

    /// Appends an item to `Journal` in a given `section`, returning the offset
    /// where the item was written and the size of the item (which may differ
    /// from the raw encoded size if compression is enabled).
    ///
    /// # Panics
    ///
    /// Panics when `section` contained data at initialization and has not completed a replay
    /// from offset zero.
    pub async fn append(mut self, section: u64, item: &V) -> Result<(Self, u64, u32), Error> {
        let (offset, item_len) = self.0.append(section, item).await?;
        Ok((self, offset, item_len))
    }

    /// Retrieves an item from `Journal` at a given `section` and `offset`.
    ///
    /// # Errors
    ///  - [Error::AlreadyPrunedToSection] if the requested `section` has been pruned during the
    ///    current execution.
    ///  - [Error::SectionOutOfRange] if the requested `section` is empty (i.e. has never had any
    ///    data appended to it, or has been pruned in a previous execution).
    ///  - An invalid `offset` for a given section (that is, an offset that doesn't correspond to a
    ///    previously appended item) will result in an error, with the specific type being
    ///    undefined.
    pub async fn get(&self, section: u64, offset: u64) -> Result<V, Error> {
        self.0.get(section, offset).await
    }

    /// Read multiple items from the same section.
    ///
    /// Offsets should be sorted in ascending order.
    pub async fn get_many(&self, section: u64, offsets: &[u64]) -> Result<Vec<V>, Error> {
        self.0.get_many(section, offsets).await
    }

    /// Get an item if it can be done synchronously (e.g. without I/O), returning `None` otherwise.
    pub fn try_get_sync(&self, section: u64, offset: u64) -> Option<V> {
        self.0.try_get_sync(section, offset)
    }

    /// Gets the size of the journal for a specific section.
    ///
    /// Returns 0 if the section does not exist.
    pub fn size(&self, section: u64) -> Result<u64, Error> {
        self.0.size(section)
    }

    /// Ensures the given `sections` are synced to the underlying store.
    ///
    /// If a selected section does not exist (and has not been pruned), no error will be
    /// returned.
    pub async fn sync(mut self, sections: impl crate::Sections) -> Result<Self, Error> {
        self.0 = self.0.sync(sections).await?;
        Ok(self)
    }

    /// Start syncing the given `sections` to storage.
    ///
    /// An error reported by the returned [Handle] is fatal to the journal: the caller
    /// must stop using the returned journal.
    pub async fn start_sync(
        mut self,
        sections: impl crate::Sections,
    ) -> Result<(Self, Handle<()>), Error> {
        let (inner, handle) = self.0.start_sync(sections).await?;
        self.0 = inner;
        Ok((self, handle))
    }

    /// Syncs all open sections.
    pub async fn sync_all(mut self) -> Result<Self, Error> {
        self.0 = self.0.sync_all().await?;
        Ok(self)
    }

    /// Prunes all `sections` less than `min`. Returns true if any sections were pruned.
    pub async fn prune(mut self, min: u64) -> Result<(Self, bool), Error> {
        let (inner, pruned) = self.0.prune(min).await?;
        self.0 = inner;
        Ok((self, pruned))
    }

    /// Returns true when `section` is below the prune floor.
    ///
    /// The floor only tracks prunes from the current execution and resets at init, so a
    /// section pruned in a previous execution reports false.
    pub fn pruned(&self, section: u64) -> bool {
        self.0.pruned(section)
    }

    /// Returns the number of the oldest section in the journal.
    pub fn oldest_section(&self) -> Option<u64> {
        self.0.oldest_section()
    }

    /// Returns the number of the newest section in the journal.
    pub fn newest_section(&self) -> Option<u64> {
        self.0.newest_section()
    }

    /// Returns true if no sections exist.
    pub fn is_empty(&self) -> bool {
        self.0.is_empty()
    }

    /// Returns the number of sections.
    pub fn num_sections(&self) -> usize {
        self.0.num_sections()
    }

    /// Removes any underlying blobs created by the journal.
    pub async fn destroy(self) -> Result<(), Error> {
        self.0.destroy().await
    }

    /// Clear all data, resetting the journal to an empty state.
    ///
    /// Unlike `destroy`, this keeps the journal alive so it can be reused.
    pub async fn clear(mut self) -> Result<Self, Error> {
        self.0 = self.0.clear().await?;
        Ok(self)
    }
}

/// Owned replay reader over a [Journal]'s items.
///
/// Yields `(section, offset, size, item)` in order and repairs invalid trailing data as it is
/// encountered. Dropping the reader before it is exhausted destroys the journal (leaving later
/// sections unrepaired): recovery is re-initialization. Call [Replay::finish] on an exhausted
/// reader to get the journal back.
pub struct Replay<E: Storage + Metrics, V: Codec> {
    journal: Journal<E, V>,
    sections: VecDeque<SectionReplay<E::Blob>>,
    /// The first section this replay fully covers: [Replay::finish] marks it and every
    /// later section recovered.
    recovered_from: Option<u64>,
    buffer: NonZeroUsize,
    read_options: ReadOptions,
    finished: bool,
    errored: bool,
    repairing: bool,
}

impl<E: Storage + Metrics, V: CodecShared> Replay<E, V> {
    /// Validate that the front section's checksum failure is a repairable torn page, returning
    /// the truncation target.
    async fn plan_repair(&mut self, source: RError) -> Result<u64, Error> {
        // Only a checksum failure is repairable: it marks a torn write, while any other error
        // is an I/O failure this repair must not mask.
        if !matches!(source, RError::InvalidChecksum) {
            return Err(source.into());
        }

        // The bytes already replayed are validated: they bound the truncation from below.
        let current = self.sections.front().expect("replayed section is present");
        let section = current.section;
        if !self.journal.0.unrecovered.contains(&section) {
            return Err(source.into());
        }
        let size = current.reader.blob_size();
        let valid_offset = current.valid_offset;

        // Forward-validate from the replayed prefix to find where well-formed pages end.
        let recoverable = self
            .journal
            .0
            .writer(section)?
            .recoverable_prefix_len(valid_offset, self.buffer, self.read_options)
            .await?;

        // A whole-blob recoverable prefix means the checksum failure did not come from a torn
        // page: surface the original error instead of truncating valid data.
        if recoverable >= size {
            return Err(source.into());
        }

        // The cut must not drop below the validated replay prefix: that would lose data the
        // replay already handed out.
        if recoverable < valid_offset {
            return Err(Error::ItemOutOfRange(valid_offset));
        }

        Ok(recoverable)
    }

    /// Repair a torn page discovered by ordered replay and resume at the last complete item.
    async fn repair(&mut self, source: RError) -> Result<(), Error> {
        // A rejected plan mutates nothing: drop the damaged section and surface its error.
        let recoverable = match self.plan_repair(source).await {
            Ok(target) => target,
            Err(err) => {
                self.sections.pop_front();
                return Err(err);
            }
        };

        let current = self.sections.front().expect("replayed section is present");
        let (section, valid_offset) = (current.section, current.valid_offset);
        warn!(
            section,
            invalid_size = current.reader.blob_size(),
            new_size = recoverable,
            "torn page detected: truncating"
        );

        // Once mutation begins, a dropped future drops the section's writer. Keep the interruption
        // guard set until the repaired reader has replaced the stale one.
        self.repairing = true;
        let current = self
            .sections
            .pop_front()
            .expect("repaired section is present");
        drop(current.reader);
        let writer = self.journal.0.manager.take(section).await?;
        let writer = writer.truncate(recoverable).await?;
        let (writer, mut reader) = writer.replay(self.buffer, self.read_options).await?;
        self.journal.0.manager.put(section, writer);
        reader.seek_to(valid_offset)?;
        self.sections.push_front(SectionReplay {
            section,
            reader,
            skip_bytes: 0,
            offset: valid_offset,
            valid_offset,
            pending: None,
        });
        self.repairing = false;
        Ok(())
    }

    /// Truncate the front section to its validated prefix and make the repair durable.
    async fn repair_tail(&mut self, message: &'static str) -> Result<(), Error> {
        let current = self.sections.front().expect("replayed section is present");
        let (section, offset, valid_offset) =
            (current.section, current.offset, current.valid_offset);
        if !self.journal.0.unrecovered.contains(&section) {
            return Err(Error::ItemOutOfRange(offset));
        }
        warn!(
            blob = section,
            bad_offset = offset,
            new_size = valid_offset,
            "{message}"
        );

        // Tail repair is exceptional. Make it durable immediately so callers do not need to
        // track replay-time repaired sections separately. Keep the interruption guard set
        // until the repair is durable.
        self.repairing = true;
        let writer = self.journal.0.manager.take(section).await?;
        let writer = writer.truncate(valid_offset).await?;
        self.journal.0.manager.put(section, writer);
        self.repairing = false;
        Ok(())
    }

    /// Returns the next `(section, offset, size, item)`, or `None` once every section is
    /// exhausted.
    ///
    /// An error ends the section that produced it, and iteration continues with the next section.
    /// Errors while mutating storage to repair a section, and [Error::ReplayInterrupted], end the
    /// replay.
    pub async fn next(&mut self) -> Option<Result<(u64, u64, u32, V), Error>> {
        // An unfinished repair may have dropped the section's writer. A cancelled repair still
        // needs an error. A completed failure already yielded one.
        if self.repairing {
            self.repairing = false;
            self.sections.clear();
            if !self.errored {
                return self.fail(Error::ReplayInterrupted);
            }
        }
        while let Some(current) = self.sections.front_mut() {
            let blob_size = current.reader.blob_size();

            // Resume a recorded frame header or decode the next one
            let (item_size, varint_len) = match current.pending {
                Some(header) => header,
                None => {
                    // Ensure we have enough data for varint header.
                    // ensure() returns Ok(false) if exhausted with fewer bytes,
                    // but we still try to decode from remaining bytes.
                    match current.reader.ensure(MAX_U32_VARINT_SIZE).await {
                        Ok(true) => {}
                        Ok(false) => {
                            // Reader exhausted - check if buffer is empty
                            if current.reader.remaining() == 0 {
                                self.sections.pop_front();
                                continue;
                            }
                            // Buffer still has data - continue to try decoding
                        }
                        Err(err) => {
                            if let Err(err) = self.repair(err).await {
                                return self.fail(err);
                            }
                            continue;
                        }
                    }

                    // Skip bytes if needed (for start_offset)
                    if current.skip_bytes > 0 {
                        let to_skip =
                            current.skip_bytes.min(current.reader.remaining() as u64) as usize;
                        current.reader.advance(to_skip);
                        current.skip_bytes -= to_skip as u64;
                        current.offset += to_skip as u64;
                        continue;
                    }

                    // Try to decode length prefix
                    let before_remaining = current.reader.remaining();
                    match decode_length_prefix(&mut current.reader) {
                        Ok(header) => {
                            // Record the header before awaiting the body so a dropped
                            // next future resumes losslessly.
                            current.pending = Some(header);
                            header
                        }
                        Err(err) => {
                            // Could be incomplete varint - check if reader exhausted
                            if current.reader.is_exhausted()
                                || before_remaining < MAX_U32_VARINT_SIZE
                            {
                                // Treat as trailing bytes
                                if current.valid_offset < blob_size
                                    && current.offset < blob_size
                                    && let Err(err) = self
                                        .repair_tail("trailing bytes detected: truncating")
                                        .await
                                {
                                    self.sections.pop_front();
                                    return self.fail(err);
                                }
                                self.sections.pop_front();
                                continue;
                            }
                            self.sections.pop_front();
                            return self.fail(err);
                        }
                    }
                }
            };

            // Ensure we have enough data for item body
            match current.reader.ensure(item_size).await {
                Ok(true) => {}
                Ok(false) => {
                    // Incomplete item at end - truncate
                    if let Err(err) = self.repair_tail("incomplete item at end: truncating").await {
                        self.sections.pop_front();
                        return self.fail(err);
                    }
                    self.sections.pop_front();
                    continue;
                }
                Err(err) => {
                    if let Err(err) = self.repair(err).await {
                        return self.fail(err);
                    }
                    continue;
                }
            }

            // Decode the item without reading past its frame
            let item_offset = current.offset;
            let next_offset = match current
                .offset
                .checked_add(varint_len as u64)
                .and_then(|o| o.checked_add(item_size as u64))
            {
                Some(o) => o,
                None => {
                    self.sections.pop_front();
                    return self.fail(Error::OffsetOverflow);
                }
            };
            match decode_item::<V>(
                Limited::new(&mut current.reader, item_size),
                &self.journal.0.codec_config,
                self.journal.0.compression.is_some(),
            ) {
                Ok(decoded) => {
                    current.pending = None;
                    current.valid_offset = next_offset;
                    current.offset = next_offset;
                    return Some(Ok((
                        current.section,
                        item_offset,
                        item_size as u32,
                        decoded,
                    )));
                }
                Err(err) => {
                    self.sections.pop_front();
                    return self.fail(err);
                }
            }
        }
        self.finished = true;
        None
    }

    /// Records a yielded error, which is fatal to the journal.
    const fn fail(&mut self, err: Error) -> Option<Result<(u64, u64, u32, V), Error>> {
        self.errored = true;
        Some(Err(err))
    }

    /// Returns the journal.
    ///
    /// Fails when the reader was not fully drained or yielded an error: the journal is
    /// destroyed and recovery is re-initialization.
    pub fn finish(mut self) -> Result<Journal<E, V>, Error> {
        if self.errored || !self.finished {
            return Err(Error::ReplayFailed);
        }
        if let Some(start) = self.recovered_from {
            self.journal
                .0
                .unrecovered
                .retain(|section| *section < start);
        }
        Ok(self.journal)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::journal::frame::encode_frame_into;
    use commonware_codec::{EncodeSize, Write as _, varint::UInt};
    use commonware_macros::test_traced;
    use commonware_runtime::{
        Blob, BufMut, Runner, Storage, Supervisor as _, WriteOptions,
        buffer::paged::corrupt_page,
        deterministic,
        mocks::{DelayedSyncContext, PendingSyncs, RecordingContext, release_pending_syncs},
    };
    use commonware_utils::{NZU16, NZUsize, probability};
    use std::num::NonZeroU16;

    impl<E: Storage + Metrics, V: CodecShared> Inner<E, V> {
        /// Append raw bytes, which need not form a valid frame, to `section`.
        async fn append_raw(&mut self, section: u64, buf: IoBuf) -> Result<u64, Error> {
            let blob = self.manager.take(section).await?;
            let (blob, offset) = blob.append_owned(buf).await?;
            self.manager.put(section, blob);
            Ok(offset)
        }
    }

    impl<E: Storage + Metrics, V: CodecShared> Journal<E, V> {
        async fn test_reopen_at_most(self, section: u64, end: u64) -> Result<Self, Error> {
            let (context, partition, factory) = self.0.manager.test_configuration();
            let cfg = Config {
                partition,
                write_buffer: factory.write_buffer,
                page_cache: factory.page_cache_ref,
                compression: self.0.compression,
                codec_config: self.0.codec_config.clone(),
            };
            _ = self.sync_all().await?;
            Self::init_at_most(context, cfg, section, end).await
        }

        async fn test_reopen_section(self, section: u64, end: u64) -> Result<Self, Error> {
            let (context, partition, factory) = self.0.manager.test_configuration();
            let replay_buffer = factory.write_buffer;
            let cfg = Config {
                partition,
                write_buffer: replay_buffer,
                page_cache: factory.page_cache_ref,
                compression: self.0.compression,
                codec_config: self.0.codec_config.clone(),
            };
            _ = self.sync_all().await?;
            let mut journal = Self::init(context, cfg).await?;
            journal.0.manager = journal
                .0
                .manager
                .truncate_pending_section(section, end)
                .await?;
            let mut replay = journal
                .replay(0, 0, replay_buffer, ReadOptions::default())
                .await?;
            while let Some(item) = replay.next().await {
                item?;
            }
            replay.finish()
        }
    }

    const PAGE_SIZE: NonZeroU16 = NZU16!(1024);
    const PAGE_CACHE_SIZE: NonZeroUsize = NZUsize!(10);

    async fn journal_with_torn_interior_page(
        context: &deterministic::Context,
        partition: &str,
        later_section: bool,
    ) -> Journal<deterministic::Context, u64> {
        const LOGICAL_PAGE_SIZE: u64 = 64;
        const FIRST_SECTION: u64 = 0;
        const TORN_SECTION: u64 = 1;

        let cfg = Config {
            partition: partition.into(),
            compression: None,
            codec_config: (),
            page_cache: CacheRef::new(NZU16!(LOGICAL_PAGE_SIZE as u16), NZUsize!(4)),
            write_buffer: NZUsize!(256),
        };
        let mut journal = Journal::<_, u64>::init(context.child("first"), cfg.clone())
            .await
            .unwrap();
        (journal, _, _) = journal.append(FIRST_SECTION, &u64::MAX).await.unwrap();
        for value in 0..15u64 {
            let offset;
            (journal, offset, _) = journal.append(TORN_SECTION, &value).await.unwrap();
            assert_eq!(offset, value * 9);
        }
        if later_section {
            (journal, _, _) = journal.append(2, &u64::MIN).await.unwrap();
        }
        journal = journal.sync_all().await.unwrap();
        drop(journal);

        corrupt_page(
            context,
            &cfg.partition,
            &TORN_SECTION.to_be_bytes(),
            1,
            LOGICAL_PAGE_SIZE,
        )
        .await;

        Journal::<_, u64>::init(context.child("recover"), cfg)
            .await
            .unwrap()
    }

    #[test]
    fn test_segmented_capped_init_bounds_torn_suffix_reads() {
        deterministic::Runner::default().start(|context| async move {
            for source_section in [0, 1] {
                for cap in [0, 9] {
                    // Compare a short and long discarded suffix behind the same retained cap.
                    let mut counts = Vec::new();
                    for pages in [16, 4096] {
                        let partition = format!("torn-cap-{source_section}-{cap}-{pages}");
                        let mut page = Vec::new();
                        for value in 0..8u64 {
                            encode_frame_into(None, &value, &mut page).unwrap();
                        }
                        assert_eq!(page.len(), 72);
                        super::super::manager::tests::seed_torn_suffix(
                            &context,
                            &partition,
                            source_section,
                            &page,
                            pages,
                        )
                        .await;

                        // Bounded initialization must read independently of the discarded suffix.
                        let cfg = Config {
                            partition,
                            page_cache: CacheRef::new(NZU16!(72), NZUsize!(4)),
                            write_buffer: NZUsize!(144),
                            compression: None,
                            codec_config: (),
                        };
                        let (recorded, recordings) = RecordingContext::new(context.child("cap"));
                        let mut journal =
                            Journal::<_, u64>::init_at_most(recorded, cfg.clone(), 0, cap)
                                .await
                                .unwrap();
                        let retained = if source_section == 0 { cap } else { 0 };
                        assert_eq!(journal.size(0).unwrap(), retained);
                        counts.push(recordings.snapshot().reads.len());

                        // Appending after the cap must not reconnect any discarded frames.
                        (journal, _, _) = journal.append(0, &999).await.unwrap();
                        drop(journal.sync_all().await.unwrap());
                        let journal = Journal::<_, u64>::init(context.child("reopen"), cfg)
                            .await
                            .unwrap();
                        let mut replay = journal
                            .replay(0, 0, NZUsize!(144), ReadOptions::default())
                            .await
                            .unwrap();
                        let mut actual = Vec::new();
                        while let Some(item) = replay.next().await {
                            actual.push(item.unwrap().3);
                        }
                        replay.finish().unwrap();
                        let expected = if retained == 0 {
                            vec![999]
                        } else {
                            vec![0, 999]
                        };
                        assert_eq!(actual, expected);
                    }

                    // Suffix length must not change the number of reads needed for the same cap.
                    assert_eq!(
                        counts[0], counts[1],
                        "discarded suffix reads grew: {counts:?}"
                    );
                }
            }
        });
    }

    #[test_traced]
    fn test_bounded_initialization_bounds_scan() {
        deterministic::Runner::default().start(|context| async move {
            // Seed different journal lengths under the same page and buffer configuration.
            let mut counts = Vec::new();
            for count in [128, 65536] {
                let cfg = Config {
                    partition: format!("variable-cap-scan-{count}"),
                    page_cache: CacheRef::new(NZU16!(64), NZUsize!(4)),
                    write_buffer: NZUsize!(65536),
                    compression: None,
                    codec_config: (),
                };
                let mut journal = Journal::<_, u64>::init(
                    context.child(if count == 128 {
                        "seed_small"
                    } else {
                        "seed_large"
                    }),
                    cfg.clone(),
                )
                .await
                .unwrap();
                for value in 0..count {
                    (journal, _, _) = journal.append(0, &value).await.unwrap();
                }
                _ = journal.sync_all().await.unwrap();

                // Record the reads required to reopen at the same one-item boundary.
                let (recorded, recordings) =
                    RecordingContext::new(context.child(if count == 128 {
                        "cap_small"
                    } else {
                        "cap_large"
                    }));
                let journal = Journal::<_, u64>::init_at_most(recorded, cfg, 0, 9)
                    .await
                    .unwrap();
                assert_eq!(journal.size(0).unwrap(), 9);
                counts.push(recordings.snapshot().reads.len());
            }

            // Work below the cap must be independent of the discarded journal length.
            assert_eq!(
                counts[0], counts[1],
                "same one-item cap: small/large suffix read counts {counts:?}"
            );
        });
    }

    #[test_traced]
    fn test_bounded_initialization_recovers_torn_page() {
        deterministic::Runner::default().start(|context| async move {
            for cap in [0, 9, 60, 63, 64, 65, 90, 128, 135, u64::MAX] {
                // Seed two sections so the bound must repair one and remove the later section.
                let cfg = Config {
                    partition: format!("capped-torn-{cap}"),
                    page_cache: CacheRef::new(NZU16!(64), NZUsize!(4)),
                    write_buffer: NZUsize!(256),
                    compression: None,
                    codec_config: (),
                };
                let mut journal = Journal::<_, u64>::init(context.child("first"), cfg.clone())
                    .await
                    .unwrap();
                for value in 0..15 {
                    (journal, _, _) = journal.append(0, &value).await.unwrap();
                }
                (journal, _, _) = journal.append(1, &100).await.unwrap();
                _ = journal.sync_all().await.unwrap();

                // A valid later page survives after the torn page in section zero.
                corrupt_page(&context, &cfg.partition, &0u64.to_be_bytes(), 1, 64).await;
                let journal =
                    Journal::<_, u64>::init_at_most(context.child("cap"), cfg.clone(), 0, cap)
                        .await
                        .unwrap();
                let expected_bytes = cap.min(63) / 9 * 9;
                assert_eq!(journal.size(0).unwrap(), expected_bytes, "cap {cap}");
                assert_eq!(journal.newest_section(), Some(0));
                drop(journal);

                // A normal restart must retain the repaired prefix and append at its exact end.
                let journal = Journal::<_, u64>::init(context.child("reopen"), cfg)
                    .await
                    .unwrap();
                let mut replay = journal
                    .replay(0, 0, NZUsize!(256), ReadOptions::default())
                    .await
                    .unwrap();
                let mut count = 0;
                while let Some(item) = replay.next().await {
                    let item = item.unwrap();
                    assert_eq!(item.0, 0);
                    assert_eq!(item.3, count);
                    count += 1;
                }
                assert_eq!(count, expected_bytes / 9);
                let mut journal = replay.finish().unwrap();
                let position;
                (journal, position, _) = journal.append(0, &99).await.unwrap();
                assert_eq!(position, expected_bytes);
                journal.destroy().await.unwrap();
            }
        });
    }

    #[test_traced]
    #[should_panic(expected = "must be replayed before append")]
    fn test_segmented_variable_rejects_append_before_replay() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            const PARTITION: &str = "segmented-variable-append-before-replay";
            const NEW_SECTION: u64 = 2;
            const TORN_SECTION: u64 = 1;

            let journal = journal_with_torn_interior_page(&context, PARTITION, false).await;
            let (journal, offset, _) = journal.append(NEW_SECTION, &15).await.unwrap();
            assert_eq!(offset, 0);
            let (journal, offset, _) = journal.append(NEW_SECTION, &16).await.unwrap();
            assert_eq!(offset, 9);
            let _ = journal.append(TORN_SECTION, &15).await;
        });
    }

    #[test_traced]
    #[should_panic(expected = "must be replayed before append")]
    fn test_segmented_variable_partial_replay_keeps_append_guard() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            const PARTITION: &str = "segmented-variable-partial-replay-append";
            const SECTION: u64 = 1;

            let journal = journal_with_torn_interior_page(&context, PARTITION, false).await;
            let mut replay = journal
                .replay(SECTION, 9, NZUsize!(1024), ReadOptions::default())
                .await
                .unwrap();
            while let Some(item) = replay.next().await {
                item.unwrap();
            }
            let journal = replay.finish().unwrap();
            let _ = journal.append(SECTION, &7).await;
        });
    }

    #[test_traced]
    #[should_panic(expected = "must be replayed before append")]
    fn test_segmented_variable_gates_clean_older_section() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            const PARTITION: &str = "segmented-variable-gate-clean-older-section";

            // Sections 0 and 2 are intact and section 1 is torn. The oldest section is gated
            // even though it is neither torn nor the newest.
            let journal = journal_with_torn_interior_page(&context, PARTITION, true).await;
            let _ = journal.append(0, &7).await;
        });
    }

    #[test_traced]
    fn test_segmented_variable_replay_propagates_read_options() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let (context, recordings) = RecordingContext::new(context);
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(NZU16!(8), PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };
            let mut journal = Journal::init(context.child("storage"), cfg)
                .await
                .expect("failed to init");

            for section in 1..=2 {
                (journal, _, _) = journal
                    .append(section, &section)
                    .await
                    .expect("failed to append");
            }

            let mut replay = journal
                .replay(1, 0, NZUsize!(1036), ReadOptions::DONT_CACHE)
                .await
                .expect("failed to replay");
            recordings.clear();

            // The first lazy refill must carry the caller's policy.
            let (section, offset, _, item) = replay
                .next()
                .await
                .expect("missing first replay item")
                .expect("failed to read first replay item");
            assert_eq!((section, offset, item), (1, 0, 1));
            let reads = recordings.snapshot().reads;
            assert!(!reads.is_empty());
            assert!(
                reads
                    .iter()
                    .all(|options| *options == ReadOptions::DONT_CACHE)
            );

            // Crossing into the next section must preserve the same policy.
            recordings.clear();
            let (section, offset, _, item) = replay
                .next()
                .await
                .expect("missing second replay item")
                .expect("failed to read second replay item");
            assert_eq!((section, offset, item), (2, 0, 2));
            let reads = recordings.snapshot().reads;
            assert!(!reads.is_empty());
            assert!(
                reads
                    .iter()
                    .all(|options| *options == ReadOptions::DONT_CACHE)
            );
            assert!(replay.next().await.is_none());

            let journal = replay.finish().expect("failed to finish replay");
            journal.destroy().await.expect("failed to destroy");
        });
    }

    #[test_traced]
    fn test_journal_append_and_read() {
        // Initialize the deterministic context
        let executor = deterministic::Runner::default();

        // Start the test within the executor
        executor.start(|context| async move {
            // Initialize the journal
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };
            let index = 1u64;
            let data = 10;
            let mut journal = Journal::init(context.child("first"), cfg.clone())
                .await
                .expect("Failed to initialize journal");

            // Append an item to the journal
            (journal, _, _) = journal
                .append(index, &data)
                .await
                .expect("Failed to append data");

            // Check metrics
            let buffer = context.encode();
            assert!(buffer.contains("first_tracked 1"));

            // Drop and re-open the journal to simulate a restart
            journal = journal.sync(index).await.expect("Failed to sync journal");
            drop(journal);
            let journal = Journal::<_, i32>::init(context.child("second"), cfg)
                .await
                .expect("Failed to re-initialize journal");

            // Replay the journal and collect items
            let mut items = Vec::new();
            let mut replay = journal
                .replay(0, 0, NZUsize!(1024), ReadOptions::default())
                .await
                .expect("unable to setup replay");
            while let Some(result) = replay.next().await {
                match result {
                    Ok((blob_index, _, _, item)) => items.push((blob_index, item)),
                    Err(err) => panic!("Failed to read item: {err}"),
                }
            }

            // Verify that the item was replayed correctly
            assert_eq!(items.len(), 1);
            assert_eq!(items[0].0, index);
            assert_eq!(items[0].1, data);

            // Check metrics
            let buffer = context.encode();
            assert!(buffer.contains("second_tracked 1"));
        });
    }

    #[test_traced]
    fn test_journal_multiple_appends_and_reads() {
        // Initialize the deterministic context
        let executor = deterministic::Runner::default();

        // Start the test within the executor
        executor.start(|context| async move {
            // Create a journal configuration
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };

            // Initialize the journal
            let mut journal = Journal::init(context.child("first"), cfg.clone())
                .await
                .expect("Failed to initialize journal");

            // Append multiple items to different blobs
            let data_items = vec![(1u64, 1), (1u64, 2), (2u64, 3), (3u64, 4)];
            for (index, data) in &data_items {
                (journal, _, _) = journal
                    .append(*index, data)
                    .await
                    .expect("Failed to append data");
                journal = journal.sync(*index).await.expect("Failed to sync blob");
            }

            // Check metrics
            let buffer = context.encode();
            assert!(buffer.contains("first_tracked 3"));
            assert!(buffer.contains("first_synced_total 4"));

            // Drop and re-open the journal to simulate a restart
            drop(journal);
            let mut journal = Journal::init(context.child("second"), cfg)
                .await
                .expect("Failed to re-initialize journal");

            // Replay the journal and collect items
            let mut items = Vec::<(u64, u32)>::new();
            {
                let mut replay = journal
                    .replay(0, 0, NZUsize!(1024), ReadOptions::default())
                    .await
                    .expect("unable to setup replay");
                while let Some(result) = replay.next().await {
                    match result {
                        Ok((blob_index, _, _, item)) => items.push((blob_index, item)),
                        Err(err) => panic!("Failed to read item: {err}"),
                    }
                }
                journal = replay.finish().expect("failed to finish replay");
            }

            // Verify that all items were replayed correctly
            assert_eq!(items.len(), data_items.len());
            for ((expected_index, expected_data), (actual_index, actual_data)) in
                data_items.iter().zip(items.iter())
            {
                assert_eq!(actual_index, expected_index);
                assert_eq!(actual_data, expected_data);
            }

            // Cleanup
            journal.destroy().await.expect("Failed to destroy journal");
        });
    }

    #[test_traced]
    fn test_journal_prune_blobs() {
        // Initialize the deterministic context
        let executor = deterministic::Runner::default();

        // Start the test within the executor
        executor.start(|context| async move {
            // Create a journal configuration
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };

            // Initialize the journal
            let mut journal = Journal::init(context.child("first"), cfg.clone())
                .await
                .expect("Failed to initialize journal");

            // Append items to multiple blobs
            for index in 1u64..=5u64 {
                (journal, _, _) = journal
                    .append(index, &index)
                    .await
                    .expect("Failed to append data");
                journal = journal.sync(index).await.expect("Failed to sync blob");
            }

            // Add one item out-of-order
            let data = 99;
            (journal, _, _) = journal
                .append(2u64, &data)
                .await
                .expect("Failed to append data");
            journal = journal.sync(2u64).await.expect("Failed to sync blob");

            // Prune blobs with indices less than 3
            (journal, _) = journal.prune(3).await.expect("Failed to prune blobs");

            // Check metrics
            let buffer = context.encode();
            assert!(buffer.contains("first_pruned_total 2"));

            // Prune again with a section less than the previous one, should be a no-op
            (journal, _) = journal.prune(2).await.expect("Failed to no-op prune");
            let buffer = context.encode();
            assert!(buffer.contains("first_pruned_total 2"));

            // Drop and re-open the journal to simulate a restart
            drop(journal);
            let mut journal = Journal::init(context.child("second"), cfg.clone())
                .await
                .expect("Failed to re-initialize journal");

            // Replay the journal and collect items
            let mut items = Vec::<(u64, u64)>::new();
            {
                let mut replay = journal
                    .replay(0, 0, NZUsize!(1024), ReadOptions::default())
                    .await
                    .expect("unable to setup replay");
                while let Some(result) = replay.next().await {
                    match result {
                        Ok((blob_index, _, _, item)) => items.push((blob_index, item)),
                        Err(err) => panic!("Failed to read item: {err}"),
                    }
                }
                journal = replay.finish().expect("failed to finish replay");
            }

            // Verify that items from blobs 1 and 2 are not present
            assert_eq!(items.len(), 3);
            let expected_indices = [3u64, 4u64, 5u64];
            for (item, expected_index) in items.iter().zip(expected_indices.iter()) {
                assert_eq!(item.0, *expected_index);
            }

            // Prune all blobs
            (journal, _) = journal.prune(6).await.expect("Failed to prune blobs");

            // Drop the journal
            drop(journal);

            // Ensure no remaining blobs exist
            //
            // Note: We don't remove the partition, so this does not error
            // and instead returns an empty list of blobs.
            assert!(
                context
                    .scan(&cfg.partition)
                    .await
                    .expect("Failed to list blobs")
                    .is_empty()
            );
        });
    }

    #[test_traced]
    fn test_journal_prune_guard() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };

            let mut journal = Journal::init(context.child("storage"), cfg.clone())
                .await
                .expect("Failed to initialize journal");

            // Append items to sections 1-5
            for section in 1u64..=5u64 {
                (journal, _, _) = journal
                    .append(section, &(section as i32))
                    .await
                    .expect("Failed to append data");
                journal = journal.sync(section).await.expect("Failed to sync");
            }

            // Prune sections < 3
            (journal, _) = journal.prune(3).await.expect("Failed to prune");

            // The public accessor mirrors the guard
            assert!(journal.pruned(1));
            assert!(journal.pruned(2));
            assert!(!journal.pruned(3));

            // Test that accessing pruned sections returns the correct error

            // Test append on pruned section
            match journal.0.append(1, &100).await {
                Err(Error::AlreadyPrunedToSection(3)) => {}
                other => panic!("Expected AlreadyPrunedToSection(3), got {other:?}"),
            }

            match journal.0.append(2, &100).await {
                Err(Error::AlreadyPrunedToSection(3)) => {}
                other => panic!("Expected AlreadyPrunedToSection(3), got {other:?}"),
            }

            // Test get on pruned section
            match journal.get(1, 0).await {
                Err(Error::AlreadyPrunedToSection(3)) => {}
                other => panic!("Expected AlreadyPrunedToSection(3), got {other:?}"),
            }

            // Test size on pruned section
            match journal.size(1) {
                Err(Error::AlreadyPrunedToSection(3)) => {}
                other => panic!("Expected AlreadyPrunedToSection(3), got {other:?}"),
            }

            // Test that accessing sections at or after the threshold works
            assert!(journal.get(3, 0).await.is_ok());
            assert!(journal.get(4, 0).await.is_ok());
            assert!(journal.get(5, 0).await.is_ok());
            assert!(journal.size(3).is_ok());
            journal = journal
                .sync(4)
                .await
                .expect("Should be able to sync section 4");

            // Append to section at threshold should work
            (journal, _, _) = journal
                .append(3, &999)
                .await
                .expect("Should be able to append to section 3");

            // Prune more sections
            (journal, _) = journal.prune(5).await.expect("Failed to prune");

            // Verify sections 3 and 4 are now pruned
            assert!(journal.pruned(4));
            assert!(!journal.pruned(5));
            match journal.get(3, 0).await {
                Err(Error::AlreadyPrunedToSection(5)) => {}
                other => panic!("Expected AlreadyPrunedToSection(5), got {other:?}"),
            }

            match journal.get(4, 0).await {
                Err(Error::AlreadyPrunedToSection(5)) => {}
                other => panic!("Expected AlreadyPrunedToSection(5), got {other:?}"),
            }

            // Section 5 should still be accessible
            assert!(journal.get(5, 0).await.is_ok());

            // Test sync on pruned section. The failure destroys the journal.
            match journal.sync(4).await {
                Err(Error::AlreadyPrunedToSection(5)) => {}
                other => panic!("Expected AlreadyPrunedToSection(5), got {other:?}"),
            }
        });
    }

    #[test_traced]
    fn test_journal_prune_guard_across_restart() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };

            // First session: create and prune
            {
                let mut journal = Journal::init(context.child("first"), cfg.clone())
                    .await
                    .expect("Failed to initialize journal");

                for section in 1u64..=5u64 {
                    (journal, _, _) = journal
                        .append(section, &(section as i32))
                        .await
                        .expect("Failed to append data");
                    journal = journal.sync(section).await.expect("Failed to sync");
                }

                journal.prune(3).await.expect("Failed to prune");
            }

            // Second session: verify oldest_retained_section is reset
            {
                let journal = Journal::<_, i32>::init(context.child("second"), cfg.clone())
                    .await
                    .expect("Failed to re-initialize journal");

                // The floor is execution-scoped, so pruned reports false after restart
                assert!(!journal.pruned(1));
                assert!(!journal.pruned(2));

                // But the actual sections 1 and 2 should be gone from storage
                // so get should return SectionOutOfRange, not AlreadyPrunedToSection
                match journal.get(1, 0).await {
                    Err(Error::SectionOutOfRange(1)) => {}
                    other => panic!("Expected SectionOutOfRange(1), got {other:?}"),
                }

                match journal.get(2, 0).await {
                    Err(Error::SectionOutOfRange(2)) => {}
                    other => panic!("Expected SectionOutOfRange(2), got {other:?}"),
                }

                // Sections 3-5 should still be accessible
                assert!(journal.get(3, 0).await.is_ok());
                assert!(journal.get(4, 0).await.is_ok());
                assert!(journal.get(5, 0).await.is_ok());
            }
        });
    }

    #[test_traced]
    fn test_journal_with_invalid_blob_name() {
        // Initialize the deterministic context
        let executor = deterministic::Runner::default();

        // Start the test within the executor
        executor.start(|context| async move {
            // Create a journal configuration
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };

            // Manually create a blob with an invalid name (not 8 bytes)
            let invalid_blob_name = b"invalid"; // Less than 8 bytes
            let (blob, _) = context
                .open(&cfg.partition, invalid_blob_name)
                .await
                .expect("Failed to create blob with invalid name");
            blob.sync().await.expect("Failed to sync blob");

            // Attempt to initialize the journal
            drop(blob);
            let result = Journal::<_, u64>::init(context, cfg).await;

            // Expect an error
            assert!(matches!(result, Err(Error::InvalidBlobName(_))));
        });
    }

    #[test_traced]
    fn test_journal_read_size_missing() {
        // Initialize the deterministic context
        let executor = deterministic::Runner::default();

        // Start the test within the executor
        executor.start(|context| async move {
            // Create a journal configuration
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };

            // Manually create a blob with incomplete size data
            let section = 1u64;
            let blob_name = section.to_be_bytes();
            let (blob, _) = context
                .open(&cfg.partition, &blob_name)
                .await
                .expect("Failed to create blob");

            // Write incomplete varint by encoding u32::MAX (5 bytes) and truncating to 1 byte
            let mut incomplete_data = Vec::new();
            UInt(u32::MAX).write(&mut incomplete_data);
            incomplete_data.truncate(1);
            blob.write_at(0, incomplete_data, WriteOptions::SYNC)
                .await
                .expect("Failed to write incomplete data");

            // Initialize the journal
            drop(blob);
            let journal = Journal::init(context, cfg)
                .await
                .expect("Failed to initialize journal");

            // Attempt to replay the journal
            let mut replay = journal
                .replay(0, 0, NZUsize!(1024), ReadOptions::default())
                .await
                .expect("unable to setup replay");
            let mut items = Vec::<(u64, u64)>::new();
            while let Some(result) = replay.next().await {
                match result {
                    Ok((blob_index, _, _, item)) => items.push((blob_index, item)),
                    Err(err) => panic!("Failed to read item: {err}"),
                }
            }
            assert!(items.is_empty());
        });
    }

    #[test_traced]
    fn test_journal_replay_empty_finishes_immediately() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };
            let journal = Journal::<_, i32>::init(context.child("storage"), cfg)
                .await
                .expect("Failed to initialize journal");

            // An empty journal's reader is exhausted from the start
            let replay = journal
                .replay(0, 0, NZUsize!(1024), ReadOptions::default())
                .await
                .expect("Failed to replay");
            let journal = replay.finish().expect("failed to finish replay");
            journal.destroy().await.expect("Failed to destroy");
        });
    }

    #[test_traced]
    fn test_journal_replay_finish_before_drain_fails() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };
            let mut journal = Journal::init(context.child("storage"), cfg)
                .await
                .expect("Failed to initialize journal");
            (journal, _, _) = journal.append(1, &7i32).await.expect("Failed to append");
            journal = journal.sync(1).await.expect("Failed to sync");

            let replay = journal
                .replay(0, 0, NZUsize!(1024), ReadOptions::default())
                .await
                .expect("Failed to replay");
            assert!(matches!(replay.finish(), Err(Error::ReplayFailed)));
        });
    }

    #[test_traced]
    fn test_journal_replay_reports_resize_error_on_trailing_bytes() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };

            // Leave one byte in the first page so the trailing bytes below cross the page
            // boundary and repair must issue a physical resize.
            let section = 1u64;
            let item = [10u8; 1021];
            let item_record_size =
                UInt(item.encode_size() as u32).encode_size() + item.encode_size();
            assert_eq!(item_record_size, PAGE_SIZE.get() as usize - 1);

            let mut journal = Journal::init(context.child("first"), cfg.clone())
                .await
                .expect("Failed to initialize journal");
            (journal, _, _) = journal
                .append(section, &item)
                .await
                .expect("Failed to append item");
            journal
                .0
                .append_raw(section, IoBuf::copy_from_slice(&[0xFF, 0xFF]))
                .await
                .expect("Failed to append trailing bytes");
            journal = journal.sync(section).await.expect("Failed to sync journal");
            drop(journal);

            let journal = Journal::init(context.child("second"), cfg)
                .await
                .expect("Failed to re-initialize journal");
            *context.storage_fault_config().write() = deterministic::FaultConfig {
                resize_rate: Some(deterministic::ResizeConfig {
                    failure_rate: probability!(1.0),
                    partial_rate: probability!(0.0),
                }),
                ..Default::default()
            };

            let mut replay = journal
                .replay(0, 0, NZUsize!(1024), ReadOptions::default())
                .await
                .expect("unable to setup replay");

            let first = replay
                .next()
                .await
                .expect("expected item before trailing bytes")
                .expect("failed to replay valid item");
            assert_eq!(first, (section, 0, item.encode_size() as u32, item));

            // The trailing bytes cross the page boundary, so repair must issue a physical resize.
            match replay.next().await {
                Some(Err(_)) => {}
                other => {
                    panic!("expected resize error while repairing trailing bytes, got {other:?}")
                }
            }
            assert!(replay.next().await.is_none());
        });
    }

    #[test_traced]
    fn test_journal_replay_finish_after_error_fails() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };

            // Same layout as the resize-error test: trailing bytes cross the page
            // boundary so repair must issue a physical resize, which the fault fails.
            let section = 1u64;
            let item = [10u8; 1021];
            let mut journal = Journal::init(context.child("first"), cfg.clone())
                .await
                .expect("Failed to initialize journal");
            (journal, _, _) = journal
                .append(section, &item)
                .await
                .expect("Failed to append item");
            journal
                .0
                .append_raw(section, IoBuf::copy_from_slice(&[0xFF, 0xFF]))
                .await
                .expect("Failed to append trailing bytes");
            journal = journal.sync(section).await.expect("Failed to sync journal");
            drop(journal);

            let journal = Journal::<_, [u8; 1021]>::init(context.child("second"), cfg)
                .await
                .expect("Failed to re-initialize journal");
            *context.storage_fault_config().write() = deterministic::FaultConfig {
                resize_rate: Some(deterministic::ResizeConfig {
                    failure_rate: probability!(1.0),
                    partial_rate: probability!(0.0),
                }),
                ..Default::default()
            };

            let mut replay = journal
                .replay(0, 0, NZUsize!(1024), ReadOptions::default())
                .await
                .expect("unable to setup replay");
            let _ = replay
                .next()
                .await
                .expect("expected item before trailing bytes")
                .expect("failed to replay valid item");
            assert!(matches!(replay.next().await, Some(Err(_))));
            assert!(replay.next().await.is_none());

            // The yielded error is fatal, so finish must refuse to return the journal
            assert!(matches!(replay.finish(), Err(Error::ReplayFailed)));
        });
    }

    #[test_traced]
    fn test_journal_replay_dropped_during_repair_fails_replay() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };

            // Same layout as the resize-error test: trailing bytes cross the page
            // boundary so repair must issue a physical resize (and its sync).
            let section = 1u64;
            let item = [10u8; 1021];
            let mut journal = Journal::init(context.child("first"), cfg.clone())
                .await
                .expect("Failed to initialize journal");
            (journal, _, _) = journal
                .append(section, &item)
                .await
                .expect("Failed to append item");
            journal
                .0
                .append_raw(section, IoBuf::copy_from_slice(&[0xFF, 0xFF]))
                .await
                .expect("Failed to append trailing bytes");
            journal = journal.sync(section).await.expect("Failed to sync journal");
            drop(journal);

            // Gate syncs so the repair suspends, then drop the in-flight next()
            let pending = PendingSyncs::default();
            let gated = DelayedSyncContext {
                inner: context.child("second"),
                pending: pending.clone(),
            };
            let journal = Journal::<_, [u8; 1021]>::init(gated, cfg.clone())
                .await
                .expect("Failed to re-initialize journal");
            let mut replay = journal
                .replay(0, 0, NZUsize!(1024), ReadOptions::default())
                .await
                .expect("unable to setup replay");
            let _ = replay
                .next()
                .await
                .expect("expected item before trailing bytes")
                .expect("failed to replay valid item");
            pending.arm();
            {
                let fut = replay.next();
                futures::pin_mut!(fut);
                assert!(
                    futures::poll!(fut.as_mut()).is_pending(),
                    "repair must suspend on the gated sync"
                );
            }
            release_pending_syncs(&pending);

            // The interrupted repair fails the replay rather than resuming over it
            assert!(matches!(
                replay.next().await,
                Some(Err(Error::ReplayInterrupted))
            ));
            assert!(replay.next().await.is_none());
            drop(replay);

            // Re-initialization repairs from durable state
            let journal = Journal::<_, [u8; 1021]>::init(context.child("third"), cfg)
                .await
                .expect("Failed to re-initialize journal");
            let mut replay = journal
                .replay(0, 0, NZUsize!(1024), ReadOptions::default())
                .await
                .expect("unable to setup replay");
            let first = replay
                .next()
                .await
                .expect("expected item after recovery")
                .expect("failed to replay valid item");
            assert_eq!(first, (section, 0, item.encode_size() as u32, item));
            assert!(replay.next().await.is_none());
            let journal = replay.finish().expect("failed to finish replay");
            journal.destroy().await.expect("Failed to destroy");
        });
    }

    #[test_traced]
    fn test_journal_read_item_missing() {
        // Initialize the deterministic context
        let executor = deterministic::Runner::default();

        // Start the test within the executor
        executor.start(|context| async move {
            // Create a journal configuration
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };

            // Manually create a blob with missing item data
            let section = 1u64;
            let blob_name = section.to_be_bytes();
            let (blob, _) = context
                .open(&cfg.partition, &blob_name)
                .await
                .expect("Failed to create blob");

            // Write size but incomplete item data
            let item_size: u32 = 10; // Size indicates 10 bytes of data
            let mut buf = Vec::new();
            UInt(item_size).write(&mut buf); // Varint encoding
            let data = [2u8; 5];
            BufMut::put_slice(&mut buf, &data);
            blob.write_at(0, buf, WriteOptions::SYNC)
                .await
                .expect("Failed to write incomplete item");

            // Initialize the journal
            drop(blob);
            let journal = Journal::init(context, cfg)
                .await
                .expect("Failed to initialize journal");

            // Attempt to replay the journal
            let mut replay = journal
                .replay(0, 0, NZUsize!(1024), ReadOptions::default())
                .await
                .expect("unable to setup replay");
            let mut items = Vec::<(u64, u64)>::new();
            while let Some(result) = replay.next().await {
                match result {
                    Ok((blob_index, _, _, item)) => items.push((blob_index, item)),
                    Err(err) => panic!("Failed to read item: {err}"),
                }
            }
            assert!(items.is_empty());
        });
    }

    #[test_traced]
    fn test_journal_read_checksum_missing() {
        // Initialize the deterministic context
        let executor = deterministic::Runner::default();

        // Start the test within the executor
        executor.start(|context| async move {
            // Create a journal configuration
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };

            // Manually create a blob with missing checksum
            let section = 1u64;
            let blob_name = section.to_be_bytes();
            let (blob, _) = context
                .open(&cfg.partition, &blob_name)
                .await
                .expect("Failed to create blob");

            // Prepare item data
            let item_data = b"Test data";
            let item_size = item_data.len() as u32;

            // Write size (varint) and data, but no checksum
            let mut buf = Vec::new();
            UInt(item_size).write(&mut buf);
            BufMut::put_slice(&mut buf, item_data);
            blob.write_at(0, buf, WriteOptions::SYNC)
                .await
                .expect("Failed to write item without checksum");

            // Initialize the journal
            drop(blob);
            let journal = Journal::init(context, cfg)
                .await
                .expect("Failed to initialize journal");

            // Attempt to replay the journal
            //
            // This will truncate the leftover bytes from our manual write.
            let mut replay = journal
                .replay(0, 0, NZUsize!(1024), ReadOptions::default())
                .await
                .expect("unable to setup replay");
            let mut items = Vec::<(u64, u64)>::new();
            while let Some(result) = replay.next().await {
                match result {
                    Ok((blob_index, _, _, item)) => items.push((blob_index, item)),
                    Err(err) => panic!("Failed to read item: {err}"),
                }
            }
            assert!(items.is_empty());
        });
    }

    #[test_traced]
    fn test_journal_read_checksum_mismatch() {
        // Initialize the deterministic context
        let executor = deterministic::Runner::default();

        // Start the test within the executor
        executor.start(|context| async move {
            // Create a journal configuration
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };

            // Manually create a blob with incorrect checksum
            let section = 1u64;
            let blob_name = section.to_be_bytes();
            let (blob, _) = context
                .open(&cfg.partition, &blob_name)
                .await
                .expect("Failed to create blob");

            // Prepare item data
            let item_data = b"Test data";
            let item_size = item_data.len() as u32;
            let incorrect_checksum: u32 = 0xDEADBEEF;

            // Write size (varint), data, and incorrect checksum
            let mut buf = Vec::new();
            UInt(item_size).write(&mut buf);
            BufMut::put_slice(&mut buf, item_data);
            buf.put_u32(incorrect_checksum);
            blob.write_at(0, buf, WriteOptions::SYNC)
                .await
                .expect("Failed to write item with bad checksum");

            // Initialize the journal
            drop(blob);
            let mut journal = Journal::init(context.child("storage"), cfg.clone())
                .await
                .expect("Failed to initialize journal");

            // Attempt to replay the journal
            {
                let mut replay = journal
                    .replay(0, 0, NZUsize!(1024), ReadOptions::default())
                    .await
                    .expect("unable to setup replay");
                let mut items = Vec::<(u64, u64)>::new();
                while let Some(result) = replay.next().await {
                    match result {
                        Ok((blob_index, _, _, item)) => items.push((blob_index, item)),
                        Err(err) => panic!("Failed to read item: {err}"),
                    }
                }
                journal = replay.finish().expect("failed to finish replay");
                assert!(items.is_empty());
            }
            drop(journal);

            // Confirm blob is expected length
            let (_, blob_size) = context
                .open(&cfg.partition, &section.to_be_bytes())
                .await
                .expect("Failed to open blob");
            assert_eq!(blob_size, 0);
        });
    }

    #[test_traced]
    fn test_segmented_variable_replay_repairs_torn_interior_page_when_reached() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            const FIRST_SECTION: u64 = 0;
            const PARTITION: &str = "segmented-variable-torn-interior";
            const TORN_SECTION: u64 = 1;

            // Fifteen nine-byte frames occupy 135 bytes. Pages 0 and 2 remain valid while page 1
            // is torn, so backward sizing reports the full tail. Forward page validation must
            // first truncate to 64 bytes. Frame replay can then retain the seven complete frames
            // ending at byte 63 and safely discard the one-byte frame prefix at the page boundary.
            // The replay buffer spans all three pages, proving recovery does not lose page 0 when
            // a prefetched later page fails validation. FIRST_SECTION establishes the ordered
            // lifecycle boundary: replay setup and consumption of an earlier section must not
            // read or repair this later section.
            let journal = journal_with_torn_interior_page(&context, PARTITION, false).await;
            let original_size = context
                .logical_blob(PARTITION, &TORN_SECTION.to_be_bytes())
                .unwrap()
                .len();
            let mut replay = journal
                .replay(FIRST_SECTION, 0, NZUsize!(1024), ReadOptions::default())
                .await
                .unwrap();

            let size = context
                .logical_blob(PARTITION, &TORN_SECTION.to_be_bytes())
                .unwrap()
                .len();
            assert_eq!(
                size, original_size,
                "replay setup must not repair a later section"
            );

            let (section, offset, _, value) = replay.next().await.unwrap().unwrap();
            assert_eq!((section, offset, value), (FIRST_SECTION, 0, u64::MAX));
            let size = context
                .logical_blob(PARTITION, &TORN_SECTION.to_be_bytes())
                .unwrap()
                .len();
            assert_eq!(
                size, original_size,
                "consuming an earlier section must not repair a later section"
            );

            let mut values = Vec::new();
            while let Some(result) = replay.next().await {
                let (section, offset, _, value) = result.unwrap();
                assert_eq!(section, TORN_SECTION);
                assert_eq!(offset, value * 9);
                values.push(value);
            }
            assert_eq!(values, (0..7).collect::<Vec<_>>());

            let journal = replay.finish().unwrap();
            assert_eq!(journal.size(TORN_SECTION).unwrap(), 63);
            let (journal, offset, _) = journal.append(TORN_SECTION, &7).await.unwrap();
            assert_eq!(offset, 63);
            journal.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_segmented_variable_replay_rejects_start_beyond_torn_prefix_without_repair() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            const PARTITION: &str = "segmented-variable-torn-start";
            const SECTION: u64 = 1;
            const START_OFFSET: u64 = 72;

            let journal = journal_with_torn_interior_page(&context, PARTITION, false).await;
            let original_size = context
                .logical_blob(PARTITION, &SECTION.to_be_bytes())
                .unwrap()
                .len();
            let mut replay = journal
                .replay(
                    SECTION,
                    START_OFFSET,
                    NZUsize!(1024),
                    ReadOptions::default(),
                )
                .await
                .expect("apparent tail still covers the requested start");

            assert!(matches!(
                replay.next().await,
                Some(Err(Error::ItemOutOfRange(START_OFFSET)))
            ));
            let size = context
                .logical_blob(PARTITION, &SECTION.to_be_bytes())
                .unwrap()
                .len();
            assert_eq!(
                size, original_size,
                "an unvalidated start offset must not become a repair boundary"
            );
            assert!(matches!(replay.finish(), Err(Error::ReplayFailed)));
        });
    }

    #[test_traced]
    fn test_segmented_variable_replay_stops_after_failed_interior_repair() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let journal = journal_with_torn_interior_page(
                &context,
                "segmented-variable-failed-interior-repair",
                true,
            )
            .await;
            *context.storage_fault_config().write() = deterministic::FaultConfig {
                resize_rate: Some(deterministic::ResizeConfig {
                    failure_rate: probability!(1.0),
                    partial_rate: probability!(0.0),
                }),
                ..Default::default()
            };

            // Section 0 delays validation of the torn section until iteration. Section 2 proves a
            // fatal repair error cannot be treated like an ordinary per-section decode error.
            let mut replay = journal
                .replay(0, 0, NZUsize!(1024), ReadOptions::default())
                .await
                .unwrap();
            assert!(matches!(replay.next().await, Some(Ok((0, 0, _, u64::MAX)))));
            assert!(matches!(replay.next().await, Some(Err(Error::Runtime(_)))));
            assert!(replay.next().await.is_none());
            assert!(matches!(replay.finish(), Err(Error::ReplayFailed)));
        });
    }

    #[test_traced]
    fn test_segmented_variable_replay_stops_after_failed_tail_repair() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let journal = journal_with_torn_interior_page(
                &context,
                "segmented-variable-failed-tail-repair",
                true,
            )
            .await;
            let mut replay = journal
                .replay(0, 0, NZUsize!(1024), ReadOptions::default())
                .await
                .unwrap();

            assert!(matches!(replay.next().await, Some(Ok((0, 0, _, u64::MAX)))));
            for expected in 0..7 {
                let (section, offset, _, value) = replay.next().await.unwrap().unwrap();
                assert_eq!((section, offset, value), (1, expected * 9, expected));
            }

            *context.storage_fault_config().write() = deterministic::FaultConfig {
                write_rate: Some(deterministic::WriteConfig {
                    failure_rate: probability!(1.0),
                    retention_rate: probability!(1.0),
                    mode: deterministic::PartialWriteMode::Prefix,
                }),
                ..Default::default()
            };
            assert!(matches!(replay.next().await, Some(Err(Error::Runtime(_)))));
            *context.storage_fault_config().write() = deterministic::FaultConfig::default();

            assert!(replay.next().await.is_none());
            assert!(matches!(replay.finish(), Err(Error::ReplayFailed)));
        });
    }

    #[test_traced]
    fn test_journal_truncation_recovery() {
        // Initialize the deterministic context
        let executor = deterministic::Runner::default();

        // Start the test within the executor
        executor.start(|context| async move {
            // Create a journal configuration
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };

            // Initialize the journal
            let mut journal = Journal::init(context.child("first"), cfg.clone())
                .await
                .expect("Failed to initialize journal");

            // Append 1 item to the first index
            (journal, _, _) = journal.append(1, &1).await.expect("Failed to append data");

            // Append multiple items to the second section
            let data_items = vec![(2u64, 2), (2u64, 3), (2u64, 4)];
            for (index, data) in &data_items {
                (journal, _, _) = journal
                    .append(*index, data)
                    .await
                    .expect("Failed to append data");
                journal = journal.sync(*index).await.expect("Failed to sync blob");
            }

            // Sync all sections and drop the journal
            journal = journal.sync_all().await.expect("Failed to sync");
            drop(journal);

            // Manually corrupt the end of the second blob
            let (blob, blob_size) = context
                .open(&cfg.partition, &2u64.to_be_bytes())
                .await
                .expect("Failed to open blob");
            blob.resize(blob_size - 4)
                .await
                .expect("Failed to corrupt blob");
            blob.sync().await.expect("Failed to sync blob");

            // Re-initialize the journal to simulate a restart
            drop(blob);
            let mut journal = Journal::init(context.child("second"), cfg.clone())
                .await
                .expect("Failed to re-initialize journal");

            // Attempt to replay the journal
            let mut items = Vec::<(u64, u32)>::new();
            {
                let mut replay = journal
                    .replay(0, 0, NZUsize!(1024), ReadOptions::default())
                    .await
                    .expect("unable to setup replay");
                while let Some(result) = replay.next().await {
                    match result {
                        Ok((blob_index, _, _, item)) => items.push((blob_index, item)),
                        Err(err) => panic!("Failed to read item: {err}"),
                    }
                }
                journal = replay.finish().expect("failed to finish replay");
            }
            drop(journal);

            // Verify that replay stopped after corruption detected (the second blob).
            assert_eq!(items.len(), 1);
            assert_eq!(items[0].0, 1);
            assert_eq!(items[0].1, 1);

            // Confirm second blob was truncated.
            let (_, blob_size) = context
                .open(&cfg.partition, &2u64.to_be_bytes())
                .await
                .expect("Failed to open blob");
            assert_eq!(blob_size, 0);

            // Attempt to replay journal after truncation
            let mut journal = Journal::init(context.child("third"), cfg.clone())
                .await
                .expect("Failed to re-initialize journal");

            // Attempt to replay the journal
            let mut items = Vec::<(u64, u32)>::new();
            {
                let mut replay = journal
                    .replay(0, 0, NZUsize!(1024), ReadOptions::default())
                    .await
                    .expect("unable to setup replay");
                while let Some(result) = replay.next().await {
                    match result {
                        Ok((blob_index, _, _, item)) => items.push((blob_index, item)),
                        Err(err) => panic!("Failed to read item: {err}"),
                    }
                }
                journal = replay.finish().expect("failed to finish replay");
            }

            // Verify that only non-corrupted items were replayed
            assert_eq!(items.len(), 1);
            assert_eq!(items[0].0, 1);
            assert_eq!(items[0].1, 1);

            // Append a new item to truncated partition
            (journal, _, _) = journal.append(2, &5).await.expect("Failed to append data");
            journal = journal.sync(2).await.expect("Failed to sync blob");

            // Get the new item (offset is 0 since blob was truncated)
            let item = journal.get(2, 0).await.expect("Failed to get item");
            assert_eq!(item, 5);

            // Drop the journal (data already synced)
            drop(journal);

            // Re-initialize the journal to simulate a restart
            let journal = Journal::init(context.child("storage"), cfg.clone())
                .await
                .expect("Failed to re-initialize journal");

            // Attempt to replay the journal
            let mut items = Vec::<(u64, u32)>::new();
            {
                let mut replay = journal
                    .replay(0, 0, NZUsize!(1024), ReadOptions::default())
                    .await
                    .expect("unable to setup replay");
                while let Some(result) = replay.next().await {
                    match result {
                        Ok((blob_index, _, _, item)) => items.push((blob_index, item)),
                        Err(err) => panic!("Failed to read item: {err}"),
                    }
                }
            }

            // Verify that only non-corrupted items were replayed
            assert_eq!(items.len(), 2);
            assert_eq!(items[0].0, 1);
            assert_eq!(items[0].1, 1);
            assert_eq!(items[1].0, 2);
            assert_eq!(items[1].1, 5);
        });
    }

    #[test_traced]
    fn test_journal_handling_extra_data() {
        // Initialize the deterministic context
        let executor = deterministic::Runner::default();

        // Start the test within the executor
        executor.start(|context| async move {
            // Create a journal configuration
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };

            // Initialize the journal
            let mut journal = Journal::init(context.child("first"), cfg.clone())
                .await
                .expect("Failed to initialize journal");

            // Append 1 item to the first index
            (journal, _, _) = journal.append(1, &1).await.expect("Failed to append data");

            // Append multiple items to the second index
            let data_items = vec![(2u64, 2), (2u64, 3), (2u64, 4)];
            for (index, data) in &data_items {
                (journal, _, _) = journal
                    .append(*index, data)
                    .await
                    .expect("Failed to append data");
                journal = journal.sync(*index).await.expect("Failed to sync blob");
            }

            // Sync all sections and drop the journal
            journal = journal.sync_all().await.expect("Failed to sync");
            drop(journal);

            // Manually add extra data to the end of the second blob
            let (blob, blob_size) = context
                .open(&cfg.partition, &2u64.to_be_bytes())
                .await
                .expect("Failed to open blob");
            blob.write_at(blob_size, vec![0u8; 16], WriteOptions::SYNC)
                .await
                .expect("Failed to add extra data");

            // Re-initialize the journal to simulate a restart
            drop(blob);
            let journal = Journal::init(context.child("second"), cfg)
                .await
                .expect("Failed to re-initialize journal");

            // Attempt to replay the journal
            let mut items = Vec::<(u64, i32)>::new();
            let mut replay = journal
                .replay(0, 0, NZUsize!(1024), ReadOptions::default())
                .await
                .expect("unable to setup replay");
            while let Some(result) = replay.next().await {
                match result {
                    Ok((blob_index, _, _, item)) => items.push((blob_index, item)),
                    Err(err) => panic!("Failed to read item: {err}"),
                }
            }
        });
    }

    #[test_traced]
    fn test_journal_truncate() {
        // Initialize the deterministic context
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Create journal
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };
            let mut journal = Journal::init(context, cfg).await.unwrap();

            // Check size of non-existent section
            let size = journal.size(1).unwrap();
            assert_eq!(size, 0);

            // Append data to section 1
            (journal, _, _) = journal.append(1, &42i32).await.unwrap();

            // Check size of section 1 - should be greater than 0
            let size = journal.size(1).unwrap();
            assert!(size > 0);

            // Append more data and verify size increases
            (journal, _, _) = journal.append(1, &43i32).await.unwrap();
            let new_size = journal.size(1).unwrap();
            assert!(new_size > size);

            // Check size of different section - should still be 0
            let size = journal.size(2).unwrap();
            assert_eq!(size, 0);

            // Append data to section 2
            (journal, _, _) = journal.append(2, &44i32).await.unwrap();

            // Check size of section 2 - should be greater than 0
            let size = journal.size(2).unwrap();
            assert!(size > 0);

            // Rollback everything in section 1 and 2
            journal = journal.test_reopen_at_most(1, 0).await.unwrap();

            // Check size of section 1 - should be 0
            let size = journal.size(1).unwrap();
            assert_eq!(size, 0);

            // Check size of section 2 - should be 0
            let size = journal.size(2).unwrap();
            assert_eq!(size, 0);
        });
    }

    #[test_traced]
    fn test_journal_truncate_max_section() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };
            let mut journal = Journal::init(context, cfg).await.unwrap();

            // Append to the maximal section. `section + 1` has no representable successor.
            let offset;
            (journal, offset, _) = journal.append(u64::MAX, &42i32).await.unwrap();
            let size = journal.size(u64::MAX).unwrap();
            assert!(size > 0);

            // Truncating the maximal section removes no sections above it and must not panic.
            journal = journal.test_reopen_at_most(u64::MAX, size).await.unwrap();

            // The section is intact and readable.
            assert_eq!(journal.size(u64::MAX).unwrap(), size);
            assert_eq!(journal.get(u64::MAX, offset).await.unwrap(), 42i32);
        });
    }

    #[test_traced]
    fn test_journal_truncate_section() {
        // Initialize the deterministic context
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Create journal
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };
            let mut journal = Journal::init(context, cfg).await.unwrap();

            // Check size of non-existent section
            let size = journal.size(1).unwrap();
            assert_eq!(size, 0);

            // Append data to section 1
            (journal, _, _) = journal.append(1, &42i32).await.unwrap();

            // Check size of section 1 - should be greater than 0
            let size = journal.size(1).unwrap();
            assert!(size > 0);

            // Append more data and verify size increases
            (journal, _, _) = journal.append(1, &43i32).await.unwrap();
            let new_size = journal.size(1).unwrap();
            assert!(new_size > size);

            // Check size of different section - should still be 0
            let size = journal.size(2).unwrap();
            assert_eq!(size, 0);

            // Append data to section 2
            (journal, _, _) = journal.append(2, &44i32).await.unwrap();

            // Check size of section 2 - should be greater than 0
            let size = journal.size(2).unwrap();
            assert!(size > 0);

            // Rollback everything in section 1
            journal = journal.test_reopen_section(1, 0).await.unwrap();

            // Check size of section 1 - should be 0
            let size = journal.size(1).unwrap();
            assert_eq!(size, 0);

            // Check size of section 2 - should be greater than 0
            let size = journal.size(2).unwrap();
            assert!(size > 0);
        });
    }

    #[test_traced]
    fn test_journal_small_items() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };

            let mut journal = Journal::init(context.child("first"), cfg.clone())
                .await
                .expect("Failed to initialize journal");

            // Append many small (1-byte) items to the same section
            let num_items = 100;
            let mut offsets = Vec::new();
            for i in 0..num_items {
                let offset;
                let size;
                (journal, offset, size) = journal
                    .append(1, &(i as u8))
                    .await
                    .expect("Failed to append data");
                assert_eq!(size, 1, "u8 should encode to 1 byte");
                offsets.push(offset);
            }
            journal = journal.sync(1).await.expect("Failed to sync");

            // Read each item back via random access
            for (i, &offset) in offsets.iter().enumerate() {
                let item: u8 = journal.get(1, offset).await.expect("Failed to get item");
                assert_eq!(item, i as u8, "Item mismatch at offset {offset}");
            }

            // Drop and reopen to test replay
            drop(journal);
            let journal = Journal::<_, u8>::init(context.child("second"), cfg)
                .await
                .expect("Failed to re-initialize journal");

            // Replay and verify all items
            let mut replay = journal
                .replay(0, 0, NZUsize!(1024), ReadOptions::default())
                .await
                .expect("Failed to setup replay");

            let mut count = 0;
            while let Some(result) = replay.next().await {
                let (section, offset, size, item) = result.expect("Failed to replay item");
                assert_eq!(section, 1);
                assert_eq!(offset, offsets[count]);
                assert_eq!(size, 1);
                assert_eq!(item, count as u8);
                count += 1;
            }
            assert_eq!(count, num_items, "Should replay all items");
        });
    }

    #[test_traced]
    fn test_journal_truncate_many_sections() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };
            let mut journal = Journal::init(context.child("storage"), cfg.clone())
                .await
                .unwrap();

            // Create sections 1-10 with data
            for section in 1u64..=10 {
                (journal, _, _) = journal.append(section, &(section as i32)).await.unwrap();
            }
            journal = journal.sync_all().await.unwrap();

            // Verify all sections exist
            for section in 1u64..=10 {
                let size = journal.size(section).unwrap();
                assert!(size > 0, "section {section} should have data");
            }

            // Truncate to section 5 (should remove sections 6-10)
            let size = journal.size(5).unwrap();
            journal = journal.test_reopen_at_most(5, size).await.unwrap();

            // Verify sections 1-5 still exist with correct data
            for section in 1u64..=5 {
                let size = journal.size(section).unwrap();
                assert!(size > 0, "section {section} should still have data");
            }

            // Verify sections 6-10 are removed (size should be 0)
            for section in 6u64..=10 {
                let size = journal.size(section).unwrap();
                assert_eq!(size, 0, "section {section} should be removed");
            }

            // Verify data integrity via replay
            {
                let mut replay = journal
                    .replay(0, 0, NZUsize!(1024), ReadOptions::default())
                    .await
                    .unwrap();
                let mut items = Vec::new();
                while let Some(result) = replay.next().await {
                    let (section, _, _, item) = result.unwrap();
                    items.push((section, item));
                }
                journal = replay.finish().expect("failed to finish replay");
                assert_eq!(items.len(), 5);
                for (i, (section, item)) in items.iter().enumerate() {
                    assert_eq!(*section, (i + 1) as u64);
                    assert_eq!(*item, (i + 1) as i32);
                }
            }

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

    #[test_traced]
    fn test_journal_truncate_partial_truncation() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };
            let mut journal = Journal::init(context.child("storage"), cfg.clone())
                .await
                .unwrap();

            // Append 5 items and record sizes after each
            let mut sizes = Vec::new();
            for i in 0..5 {
                (journal, _, _) = journal.append(1, &i).await.unwrap();
                journal = journal.sync(1).await.unwrap();
                sizes.push(journal.size(1).unwrap());
            }

            // Truncate to keep only first 3 items
            let target_size = sizes[2];
            journal = journal.test_reopen_at_most(1, target_size).await.unwrap();

            // Verify size is correct
            let new_size = journal.size(1).unwrap();
            assert_eq!(new_size, target_size);

            // Verify first 3 items via replay
            {
                let mut replay = journal
                    .replay(0, 0, NZUsize!(1024), ReadOptions::default())
                    .await
                    .unwrap();
                let mut items = Vec::new();
                while let Some(result) = replay.next().await {
                    let (_, _, _, item) = result.unwrap();
                    items.push(item);
                }
                journal = replay.finish().expect("failed to finish replay");
                assert_eq!(items.len(), 3);
                for (i, item) in items.iter().enumerate() {
                    assert_eq!(*item, i as i32);
                }
            }

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

    #[test_traced]
    fn test_journal_truncate_nonexistent_target() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };
            let mut journal = Journal::init(context.child("storage"), cfg.clone())
                .await
                .unwrap();

            // Create sections 5, 6, 7 (skip 1-4)
            for section in 5u64..=7 {
                (journal, _, _) = journal.append(section, &(section as i32)).await.unwrap();
            }
            journal = journal.sync_all().await.unwrap();

            // Truncate to section 3 (doesn't exist)
            journal = journal.test_reopen_at_most(3, 0).await.unwrap();

            // Verify sections 5, 6, 7 are removed
            for section in 5u64..=7 {
                let size = journal.size(section).unwrap();
                assert_eq!(size, 0, "section {section} should be removed");
            }

            // Verify replay returns nothing
            {
                let mut replay = journal
                    .replay(0, 0, NZUsize!(1024), ReadOptions::default())
                    .await
                    .unwrap();
                assert!(replay.next().await.is_none());
                journal = replay.finish().expect("failed to finish replay");
            }

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

    #[test_traced]
    fn test_journal_truncate_persistence() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };

            // Create sections 1-5 with data
            let mut journal = Journal::init(context.child("first"), cfg.clone())
                .await
                .unwrap();
            for section in 1u64..=5 {
                (journal, _, _) = journal.append(section, &(section as i32)).await.unwrap();
            }
            journal = journal.sync_all().await.unwrap();

            // Truncate to section 2
            let size = journal.size(2).unwrap();
            journal = journal.test_reopen_at_most(2, size).await.unwrap();
            journal = journal.sync_all().await.unwrap();
            drop(journal);

            // Re-init and verify only sections 1-2 exist
            let mut journal = Journal::<_, i32>::init(context.child("second"), cfg.clone())
                .await
                .unwrap();

            // Verify sections 1-2 have data
            for section in 1u64..=2 {
                let size = journal.size(section).unwrap();
                assert!(size > 0, "section {section} should have data after restart");
            }

            // Verify sections 3-5 are gone
            for section in 3u64..=5 {
                let size = journal.size(section).unwrap();
                assert_eq!(size, 0, "section {section} should be gone after restart");
            }

            // Verify data integrity via replay
            {
                let mut replay = journal
                    .replay(0, 0, NZUsize!(1024), ReadOptions::default())
                    .await
                    .unwrap();
                let mut items = Vec::new();
                while let Some(result) = replay.next().await {
                    let (section, _, _, item) = result.unwrap();
                    items.push((section, item));
                }
                journal = replay.finish().expect("failed to finish replay");
                assert_eq!(items.len(), 2);
                assert_eq!(items[0], (1, 1));
                assert_eq!(items[1], (2, 2));
            }

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

    #[test_traced]
    fn test_journal_truncate_to_zero_removes_all_newer() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };
            let mut journal = Journal::init(context.child("storage"), cfg.clone())
                .await
                .unwrap();

            // Create sections 1, 2, 3
            for section in 1u64..=3 {
                (journal, _, _) = journal.append(section, &(section as i32)).await.unwrap();
            }
            journal = journal.sync_all().await.unwrap();

            // Truncate section 1 to size 0
            journal = journal.test_reopen_at_most(1, 0).await.unwrap();

            // Verify section 1 exists but is empty
            let size = journal.size(1).unwrap();
            assert_eq!(size, 0, "section 1 should be empty");

            // Verify sections 2, 3 are completely removed
            for section in 2u64..=3 {
                let size = journal.size(section).unwrap();
                assert_eq!(size, 0, "section {section} should be removed");
            }

            // Verify replay returns nothing
            {
                let mut replay = journal
                    .replay(0, 0, NZUsize!(1024), ReadOptions::default())
                    .await
                    .unwrap();
                assert!(replay.next().await.is_none());
                journal = replay.finish().expect("failed to finish replay");
            }

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

    #[test_traced]
    fn test_journal_replay_start_offset_with_trailing_bytes() {
        // Regression: valid_offset must be initialized to start_offset, not 0.
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };
            let mut journal = Journal::init(context.child("first"), cfg.clone())
                .await
                .expect("Failed to initialize journal");

            // Append several items to build up valid data
            for i in 0..5i32 {
                (journal, _, _) = journal.append(1, &i).await.unwrap();
            }
            journal = journal.sync(1).await.unwrap();
            let valid_logical_size = journal.size(1).unwrap();
            drop(journal);

            // Get the physical blob size before corruption
            let (blob, physical_size_before) = context
                .open(&cfg.partition, &1u64.to_be_bytes())
                .await
                .unwrap();

            // Write incomplete varint: 0xFF has continuation bit set, needs more bytes
            // This creates 2 trailing bytes that cannot form a valid item
            blob.write_at(physical_size_before, vec![0xFF, 0xFF], WriteOptions::SYNC)
                .await
                .unwrap();

            // Reopen journal and replay starting PAST all valid items
            // (start_offset = valid_logical_size means we skip all valid data)
            // The first thing encountered will be the trailing corrupt bytes
            let start_offset = valid_logical_size;
            {
                drop(blob);
                let journal = Journal::<_, i32>::init(context.child("second"), cfg.clone())
                    .await
                    .unwrap();

                let mut replay = journal
                    .replay(1, start_offset, NZUsize!(1024), ReadOptions::default())
                    .await
                    .unwrap();

                // Consume the reader - should detect trailing bytes and truncate
                while let Some(_result) = replay.next().await {}
            }

            // Verify that valid data before start_offset was NOT lost
            let physical_size_after = context
                .logical_blob(&cfg.partition, &1u64.to_be_bytes())
                .unwrap()
                .len() as u64;

            // The blob should have been truncated back to the valid physical size
            // (removing the trailing corrupt bytes) but NOT to 0
            assert!(
                physical_size_after >= physical_size_before,
                "Valid data was lost! Physical blob truncated from {physical_size_before} to \
                 {physical_size_after}. Logical valid size was {valid_logical_size}. \
                 This indicates valid_offset was incorrectly initialized to 0 instead of start_offset."
            );
        });
    }

    #[test_traced]
    fn test_journal_replay_rejects_mid_frame_offset() {
        deterministic::Runner::default().start(|context| async move {
            let cfg = Config {
                partition: "mid-frame".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };
            let journal = Journal::<_, [u8; 16]>::init(context.child("storage"), cfg.clone())
                .await
                .unwrap();
            let (journal, _, _) = journal.append(1, &[127; 16]).await.unwrap();
            let journal = journal.sync_all().await.unwrap();

            // The first payload byte looks like a length larger than the remaining frame.
            let mut replay = journal
                .replay(1, 1, NZUsize!(1024), ReadOptions::default())
                .await
                .unwrap();
            assert!(matches!(
                replay.next().await,
                Some(Err(Error::ItemOutOfRange(1)))
            ));
            drop(replay);

            let journal = Journal::<_, [u8; 16]>::init(context.child("reopen"), cfg)
                .await
                .unwrap();
            let mut replay = journal
                .replay(1, 0, NZUsize!(1024), ReadOptions::default())
                .await
                .unwrap();
            assert_eq!(replay.next().await.unwrap().unwrap().3, [127; 16]);
            assert!(replay.next().await.is_none());
            replay.finish().unwrap().destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_journal_replay_rejects_start_offset_past_section() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };
            let mut journal = Journal::init(context.child("storage"), cfg).await.unwrap();
            (journal, _, _) = journal.append(1, &7i32).await.unwrap();

            // A failed replay consumes the journal
            let result = journal
                .replay(1, u64::MAX, NZUsize!(1024), ReadOptions::default())
                .await;
            assert!(matches!(result, Err(Error::ItemOutOfRange(u64::MAX))));
        });
    }

    #[test_traced]
    fn test_journal_large_item_spanning_pages() {
        // 2048 bytes spans 2 full pages (PAGE_SIZE = 1024).
        const LARGE_SIZE: usize = 2048;
        type LargeItem = [u8; LARGE_SIZE];

        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(4096),
            };
            let mut journal = Journal::init(context.child("first"), cfg.clone())
                .await
                .expect("Failed to initialize journal");

            // Create a large item that spans multiple pages.
            let mut large_data: LargeItem = [0u8; LARGE_SIZE];
            for (i, byte) in large_data.iter_mut().enumerate() {
                *byte = (i % 256) as u8;
            }
            assert!(
                LARGE_SIZE > PAGE_SIZE.get() as usize,
                "Item must be larger than page size"
            );

            // Append the large item
            let offset;
            let size;
            (journal, offset, size) = journal
                .append(1, &large_data)
                .await
                .expect("Failed to append large item");
            assert_eq!(size as usize, LARGE_SIZE);
            journal = journal.sync(1).await.expect("Failed to sync");

            // Read the item back via random access
            let retrieved: LargeItem = journal
                .get(1, offset)
                .await
                .expect("Failed to get large item");
            assert_eq!(retrieved, large_data, "Random access read mismatch");

            // Drop and reopen to test replay
            drop(journal);
            let mut journal = Journal::<_, LargeItem>::init(context.child("second"), cfg.clone())
                .await
                .expect("Failed to re-initialize journal");

            // Replay and verify the large item
            {
                let mut replay = journal
                    .replay(0, 0, NZUsize!(1024), ReadOptions::default())
                    .await
                    .expect("Failed to setup replay");

                let mut items = Vec::new();
                while let Some(result) = replay.next().await {
                    let (section, off, sz, item) = result.expect("Failed to replay item");
                    items.push((section, off, sz, item));
                }
                journal = replay.finish().expect("failed to finish replay");

                assert_eq!(items.len(), 1, "Should have exactly one item");
                let (section, off, sz, item) = &items[0];
                assert_eq!(*section, 1);
                assert_eq!(*off, offset);
                assert_eq!(*sz as usize, LARGE_SIZE);
                assert_eq!(*item, large_data, "Replay read mismatch");
            }

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

    #[test_traced]
    fn test_journal_large_item_direct_path() {
        // Items larger than the write buffer are written directly to the blob. The first append
        // takes the direct path from an empty tip; the second takes it with a non-empty tip
        // (holding the first item's sub-page remainder), covering both top-up branches. The
        // returned offsets must remain correct since callers persist them for random access.
        const LARGE_SIZE: usize = 2048;
        type LargeItem = [u8; LARGE_SIZE];

        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };
            let mut journal = Journal::init(context.child("first"), cfg.clone())
                .await
                .expect("Failed to initialize journal");

            let mut first: LargeItem = [0u8; LARGE_SIZE];
            for (i, byte) in first.iter_mut().enumerate() {
                *byte = (i % 256) as u8;
            }
            let mut second: LargeItem = [0u8; LARGE_SIZE];
            for (i, byte) in second.iter_mut().enumerate() {
                *byte = ((i + 7) % 251) as u8;
            }

            let first_offset;
            (journal, first_offset, _) = journal
                .append(1, &first)
                .await
                .expect("Failed to append first item");
            let second_offset;
            (journal, second_offset, _) = journal
                .append(1, &second)
                .await
                .expect("Failed to append second item");

            // Both items are readable at their returned offsets before any sync.
            let retrieved: LargeItem = journal.get(1, first_offset).await.unwrap();
            assert_eq!(retrieved, first);
            let retrieved: LargeItem = journal.get(1, second_offset).await.unwrap();
            assert_eq!(retrieved, second);

            // Everything survives a sync and reopen.
            journal = journal.sync(1).await.expect("Failed to sync");
            drop(journal);
            let mut journal = Journal::<_, LargeItem>::init(context.child("second"), cfg.clone())
                .await
                .expect("Failed to re-initialize journal");

            let retrieved: LargeItem = journal.get(1, first_offset).await.unwrap();
            assert_eq!(retrieved, first);
            let retrieved: LargeItem = journal.get(1, second_offset).await.unwrap();
            assert_eq!(retrieved, second);

            {
                let mut replay = journal
                    .replay(0, 0, NZUsize!(1024), ReadOptions::default())
                    .await
                    .expect("Failed to setup replay");

                let mut items = Vec::new();
                while let Some(result) = replay.next().await {
                    let (section, off, _, item) = result.expect("Failed to replay item");
                    items.push((section, off, item));
                }
                journal = replay.finish().expect("failed to finish replay");
                assert_eq!(items.len(), 2);
                assert_eq!(items[0], (1, first_offset, first));
                assert_eq!(items[1], (1, second_offset, second));
            }

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

    #[test_traced]
    fn test_journal_non_contiguous_sections() {
        // Test that sections with gaps in numbering work correctly.
        // Sections 1, 5, 10 should all be independent and accessible.
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };
            let mut journal = Journal::init(context.child("first"), cfg.clone())
                .await
                .expect("Failed to initialize journal");

            // Create sections with gaps: 1, 5, 10
            let sections_and_data = [(1u64, 100i32), (5u64, 500i32), (10u64, 1000i32)];
            let mut offsets = Vec::new();

            for (section, data) in &sections_and_data {
                let offset;
                (journal, offset, _) = journal
                    .append(*section, data)
                    .await
                    .expect("Failed to append");
                offsets.push(offset);
            }
            journal = journal.sync_all().await.expect("Failed to sync");

            // Verify random access to each section
            for (i, (section, expected_data)) in sections_and_data.iter().enumerate() {
                let retrieved: i32 = journal
                    .get(*section, offsets[i])
                    .await
                    .expect("Failed to get item");
                assert_eq!(retrieved, *expected_data);
            }

            // Verify non-existent sections return appropriate errors
            for missing_section in [0u64, 2, 3, 4, 6, 7, 8, 9, 11] {
                let result = journal.get(missing_section, 0).await;
                assert!(
                    matches!(result, Err(Error::SectionOutOfRange(_))),
                    "Expected SectionOutOfRange for section {}, got {:?}",
                    missing_section,
                    result
                );
            }

            // Drop and reopen to test replay
            drop(journal);
            let mut journal = Journal::<_, i32>::init(context.child("second"), cfg.clone())
                .await
                .expect("Failed to re-initialize journal");

            // Replay and verify all items in order
            {
                let mut replay = journal
                    .replay(0, 0, NZUsize!(1024), ReadOptions::default())
                    .await
                    .expect("Failed to setup replay");

                let mut items = Vec::new();
                while let Some(result) = replay.next().await {
                    let (section, _, _, item) = result.expect("Failed to replay item");
                    items.push((section, item));
                }
                journal = replay.finish().expect("failed to finish replay");

                assert_eq!(items.len(), 3, "Should have 3 items");
                assert_eq!(items[0], (1, 100));
                assert_eq!(items[1], (5, 500));
                assert_eq!(items[2], (10, 1000));
            }

            // Test replay starting from middle section (5)
            {
                let mut replay = journal
                    .replay(5, 0, NZUsize!(1024), ReadOptions::default())
                    .await
                    .expect("Failed to setup replay from section 5");

                let mut items = Vec::new();
                while let Some(result) = replay.next().await {
                    let (section, _, _, item) = result.expect("Failed to replay item");
                    items.push((section, item));
                }
                journal = replay.finish().expect("failed to finish replay");

                assert_eq!(items.len(), 2, "Should have 2 items from section 5 onwards");
                assert_eq!(items[0], (5, 500));
                assert_eq!(items[1], (10, 1000));
            }

            // Test replay starting from non-existent section (should skip to next)
            {
                let mut replay = journal
                    .replay(3, 0, NZUsize!(1024), ReadOptions::default())
                    .await
                    .expect("Failed to setup replay from section 3");

                let mut items = Vec::new();
                while let Some(result) = replay.next().await {
                    let (section, _, _, item) = result.expect("Failed to replay item");
                    items.push((section, item));
                }
                journal = replay.finish().expect("failed to finish replay");

                // Should get sections 5 and 10 (skipping non-existent 3, 4)
                assert_eq!(items.len(), 2);
                assert_eq!(items[0], (5, 500));
                assert_eq!(items[1], (10, 1000));
            }

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

    #[test_traced]
    fn test_journal_empty_section_in_middle() {
        // Test that replay correctly handles an empty section between sections with data.
        // Section 1 has data, section 2 is empty, section 3 has data.
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };
            let mut journal = Journal::init(context.child("first"), cfg.clone())
                .await
                .expect("Failed to initialize journal");

            // Append to section 1
            (journal, _, _) = journal.append(1, &100i32).await.expect("Failed to append");

            // Create section 2, then truncate it to empty.
            (journal, _, _) = journal.append(2, &200i32).await.expect("Failed to append");
            journal = journal.sync(2).await.expect("Failed to sync");
            journal = journal
                .test_reopen_section(2, 0)
                .await
                .expect("Failed to truncate");

            // Append to section 3
            (journal, _, _) = journal.append(3, &300i32).await.expect("Failed to append");

            journal = journal.sync_all().await.expect("Failed to sync");

            // Verify section sizes
            assert!(journal.size(1).unwrap() > 0);
            assert_eq!(journal.size(2).unwrap(), 0);
            assert!(journal.size(3).unwrap() > 0);

            // Drop and reopen to test replay
            drop(journal);
            let mut journal = Journal::<_, i32>::init(context.child("second"), cfg.clone())
                .await
                .expect("Failed to re-initialize journal");

            // Replay all - should get items from sections 1 and 3, skipping empty section 2
            {
                let mut replay = journal
                    .replay(0, 0, NZUsize!(1024), ReadOptions::default())
                    .await
                    .expect("Failed to setup replay");

                let mut items = Vec::new();
                while let Some(result) = replay.next().await {
                    let (section, _, _, item) = result.expect("Failed to replay item");
                    items.push((section, item));
                }
                journal = replay.finish().expect("failed to finish replay");

                assert_eq!(
                    items.len(),
                    2,
                    "Should have 2 items (skipping empty section)"
                );
                assert_eq!(items[0], (1, 100));
                assert_eq!(items[1], (3, 300));
            }

            // Replay starting from empty section 2 - should get only section 3
            {
                let mut replay = journal
                    .replay(2, 0, NZUsize!(1024), ReadOptions::default())
                    .await
                    .expect("Failed to setup replay from section 2");

                let mut items = Vec::new();
                while let Some(result) = replay.next().await {
                    let (section, _, _, item) = result.expect("Failed to replay item");
                    items.push((section, item));
                }
                journal = replay.finish().expect("failed to finish replay");

                assert_eq!(items.len(), 1, "Should have 1 item from section 3");
                assert_eq!(items[0], (3, 300));
            }

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

    #[test_traced]
    fn test_journal_item_exactly_page_size() {
        // Test that items exactly equal to PAGE_SIZE work correctly.
        // This is a boundary condition where item fills exactly one page.
        const ITEM_SIZE: usize = PAGE_SIZE.get() as usize;
        type ExactItem = [u8; ITEM_SIZE];

        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(4096),
            };
            let mut journal = Journal::init(context.child("first"), cfg.clone())
                .await
                .expect("Failed to initialize journal");

            // Create an item exactly PAGE_SIZE bytes
            let mut exact_data: ExactItem = [0u8; ITEM_SIZE];
            for (i, byte) in exact_data.iter_mut().enumerate() {
                *byte = (i % 256) as u8;
            }

            // Append the exact-size item
            let offset;
            let size;
            (journal, offset, size) = journal
                .append(1, &exact_data)
                .await
                .expect("Failed to append exact item");
            assert_eq!(size as usize, ITEM_SIZE);
            journal = journal.sync(1).await.expect("Failed to sync");

            // Read the item back via random access
            let retrieved: ExactItem = journal
                .get(1, offset)
                .await
                .expect("Failed to get exact item");
            assert_eq!(retrieved, exact_data, "Random access read mismatch");

            // Drop and reopen to test replay
            drop(journal);
            let mut journal = Journal::<_, ExactItem>::init(context.child("second"), cfg.clone())
                .await
                .expect("Failed to re-initialize journal");

            // Replay and verify
            {
                let mut replay = journal
                    .replay(0, 0, NZUsize!(1024), ReadOptions::default())
                    .await
                    .expect("Failed to setup replay");

                let mut items = Vec::new();
                while let Some(result) = replay.next().await {
                    let (section, off, sz, item) = result.expect("Failed to replay item");
                    items.push((section, off, sz, item));
                }
                journal = replay.finish().expect("failed to finish replay");

                assert_eq!(items.len(), 1, "Should have exactly one item");
                let (section, off, sz, item) = &items[0];
                assert_eq!(*section, 1);
                assert_eq!(*off, offset);
                assert_eq!(*sz as usize, ITEM_SIZE);
                assert_eq!(*item, exact_data, "Replay read mismatch");
            }

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

    #[test_traced]
    fn test_journal_uncompressed_frames_fill_write_buffer() {
        deterministic::Runner::default().start(|context| async move {
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: ((..).into(), ()),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(2048),
            };
            let direct = vec![7u8; 126];
            let fallback = vec![9u8; 127];
            assert_eq!(direct.encode_size(), 127);
            assert_eq!(fallback.encode_size(), 128);

            let mut journal = Journal::init(context.child("first"), cfg.clone())
                .await
                .unwrap();

            // Sixteen 128-byte frames exactly fill the write buffer, so the next 130-byte frame
            // must take the owned fallback.
            for i in 0..16 {
                let offset;
                let item_len;
                (journal, offset, item_len) = journal.append(1, &direct).await.unwrap();
                assert_eq!(offset, i * 128);
                assert_eq!(item_len, 127);
            }
            let fallback_offset;
            let fallback_len;
            (journal, fallback_offset, fallback_len) = journal.append(1, &fallback).await.unwrap();
            assert_eq!(fallback_offset, 2048);
            assert_eq!(fallback_len, 128);

            // Reopen the section to verify both append paths on disk.
            journal = journal.sync(1).await.unwrap();
            drop(journal);
            let mut journal = Journal::<_, Vec<u8>>::init(context.child("second"), cfg)
                .await
                .unwrap();
            assert_eq!(journal.get(1, fallback_offset).await.unwrap(), fallback);

            // Replay returns the direct frames followed by the fallback frame.
            let mut replay = journal
                .replay(0, 0, NZUsize!(1024), ReadOptions::default())
                .await
                .unwrap();
            for i in 0..16 {
                let (section, offset, item_len, item) = replay.next().await.unwrap().unwrap();
                assert_eq!((section, offset, item_len), (1, i * 128, 127));
                assert_eq!(item, direct);
            }
            let (section, offset, item_len, item) = replay.next().await.unwrap().unwrap();
            assert_eq!((section, offset, item_len), (1, 2048, 128));
            assert_eq!(item, fallback);
            assert!(replay.next().await.is_none());
            journal = replay.finish().unwrap();
            journal.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_journal_varint_spanning_page_boundary() {
        // Test that items with data spanning page boundaries work correctly
        // when using a small page size.
        //
        // With PAGE_SIZE=16:
        // - Physical page = 16 + 12 = 28 bytes
        // - Each [u8; 128] item = 2-byte varint + 128 bytes data = 130 bytes
        // - This spans multiple 16-byte pages, testing cross-page reading
        const SMALL_PAGE: NonZeroU16 = NZU16!(16);

        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                partition: "test-partition".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(SMALL_PAGE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };
            let mut journal: Journal<_, [u8; 128]> =
                Journal::init(context.child("first"), cfg.clone())
                    .await
                    .expect("Failed to initialize journal");

            // Create items that will span many 16-byte pages
            let item1: [u8; 128] = [1u8; 128];
            let item2: [u8; 128] = [2u8; 128];
            let item3: [u8; 128] = [3u8; 128];

            // Append items - each is 130 bytes (2-byte varint + 128 data)
            // spanning ceil(130/16) = 9 pages worth of logical data
            let offset1;
            (journal, offset1, _) = journal.append(1, &item1).await.expect("Failed to append");
            let offset2;
            (journal, offset2, _) = journal.append(1, &item2).await.expect("Failed to append");
            let offset3;
            (journal, offset3, _) = journal.append(1, &item3).await.expect("Failed to append");

            journal = journal.sync(1).await.expect("Failed to sync");

            // Read items back via random access
            let retrieved1: [u8; 128] = journal.get(1, offset1).await.expect("Failed to get");
            let retrieved2: [u8; 128] = journal.get(1, offset2).await.expect("Failed to get");
            let retrieved3: [u8; 128] = journal.get(1, offset3).await.expect("Failed to get");
            assert_eq!(retrieved1, item1);
            assert_eq!(retrieved2, item2);
            assert_eq!(retrieved3, item3);

            // Drop and reopen to test replay
            drop(journal);
            let mut journal: Journal<_, [u8; 128]> =
                Journal::init(context.child("second"), cfg.clone())
                    .await
                    .expect("Failed to re-initialize journal");

            // Replay and verify all items
            {
                let mut replay = journal
                    .replay(0, 0, NZUsize!(64), ReadOptions::default())
                    .await
                    .expect("Failed to setup replay");

                let mut items = Vec::new();
                while let Some(result) = replay.next().await {
                    let (section, off, _, item) = result.expect("Failed to replay item");
                    items.push((section, off, item));
                }
                journal = replay.finish().expect("failed to finish replay");

                assert_eq!(items.len(), 3, "Should have 3 items");
                assert_eq!(items[0], (1, offset1, item1));
                assert_eq!(items[1], (1, offset2, item2));
                assert_eq!(items[2], (1, offset3, item3));
            }

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

    #[test_traced]
    fn test_journal_clear() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = Config {
                partition: "clear-test".into(),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };

            let mut journal: Journal<_, u64> = Journal::init(context.child("journal"), cfg.clone())
                .await
                .expect("Failed to initialize journal");

            // Append items across multiple sections
            for section in 0..5u64 {
                for i in 0..10u64 {
                    (journal, _, _) = journal
                        .append(section, &(section * 1000 + i))
                        .await
                        .expect("Failed to append");
                }
                journal = journal.sync(section).await.expect("Failed to sync");
            }

            // Verify we have data
            assert_eq!(journal.get(0, 0).await.unwrap(), 0);
            assert_eq!(journal.get(4, 0).await.unwrap(), 4000);

            // Clear the journal
            journal = journal.clear().await.expect("Failed to clear");

            // After clear, all reads should fail
            for section in 0..5u64 {
                assert!(matches!(
                    journal.get(section, 0).await,
                    Err(Error::SectionOutOfRange(s)) if s == section
                ));
            }

            // Append new data after clear
            for i in 0..5u64 {
                (journal, _, _) = journal
                    .append(10, &(i * 100))
                    .await
                    .expect("Failed to append after clear");
            }
            journal = journal.sync(10).await.expect("Failed to sync after clear");

            // New data should be readable
            assert_eq!(journal.get(10, 0).await.unwrap(), 0);

            // Old sections should still be missing
            assert!(matches!(
                journal.get(0, 0).await,
                Err(Error::SectionOutOfRange(0))
            ));

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

    #[test_traced]
    fn test_journal_try_get_sync() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            for (compression, partition) in [(None, "plain"), (Some(3), "compressed")] {
                let cfg = Config {
                    partition: partition.into(),
                    compression,
                    codec_config: (..).into(),
                    page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                    write_buffer: NZUsize!(1024),
                };
                let journal = Journal::<_, Bytes>::init(context.child(partition), cfg)
                    .await
                    .expect("Failed to initialize journal");

                // Uncompressed, the short frame fits the header read and is decoded from it.
                // The long frame is read in full from the write buffer.
                let short = Bytes::from(vec![1u8; 3]);
                let (journal, short_offset, _) =
                    journal.append(1, &short).await.expect("Failed to append");
                let long = Bytes::from(vec![7u8; 32]);
                let (journal, long_offset, _) =
                    journal.append(1, &long).await.expect("Failed to append");
                assert_eq!(journal.try_get_sync(1, short_offset), Some(short.clone()));
                assert_eq!(journal.try_get_sync(1, long_offset), Some(long.clone()));

                // Once synced and read back, the item is served from the page cache.
                let journal = journal.sync(1).await.expect("Failed to sync");
                journal.get(1, long_offset).await.expect("Failed to get");
                assert_eq!(journal.try_get_sync(1, long_offset), Some(long.clone()));

                // An offset at the end of the section is a miss.
                let size = journal.size(1).expect("Failed to size");
                assert!(journal.try_get_sync(1, size).is_none());

                journal.destroy().await.expect("Failed to destroy");
            }
        });
    }
}