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` method. When
//! calling `close`, all pending data is automatically synced and any open blobs are dropped.
//!
//! # 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 to produce a stream of all items in the `Journal` in order of their `section` and
//! `offset`.
//!
//! # Compression
//!
//! `Journal` supports optional compression using `zstd`. This can be enabled by setting the
//! `compression` field in the `Config` struct to a valid `zstd` compression level. This setting can
//! be changed between initializations of `Journal`, however, it must remain populated if any data
//! was written with compression 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::from_pooler(&context, NZU16!(1024), NZUsize!(10));
//!
//!     // Create a journal
//!     let mut 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
//!     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::{
    frame::{
        decode_item, decode_length_prefix, encode_frame_into, find_frame, read_frame_at, FrameInfo,
    },
    Error,
};
use commonware_codec::{varint::MAX_U32_VARINT_SIZE, Codec, CodecShared};
use commonware_runtime::{
    buffer::paged::{CacheRef, Replay, Writer},
    Blob, Buf, Handle, IoBuf, Metrics, Storage,
};
use futures::stream::{self, Stream, StreamExt};
use std::{io::Cursor, 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 ReplayState<'a, B: Blob, C> {
    section: u64,
    // Need a [Writer] because replay may repair the blob by rewinding invalid trailing data.
    blob: &'a mut Writer<B>,
    replay: Replay<B>,
    skip_bytes: u64,
    offset: u64,
    valid_offset: u64,
    codec_config: C,
    compressed: bool,
    done: bool,
}

/// 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.
pub struct Journal<E: Storage + Metrics, V: Codec> {
    manager: Manager<E, AppendFactory>,

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

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

impl<E: Storage + Metrics, V: CodecShared> Journal<E, V> {
    /// 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> {
        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?;

        Ok(Self {
            manager,
            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: &Writer<E::Blob>,
        offset: u64,
    ) -> Result<(u64, u32, V), Error> {
        read_frame_at(blob, offset, cfg, compressed).await
    }

    /// Returns an ordered stream of all items in the journal starting with the item at the given
    /// `start_section` and `offset` into that section. Each item is returned as a tuple of
    /// (section, offset, size, item).
    pub async fn replay(
        &mut self,
        start_section: u64,
        mut start_offset: u64,
        buffer: NonZeroUsize,
    ) -> Result<impl Stream<Item = Result<(u64, u64, u32, V), Error>> + Send + '_, Error> {
        // Collect all blobs to replay. This validates replay setup but does not allocate
        // `buffer` bytes per blob; page buffers are allocated later by `Replay::ensure`.
        let codec_config = self.codec_config.clone();
        let compressed = self.compression.is_some();
        let mut blobs = Vec::new();
        for (&section, blob) in self.manager.sections_from(start_section) {
            if section == start_section && start_offset > blob.size() {
                return Err(Error::ItemOutOfRange(start_offset));
            }
            let replay = blob.replay(buffer).await?;
            blobs.push((section, blob, replay, codec_config.clone(), compressed));
        }

        // Stream items as they are read to avoid occupying too much memory
        Ok(stream::iter(blobs).flat_map(
            move |(section, blob, replay, codec_config, compressed)| {
                // Calculate initial skip bytes for first blob
                let skip_bytes = if section == start_section {
                    start_offset
                } else {
                    start_offset = 0;
                    0
                };

                stream::unfold(
                    ReplayState {
                        section,
                        blob,
                        replay,
                        skip_bytes,
                        offset: 0,
                        valid_offset: skip_bytes,
                        codec_config,
                        compressed,
                        done: false,
                    },
                    move |mut state| async move {
                        if state.done {
                            return None;
                        }

                        let blob_size = state.replay.blob_size();
                        let mut batch: Vec<Result<(u64, u64, u32, V), Error>> = Vec::new();
                        loop {
                            // 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 state.replay.ensure(MAX_U32_VARINT_SIZE).await {
                                Ok(true) => {}
                                Ok(false) => {
                                    // Reader exhausted - check if buffer is empty
                                    if state.replay.remaining() == 0 {
                                        state.done = true;
                                        return if batch.is_empty() {
                                            None
                                        } else {
                                            Some((batch, state))
                                        };
                                    }
                                    // Buffer still has data - continue to try decoding
                                }
                                Err(err) => {
                                    batch.push(Err(err.into()));
                                    state.done = true;
                                    return Some((batch, state));
                                }
                            }

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

                            // Try to decode length prefix
                            let before_remaining = state.replay.remaining();
                            let (item_size, varint_len) =
                                match decode_length_prefix(&mut state.replay) {
                                    Ok(result) => result,
                                    Err(err) => {
                                        // Could be incomplete varint - check if reader exhausted
                                        if state.replay.is_exhausted()
                                            || before_remaining < MAX_U32_VARINT_SIZE
                                        {
                                            // Treat as trailing bytes
                                            if state.valid_offset < blob_size
                                                && state.offset < blob_size
                                            {
                                                warn!(
                                                    blob = state.section,
                                                    bad_offset = state.offset,
                                                    new_size = state.valid_offset,
                                                    "trailing bytes detected: truncating"
                                                );
                                                // Tail repair is exceptional; make it durable
                                                // immediately so callers do not need to track
                                                // replay-time repaired sections separately.
                                                if let Err(err) =
                                                    state.blob.resize(state.valid_offset).await
                                                {
                                                    batch.push(Err(err.into()));
                                                    state.done = true;
                                                    return Some((batch, state));
                                                }
                                                if let Err(err) = state.blob.sync().await {
                                                    batch.push(Err(err.into()));
                                                    state.done = true;
                                                    return Some((batch, state));
                                                }
                                            }
                                            state.done = true;
                                            return if batch.is_empty() {
                                                None
                                            } else {
                                                Some((batch, state))
                                            };
                                        }
                                        batch.push(Err(err));
                                        state.done = true;
                                        return Some((batch, state));
                                    }
                                };

                            // Ensure we have enough data for item body
                            match state.replay.ensure(item_size).await {
                                Ok(true) => {}
                                Ok(false) => {
                                    // Incomplete item at end - truncate
                                    warn!(
                                        blob = state.section,
                                        bad_offset = state.offset,
                                        new_size = state.valid_offset,
                                        "incomplete item at end: truncating"
                                    );
                                    if let Err(err) = state.blob.resize(state.valid_offset).await {
                                        batch.push(Err(err.into()));
                                        state.done = true;
                                        return Some((batch, state));
                                    }
                                    if let Err(err) = state.blob.sync().await {
                                        batch.push(Err(err.into()));
                                        state.done = true;
                                        return Some((batch, state));
                                    }
                                    state.done = true;
                                    return if batch.is_empty() {
                                        None
                                    } else {
                                        Some((batch, state))
                                    };
                                }
                                Err(err) => {
                                    batch.push(Err(err.into()));
                                    state.done = true;
                                    return Some((batch, state));
                                }
                            }

                            // Decode item - use take() to limit bytes read
                            let item_offset = state.offset;
                            let next_offset = match state
                                .offset
                                .checked_add(varint_len as u64)
                                .and_then(|o| o.checked_add(item_size as u64))
                            {
                                Some(o) => o,
                                None => {
                                    batch.push(Err(Error::OffsetOverflow));
                                    state.done = true;
                                    return Some((batch, state));
                                }
                            };
                            match decode_item::<V>(
                                (&mut state.replay).take(item_size),
                                &state.codec_config,
                                state.compressed,
                            ) {
                                Ok(decoded) => {
                                    batch.push(Ok((
                                        state.section,
                                        item_offset,
                                        item_size as u32,
                                        decoded,
                                    )));
                                    state.valid_offset = next_offset;
                                    state.offset = next_offset;
                                }
                                Err(err) => {
                                    batch.push(Err(err));
                                    state.done = true;
                                    return Some((batch, state));
                                }
                            }

                            // Return batch if we have items and buffer is low
                            if !batch.is_empty() && state.replay.remaining() < MAX_U32_VARINT_SIZE {
                                return Some((batch, state));
                            }
                        }
                    },
                )
                .flat_map(stream::iter)
            },
        ))
    }

    /// Encode an item.
    ///
    /// Returns `(buf, item_len)` where `item_len` is the length of the encoded (and
    /// possibly compressed) payload, excluding the size prefix.
    fn encode_item(compression: Option<u8>, item: &V) -> Result<(Vec<u8>, u32), Error> {
        let mut buf = Vec::new();
        let item_len = encode_frame_into(compression, item, &mut buf)?;
        Ok((buf, item_len))
    }

    /// 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).
    pub async fn append(&mut self, section: u64, item: &V) -> Result<(u64, u32), Error> {
        let (buf, item_len) = Self::encode_item(self.compression, item)?;
        self.append_raw(section, IoBuf::from(buf))
            .await
            .map(|offset| (offset, item_len))
    }

    /// Append pre-encoded bytes to the given section, returning the byte offset
    /// where the data was written.
    ///
    /// The buffer must be in the on-disk format produced by [Self::encode_item].
    async fn append_raw(&mut self, section: u64, buf: IoBuf) -> Result<u64, Error> {
        let blob = self.manager.get_or_create(section).await?;
        let offset = blob.append_owned(buf).await?;
        trace!(blob = section, offset, "appended item");
        Ok(offset)
    }

    /// 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> {
        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)
    }

    /// 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> {
        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)
    }

    /// 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> {
        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 = Cursor::new(&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>(
                &header[varint_len..varint_len + data_len],
                &self.codec_config,
                compressed,
            )
            .ok();
        }

        // Otherwise try reading the full item from cache.
        let mut buf = vec![0u8; item_len];
        if !blob.try_read_sync_into(&mut buf, offset) {
            return None;
        }
        decode_item::<V>(
            &buf[varint_len..varint_len + data_len],
            &self.codec_config,
            compressed,
        )
        .ok()
    }

    /// 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.manager.size(section)
    }

    /// Rewinds the journal to the given `section` and `size`.
    ///
    /// This removes any data beyond the specified `section` and `size`.
    ///
    /// # Warnings
    ///
    /// * This operation is not guaranteed to survive restarts until sync is called.
    /// * This operation is not atomic, but it will always leave the journal in a consistent state
    ///   in the event of failure since blobs are always removed in reverse order of section.
    pub async fn rewind(&mut self, section: u64, size: u64) -> Result<(), Error> {
        self.manager.rewind(section, size).await
    }

    /// Rewinds the `section` to the given `size`.
    ///
    /// Unlike [Self::rewind], this method does not modify anything other than the given `section`.
    ///
    /// # Warning
    ///
    /// This operation is not guaranteed to survive restarts until sync is called.
    pub async fn rewind_section(&mut self, section: u64, size: u64) -> Result<(), Error> {
        self.manager.rewind_section(section, size).await
    }

    /// Ensures the given `sections` are synced to the underlying store.
    ///
    /// If a selected section does not exist, no error will be returned.
    pub async fn sync(&mut self, sections: impl crate::Sections) -> Result<(), Error> {
        self.manager.sync(sections).await
    }

    /// Start syncing the given `sections` to storage.
    pub async fn start_sync(
        &mut self,
        sections: impl crate::Sections,
    ) -> Result<Handle<()>, Error> {
        self.manager.start_sync(sections).await
    }

    /// Syncs all open sections.
    pub async fn sync_all(&mut self) -> Result<(), Error> {
        self.manager.sync_all().await
    }

    /// Prunes all `sections` less than `min`. Returns true if any sections were pruned.
    pub async fn prune(&mut self, min: u64) -> Result<bool, Error> {
        self.manager.prune(min).await
    }

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

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

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

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

    /// Removes any underlying blobs created by the journal.
    pub async fn destroy(self) -> Result<(), Error> {
        self.manager.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<(), Error> {
        self.manager.clear().await
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use commonware_codec::{varint::UInt, EncodeSize, Write as _};
    use commonware_macros::test_traced;
    use commonware_runtime::{deterministic, Blob, BufMut, Runner, Storage, Supervisor as _};
    use commonware_utils::{NZUsize, NZU16};
    use futures::{pin_mut, StreamExt};
    use std::num::NonZeroU16;

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

    #[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::from_pooler(&context, 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
                .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.sync(index).await.expect("Failed to sync journal");
            drop(journal);
            let mut 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 stream = journal
                .replay(0, 0, NZUsize!(1024))
                .await
                .expect("unable to setup replay");
            pin_mut!(stream);
            while let Some(result) = stream.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::from_pooler(&context, 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
                    .append(*index, data)
                    .await
                    .expect("Failed to append data");
                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 stream = journal
                    .replay(0, 0, NZUsize!(1024))
                    .await
                    .expect("unable to setup replay");
                pin_mut!(stream);
                while let Some(result) = stream.next().await {
                    match result {
                        Ok((blob_index, _, _, item)) => items.push((blob_index, item)),
                        Err(err) => panic!("Failed to read item: {err}"),
                    }
                }
            }

            // 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::from_pooler(&context, 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
                    .append(index, &index)
                    .await
                    .expect("Failed to append data");
                journal.sync(index).await.expect("Failed to sync blob");
            }

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

            // Prune blobs with indices less than 3
            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.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 stream = journal
                    .replay(0, 0, NZUsize!(1024))
                    .await
                    .expect("unable to setup replay");
                pin_mut!(stream);
                while let Some(result) = stream.next().await {
                    match result {
                        Ok((blob_index, _, _, item)) => items.push((blob_index, item)),
                        Err(err) => panic!("Failed to read item: {err}"),
                    }
                }
            }

            // 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.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::from_pooler(&context, 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
                    .append(section, &(section as i32))
                    .await
                    .expect("Failed to append data");
                journal.sync(section).await.expect("Failed to sync");
            }

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

            // Test that accessing pruned sections returns the correct error

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

            match journal.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 rewind on pruned section
            match journal.rewind(2, 0).await {
                Err(Error::AlreadyPrunedToSection(3)) => {}
                other => panic!("Expected AlreadyPrunedToSection(3), got {other:?}"),
            }

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

            // Test sync on pruned section
            match journal.sync(2).await {
                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());
            assert!(journal.sync(4).await.is_ok());

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

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

            // Verify sections 3 and 4 are now pruned
            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_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::from_pooler(&context, 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
                        .append(section, &(section as i32))
                        .await
                        .expect("Failed to append data");
                    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");

                // 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::from_pooler(&context, 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
            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::from_pooler(&context, 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_sync(0, incomplete_data)
                .await
                .expect("Failed to write incomplete data");

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

            // Attempt to replay the journal
            let stream = journal
                .replay(0, 0, NZUsize!(1024))
                .await
                .expect("unable to setup replay");
            pin_mut!(stream);
            let mut items = Vec::<(u64, u64)>::new();
            while let Some(result) = stream.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_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::from_pooler(&context, 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
                .append(section, &item)
                .await
                .expect("Failed to append item");
            journal
                .append_raw(section, IoBuf::copy_from_slice(&[0xFF, 0xFF]))
                .await
                .expect("Failed to append trailing bytes");
            journal.sync(section).await.expect("Failed to sync journal");
            drop(journal);

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

            let stream = journal
                .replay(0, 0, NZUsize!(1024))
                .await
                .expect("unable to setup replay");
            pin_mut!(stream);

            let first = stream
                .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 stream.next().await {
                Some(Err(_)) => {}
                other => {
                    panic!("expected resize error while repairing trailing bytes, got {other:?}")
                }
            }
            assert!(stream.next().await.is_none());
        });
    }

    #[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::from_pooler(&context, 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_sync(0, buf)
                .await
                .expect("Failed to write incomplete item");

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

            // Attempt to replay the journal
            let stream = journal
                .replay(0, 0, NZUsize!(1024))
                .await
                .expect("unable to setup replay");
            pin_mut!(stream);
            let mut items = Vec::<(u64, u64)>::new();
            while let Some(result) = stream.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::from_pooler(&context, 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_sync(0, buf)
                .await
                .expect("Failed to write item without checksum");

            // Initialize the journal
            let mut 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 stream = journal
                .replay(0, 0, NZUsize!(1024))
                .await
                .expect("unable to setup replay");
            pin_mut!(stream);
            let mut items = Vec::<(u64, u64)>::new();
            while let Some(result) = stream.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::from_pooler(&context, 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_sync(0, buf)
                .await
                .expect("Failed to write item with bad checksum");

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

            // Attempt to replay the journal
            {
                let stream = journal
                    .replay(0, 0, NZUsize!(1024))
                    .await
                    .expect("unable to setup replay");
                pin_mut!(stream);
                let mut items = Vec::<(u64, u64)>::new();
                while let Some(result) = stream.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());
            }
            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_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::from_pooler(&context, 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.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
                    .append(*index, data)
                    .await
                    .expect("Failed to append data");
                journal.sync(*index).await.expect("Failed to sync blob");
            }

            // Sync all sections and drop the 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
            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 stream = journal
                    .replay(0, 0, NZUsize!(1024))
                    .await
                    .expect("unable to setup replay");
                pin_mut!(stream);
                while let Some(result) = stream.next().await {
                    match result {
                        Ok((blob_index, _, _, item)) => items.push((blob_index, item)),
                        Err(err) => panic!("Failed to read item: {err}"),
                    }
                }
            }
            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 stream = journal
                    .replay(0, 0, NZUsize!(1024))
                    .await
                    .expect("unable to setup replay");
                pin_mut!(stream);
                while let Some(result) = stream.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(), 1);
            assert_eq!(items[0].0, 1);
            assert_eq!(items[0].1, 1);

            // Append a new item to truncated partition
            let (_offset, _) = journal.append(2, &5).await.expect("Failed to append data");
            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 mut 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 stream = journal
                    .replay(0, 0, NZUsize!(1024))
                    .await
                    .expect("unable to setup replay");
                pin_mut!(stream);
                while let Some(result) = stream.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::from_pooler(&context, 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.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
                    .append(*index, data)
                    .await
                    .expect("Failed to append data");
                journal.sync(*index).await.expect("Failed to sync blob");
            }

            // Sync all sections and drop the 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_sync(blob_size, vec![0u8; 16])
                .await
                .expect("Failed to add extra data");

            // Re-initialize the journal to simulate a restart
            let mut 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 stream = journal
                .replay(0, 0, NZUsize!(1024))
                .await
                .expect("unable to setup replay");
            pin_mut!(stream);
            while let Some(result) = stream.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_rewind() {
        // 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::from_pooler(&context, 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.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.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.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.rewind(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_rewind_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::from_pooler(&context, 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.append(u64::MAX, &42i32).await.unwrap();
            let size = journal.size(u64::MAX).unwrap();
            assert!(size > 0);

            // Rewinding the maximal section removes no sections above it and must not panic.
            journal.rewind(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_rewind_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::from_pooler(&context, 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.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.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.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.rewind_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::from_pooler(&context, 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, 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.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 mut journal = Journal::<_, u8>::init(context.child("second"), cfg)
                .await
                .expect("Failed to re-initialize journal");

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

            let mut count = 0;
            while let Some(result) = stream.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_rewind_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::from_pooler(&context, 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.append(section, &(section as i32)).await.unwrap();
            }
            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");
            }

            // Rewind to section 5 (should remove sections 6-10)
            journal.rewind(5, journal.size(5).unwrap()).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 stream = journal.replay(0, 0, NZUsize!(1024)).await.unwrap();
                pin_mut!(stream);
                let mut items = Vec::new();
                while let Some(result) = stream.next().await {
                    let (section, _, _, item) = result.unwrap();
                    items.push((section, item));
                }
                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_rewind_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::from_pooler(&context, 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.append(1, &i).await.unwrap();
                journal.sync(1).await.unwrap();
                sizes.push(journal.size(1).unwrap());
            }

            // Rewind to keep only first 3 items
            let target_size = sizes[2];
            journal.rewind(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 stream = journal.replay(0, 0, NZUsize!(1024)).await.unwrap();
                pin_mut!(stream);
                let mut items = Vec::new();
                while let Some(result) = stream.next().await {
                    let (_, _, _, item) = result.unwrap();
                    items.push(item);
                }
                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_rewind_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::from_pooler(&context, 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.append(section, &(section as i32)).await.unwrap();
            }
            journal.sync_all().await.unwrap();

            // Rewind to section 3 (doesn't exist)
            journal.rewind(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 stream = journal.replay(0, 0, NZUsize!(1024)).await.unwrap();
                pin_mut!(stream);
                let items: Vec<_> = stream.collect().await;
                assert!(items.is_empty());
            }

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

    #[test_traced]
    fn test_journal_rewind_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::from_pooler(&context, 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.append(section, &(section as i32)).await.unwrap();
            }
            journal.sync_all().await.unwrap();

            // Rewind to section 2
            let size = journal.size(2).unwrap();
            journal.rewind(2, size).await.unwrap();
            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 stream = journal.replay(0, 0, NZUsize!(1024)).await.unwrap();
                pin_mut!(stream);
                let mut items = Vec::new();
                while let Some(result) = stream.next().await {
                    let (section, _, _, item) = result.unwrap();
                    items.push((section, item));
                }
                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_rewind_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::from_pooler(&context, 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.append(section, &(section as i32)).await.unwrap();
            }
            journal.sync_all().await.unwrap();

            // Rewind section 1 to size 0
            journal.rewind(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 stream = journal.replay(0, 0, NZUsize!(1024)).await.unwrap();
                pin_mut!(stream);
                let items: Vec<_> = stream.collect().await;
                assert!(items.is_empty());
            }

            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::from_pooler(&context, 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.append(1, &i).await.unwrap();
            }
            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_sync(physical_size_before, vec![0xFF, 0xFF])
                .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;
            {
                let mut journal = Journal::<_, i32>::init(context.child("second"), cfg.clone())
                    .await
                    .unwrap();

                let stream = journal
                    .replay(1, start_offset, NZUsize!(1024))
                    .await
                    .unwrap();
                pin_mut!(stream);

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

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

            // 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_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::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1024),
            };
            let mut journal = Journal::init(context.child("storage"), cfg).await.unwrap();
            journal.append(1, &7i32).await.unwrap();

            let result = journal.replay(1, u64::MAX, NZUsize!(1024)).await;
            assert!(matches!(result, Err(Error::ItemOutOfRange(u64::MAX))));
            drop(result);

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

    #[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::from_pooler(&context, 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, size) = journal
                .append(1, &large_data)
                .await
                .expect("Failed to append large item");
            assert_eq!(size as usize, LARGE_SIZE);
            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 stream = journal
                    .replay(0, 0, NZUsize!(1024))
                    .await
                    .expect("Failed to setup replay");
                pin_mut!(stream);

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

                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::from_pooler(&context, 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
                .append(1, &first)
                .await
                .expect("Failed to append first item");
            let (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.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 stream = journal
                    .replay(0, 0, NZUsize!(1024))
                    .await
                    .expect("Failed to setup replay");
                pin_mut!(stream);

                let mut items = Vec::new();
                while let Some(result) = stream.next().await {
                    let (section, off, _, item) = result.expect("Failed to replay item");
                    items.push((section, off, item));
                }
                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::from_pooler(&context, 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
                    .append(*section, data)
                    .await
                    .expect("Failed to append");
                offsets.push(offset);
            }
            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 stream = journal
                    .replay(0, 0, NZUsize!(1024))
                    .await
                    .expect("Failed to setup replay");
                pin_mut!(stream);

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

                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 stream = journal
                    .replay(5, 0, NZUsize!(1024))
                    .await
                    .expect("Failed to setup replay from section 5");
                pin_mut!(stream);

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

                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 stream = journal
                    .replay(3, 0, NZUsize!(1024))
                    .await
                    .expect("Failed to setup replay from section 3");
                pin_mut!(stream);

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

                // 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::from_pooler(&context, 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.append(1, &100i32).await.expect("Failed to append");

            // Create section 2 but don't append anything - just sync to create the blob
            // Actually, we need to append something and then rewind to make it empty
            journal.append(2, &200i32).await.expect("Failed to append");
            journal.sync(2).await.expect("Failed to sync");
            journal
                .rewind_section(2, 0)
                .await
                .expect("Failed to rewind");

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

            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 stream = journal
                    .replay(0, 0, NZUsize!(1024))
                    .await
                    .expect("Failed to setup replay");
                pin_mut!(stream);

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

                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 stream = journal
                    .replay(2, 0, NZUsize!(1024))
                    .await
                    .expect("Failed to setup replay from section 2");
                pin_mut!(stream);

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

                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::from_pooler(&context, 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, size) = journal
                .append(1, &exact_data)
                .await
                .expect("Failed to append exact item");
            assert_eq!(size as usize, ITEM_SIZE);
            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 stream = journal
                    .replay(0, 0, NZUsize!(1024))
                    .await
                    .expect("Failed to setup replay");
                pin_mut!(stream);

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

                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_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::from_pooler(&context, 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.append(1, &item1).await.expect("Failed to append");
            let (offset2, _) = journal.append(1, &item2).await.expect("Failed to append");
            let (offset3, _) = journal.append(1, &item3).await.expect("Failed to append");

            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 stream = journal
                    .replay(0, 0, NZUsize!(64))
                    .await
                    .expect("Failed to setup replay");
                pin_mut!(stream);

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

                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::from_pooler(&context, 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
                        .append(section, &(section * 1000 + i))
                        .await
                        .expect("Failed to append");
                }
                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.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
                    .append(10, &(i * 100))
                    .await
                    .expect("Failed to append after clear");
            }
            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();
        });
    }
}