//! An append-only log for storing fixed length _items_ on disk.
//!
//! In addition to replay, stored items can be fetched directly by their `position` in the journal,
//! where position is defined as the item's order of insertion starting from 0, unaffected by
//! pruning.
//!
//! _See [super::variable] for a journal that supports variable length items._
//!
//! # Format
//!
//! Data stored in a `fixed::Journal` is persisted in one of many Blobs. Each `Blob` contains a
//! configurable maximum of `items_per_blob`, with page-level data integrity.
//!
//! ```text
//! +--------+--------+-----+----------+
//! | item_0 | item_1 | ... | item_n-1 |
//! +--------+--------+-----+----------+
//!
//! n = config.items_per_blob
//! ```
//!
//! The most recent blob may not necessarily be full, in which case it will contain fewer than the
//! maximum number of items.
//!
//! Data fetched from disk is always checked for integrity before being returned. If the data is
//! found to be invalid, an error is returned instead.
//!
//! # Architecture
//!
//! Three types divide the work:
//!
//! - [`Journal`] tracks which positions are readable (`bounds`) and maps each position to a blob
//!   and byte offset.
//!
//! - `Writable` owns the files: the contiguous sealed blobs plus the one writable tail. When the
//!   tail fills, it is sealed and an fsync of it begins. `Writable` tracks that in-flight sync,
//!   and its tail writer tracks any started sync of the new tail.
//!
//! - `Checkpoint` owns the durable recovery hints (mid-blob pruning boundary, recovery
//!   watermark, staged clear target) consulted before trusting blob state on startup.
//!
//! # Open Blobs
//!
//! Every retained blob is held open; a pruned blob stays open until the last snapshot holding it
//! drops. Use a larger `items_per_blob` or prune to bound the count.
//!
//! # Partition
//!
//! Blobs are stored in the legacy partition (`cfg.partition`) if it already contains data;
//! otherwise they are stored in `{cfg.partition}-blobs`.
//!
//! The checkpoint (the durable recovery record: pruning boundary, recovery watermark, and any
//! in-progress clear intent) is stored in `{cfg.partition}-metadata`.
//!
//! # Recovery
//!
//! Blobs are filled sequentially. Recovery walks the blob range from oldest to newest and
//! compares each blob's item count to its logical capacity:
//!
//! - A short or missing non-newest blob indicates a gap in durable data; recovery stops there
//!   and truncates newer blobs.
//! - The newest blob may be short, since it is the normal append frontier. Recovery includes
//!   its items.
//!
//! # Durability invariant
//!
//! When an append fills the tail blob, the journal rolls over: it seals the full blob, starts an
//! fsync of it, and opens the next blob as the new tail. Each rollover first waits for the
//! previous rollover's fsync, so only the tail and its predecessor (the blob sealed by the last
//! rollover) can ever hold non-durable data. Every older blob is fully durable.
//!
//! A crash while one of those fsyncs is in flight can persist the blob's pages out of order, so
//! an earlier page may be lost while later pages, including a valid last page, survive. Sizing a
//! blob by its last valid page cannot detect such a hole, so recovery re-reads the two newest
//! blobs from the acknowledged prefix onward, which a bounded open lowers to its cap, and
//! truncates each at the first missing or corrupt page. Pages beneath that start had their
//! covering fsync complete, so recovery never re-reads them. Damage there is external corruption
//! that surfaces as a read error, while recovery still fails loudly when a blob no longer
//! physically backs its acknowledged items.
//!
//! The recovered size is the logical end of this contiguous prefix. If the persisted watermark
//! exceeds the recovered size, an unbounded open returns a corruption error. A bounded open
//! (`init_at_most`) compares the watermark clamped to its cap, since blobs at or above the cap
//! are discarded, and publication persists the retained end as the watermark. Both the pruning
//! boundary and watermark are persisted before `init` returns.
//!
//! The recovery watermark is therefore an external recovery checkpoint, not a complete record of
//! every item that may have become durable through `commit` or storage behavior.
//!
//! # Watermark advancement
//!
//! The watermark must never exceed what is durably on disk. `sync()` completes its data sync
//! before writing the watermark, so it can advance it to the current size. `start_sync()`
//! returns before its data sync completes, so it advances the watermark using an older proof:
//! the journal's *barrier*, the highest size whose sync it has already observed complete.
//! A durable value is safe to write at any time, so the watermark write needs no ordering
//! against the in-flight data sync.
//!
//! An advance that fails to start fails the call, destroying the journal. A started advance's
//! failure surfaces on the returned handle, and the next checkpoint write observes it and
//! fails before writing.
//!
//! The invariants:
//!
//! - The watermark only takes values the barrier has held (never an in-flight size).
//! - The barrier advances only on an observed sync success.
//! - Initialization recovery persists a lower watermark before discarding a suffix. Live
//!   handles never move their retained end backward.
//!
//! # Consistency
//!
//! 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 durably written using `commit` or `sync`.
//!
//! # Pruning
//!
//! The `prune` method allows the `Journal` to prune blobs consisting entirely of items prior to a
//! given point in history.
//!
//! # Clearing / reset
//!
//! Clearing wipes all data and restarts the journal at a requested size. An open journal can only
//! clear to a size at or beyond its current end.
//!
//! To stay crash-safe, a clear records its target size in the checkpoint *before* deleting any
//! blob. If a crash interrupts the deletion, reopening sees that recorded target and finishes the
//! clear, rather than mistaking the half-deleted blobs for corruption.
//!
//! Callers reach this through `clear_to_size` (clear an open journal) or `init_at_size` (open
//! straight into a cleared, empty journal at a given size).
//!
//! # Replay
//!
//! The `replay` method supports fast reading of all unpruned items into memory.

use super::{
    Batch, Decoded,
    blobs::{Blob, Blobs, Partition, Replay as BlobReplay, Writable},
    checkpoint::Checkpoint,
};
#[commonware_macros::stability(ALPHA)]
use crate::journal::authenticated;
use crate::{
    Context, SyncCompletion,
    journal::{
        Error,
        contiguous::{Many, Mutable, metrics::Metrics},
        durability::Barrier,
    },
};
use bytes::Bytes;
use commonware_codec::{CodecFixedShared, Copying, DecodeExt as _, ReadExt as _};
use commonware_runtime::{
    Blob as RBlob, Buf, Handle, IoBuf, ReadOptions,
    buffer::paged::{CacheRef, Recovery as PagedRecovery},
};
use commonware_utils::{Cached, futures::try_join_all};
use futures::{FutureExt as _, Stream};
use std::{
    collections::{BTreeMap, btree_map::Entry},
    future::Future,
    marker::PhantomData,
    num::{NonZeroU64, NonZeroUsize},
    ops::Range,
    sync::Arc,
};
use tracing::warn;

// Reusable scratch for the synchronous read paths ([`Reader::try_read_sync`] and
// [`Reader::probe_items`]), grown to the largest probe served on the thread. Both run on the
// hot read path, where a fresh zeroed allocation per call contends under the pool's fan-out.
commonware_utils::thread_local_cache!(static PROBE_SCRATCH: Vec<u8>);

/// Items encoded for a deferred append, created by [`Journal::prepare_append`] and consumed by
/// [`Journal::append_prepared`].
pub struct PreparedAppend<A> {
    buf: Vec<u8>,
    _marker: PhantomData<A>,
}

/// Return the first retained logical position in `blob`.
#[inline]
fn first_in_blob(pruning_boundary: u64, blob: u64, items_per_blob: u64) -> Result<u64, Error> {
    let start = super::blob_first_position(blob, items_per_blob)?;
    Ok(pruning_boundary.max(start))
}

/// Build a replay batch stream over the retained blob range.
///
/// The stream is split into one state per blob so replay can start at a mid-blob pruning boundary,
/// stop at the journal's logical end, and avoid reading across blob files. `buffer` is a byte
/// budget for each blob replay, not an item count.
fn replay_stream<'a, B: RBlob, A: CodecFixedShared>(
    blobs: &Blobs<'a, B>,
    bounds: Range<u64>,
    items_per_blob: NonZeroU64,
    range: Range<u64>,
    buffer: NonZeroUsize,
    read_options: ReadOptions,
) -> Result<impl Stream<Item = Result<Batch<A>, Error>> + Send + use<'a, B, A>, Error> {
    if range.start > range.end || range.end > bounds.end {
        return Err(Error::ItemOutOfRange(if range.start > range.end {
            range.start
        } else {
            range.end
        }));
    }
    if range.start < bounds.start {
        return Err(Error::ItemPruned(range.start));
    }

    let mut states = Vec::new();
    if range.start < range.end {
        let items_per_blob = items_per_blob.get();
        let start_pos = range.start;
        let start_blob = super::position_to_blob(start_pos, items_per_blob);
        let end_blob = super::position_to_blob(range.end - 1, items_per_blob);
        let items_per_batch = (buffer.get() / A::SIZE).max(1);

        for blob in start_blob..=end_blob {
            // The oldest retained blob may begin after its natural blob boundary when pruning
            // kept only a suffix.
            let blob_first = first_in_blob(bounds.start, blob, items_per_blob)?;
            let first_pos = if blob == start_blob {
                start_pos
            } else {
                blob_first
            };
            let blob_end = super::blob_end_position(blob, items_per_blob, range.end);
            let offset = (first_pos - blob_first)
                .checked_mul(A::SIZE as u64)
                .ok_or(Error::OffsetOverflow)?;
            let blob = blobs
                .get(blob)
                .expect("positions in bounds map to a retained blob");

            states.push(FixedReplayState::<B, A> {
                replay: blob.replay_from(offset, buffer, read_options)?,
                pos: first_pos,
                end_pos: blob_end,
                items_per_batch,
                _marker: PhantomData,
            });
        }
    }

    Ok(super::replay_stream_from_states(states))
}

/// Replay state for one fixed-size blob.
struct FixedReplayState<'a, B: RBlob, A> {
    /// Sequential logical bytes for this blob.
    replay: BlobReplay<'a, B>,
    /// Next position to yield.
    pos: u64,
    /// Exclusive end position within this blob.
    end_pos: u64,
    /// Maximum number of items decoded per batch.
    items_per_batch: usize,
    _marker: PhantomData<A>,
}

impl<B: RBlob, A: CodecFixedShared> super::BatchDecoder for FixedReplayState<'_, B, A> {
    type Item = A;

    /// Decode the next batch of fixed-size items from this blob.
    async fn next(mut self) -> Option<(Decoded<A>, Self)> {
        if self.pos == self.end_pos {
            return None;
        }

        // Require at least one whole item so a short blob is reported as corruption at the
        // current position. Additional already-buffered items are decoded below.
        let start = self.pos;
        match self.replay.ensure(A::SIZE).await {
            Ok(true) => {}
            Ok(false) => {
                let error = Error::Corruption(format!("blob ended before position {start}"));
                let decoded = Decoded::failed(start, Vec::new(), error);
                return Some((decoded, self));
            }
            Err(err) => {
                let decoded = Decoded::failed(start, Vec::new(), err);
                return Some((decoded, self));
            }
        }

        // Decode only whole items that are already buffered, capped by the replay byte budget and
        // this blob's logical end.
        let available = (self.replay.remaining() / A::SIZE) as u64;
        let remaining = self.end_pos - self.pos;
        let count = available.min(self.items_per_batch as u64).min(remaining) as usize;

        // Decode the whole items of a buffered page from one view of it, which costs one page
        // lookup per page instead of one per item and shares the page's allocation with retained
        // fields. An item that straddles a page boundary, or a read boundary while replaying a
        // live view, goes through the general path. Stop this blob on error because decoding may
        // have consumed only part of a record.
        let mut items = Vec::with_capacity(count);
        while items.len() < count {
            let n = (self.replay.chunk().len() / A::SIZE).min(count - items.len());
            if n > 0 {
                let mut page = self.replay.copy_to_bytes(n * A::SIZE);
                for _ in 0..n {
                    match A::read(&mut page) {
                        Ok(item) => items.push(item),
                        Err(err) => {
                            let decoded = Decoded::failed(start, items, Error::Codec(err));
                            return Some((decoded, self));
                        }
                    }
                }
            } else {
                match self.replay.read::<A>() {
                    Ok(item) => items.push(item),
                    Err(err) => {
                        let decoded = Decoded::failed(start, items, Error::Codec(err));
                        return Some((decoded, self));
                    }
                }
            }
        }
        self.pos += count as u64;
        Some((
            Decoded {
                batch: Batch { start, items },
                error: None,
            },
            self,
        ))
    }
}

/// How a blob's on-disk item count compares to its logical capacity.
enum BlobFill {
    Full { len: u64 },
    Short { len: u64 },
    Overfull { len: u64, capacity: u64 },
}

/// The recovered journal size and durability floor derived from the reconciled pruning boundary
/// and on-disk blob lengths.
struct RecoveredBounds {
    /// Size: one past the last recovered item.
    size: u64,
    /// Recovery watermark to persist (a floor on durable size).
    recovery_watermark: u64,
}

/// Configuration for `Journal` storage.
#[derive(Clone)]
pub struct Config {
    /// Prefix for the journal partitions.
    ///
    /// Blobs are stored in `partition` (legacy) if it contains data, otherwise in
    /// `{partition}-blobs`. Metadata is stored in `{partition}-metadata`.
    pub partition: String,

    /// The maximum number of journal items to store in each blob.
    ///
    /// Retained non-tail blobs are expected to be full relative to their logical capacity. A
    /// mid-blob oldest blob may physically hold fewer than this many items, and the newest blob
    /// may contain fewer items.
    pub items_per_blob: NonZeroU64,

    /// 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,

    /// Buffer size for sequential reads during recovery.
    pub replay_buffer: NonZeroUsize,
}

/// The journal's state, boxed so the public [Journal] handle stays pointer-sized.
pub(super) struct Inner<E: Context, A> {
    /// The blobs that comprise the journal.
    blobs: Writable<E>,

    /// The durable recovery checkpoint.
    checkpoint: Checkpoint<E>,

    /// The readable positions; `bounds.end` is the next append position.
    bounds: Range<u64>,

    /// The maximum number of items per blob.
    items_per_blob: NonZeroU64,

    /// Shared with [Reader]s.
    metrics: Arc<Metrics<E>>,

    /// The known-durable size of the journal.
    barrier: Barrier,

    _phantom: PhantomData<A>,
}

/// Storage under recovery, before append or snapshot access is available.
pub struct Recovery<E: Context, A> {
    /// Runtime context for partition resets and the published journal's metrics.
    context: E,

    /// Blob layout and I/O settings shared by recovery and the published journal.
    cfg: Config,

    /// Recovery checkpoint, persisted before discarding any acknowledged suffix.
    checkpoint: Checkpoint<E>,

    /// Active blob partition used to open and remove blobs until publication.
    partition: Partition<E>,

    /// Opened recovery blobs keyed by physical index, including suffix candidates awaiting repair.
    pending: BTreeMap<u64, PagedRecovery<E::Blob>>,

    /// Blob indices excluded by the initial cap; repair removes them unless needed as an empty tail.
    discarded: Vec<u64>,

    /// Selected logical item range; its end is exclusive.
    bounds: Range<u64>,

    /// Durable lower bound on the logical item end, lowered before removing an acknowledged suffix.
    watermark: u64,

    /// Whether an explicit opening cap requires publication to persist the selected end.
    bounded: bool,

    _marker: PhantomData<A>,
}

impl<E: Context, A: CodecFixedShared> Recovery<E, A> {
    /// Recover available items without sealing the retained blobs.
    async fn open(
        context: E,
        cfg: Config,
        checkpoint: Checkpoint<E>,
        max_size: Option<u64>,
    ) -> Result<Self, Error> {
        let ceiling = max_size.unwrap_or(u64::MAX);
        if let Some(target) = checkpoint.clear_target() {
            warn!(
                clear_target = target,
                "crash repair: completing interrupted clear"
            );
            let recovery = Self::complete_clear(context, cfg, checkpoint, target).await?;

            // A persisted reset is authoritative even when the open requests another cap. The cap
            // is only checked against the reset size afterwards.
            if ceiling < target {
                return Err(Error::ItemPruned(ceiling));
            }
            return Ok(recovery);
        }
        let items_per_blob = cfg.items_per_blob.get();

        // Select the active partition and reconcile its oldest blob with the checkpoint's retained
        // start before excluding any suffix.
        let (blob_partition, names) = Partition::select(&context, &cfg.partition).await?;
        let partition = Partition::new(
            context.child("blobs"),
            blob_partition,
            cfg.page_cache.clone(),
            cfg.write_buffer,
        );
        let indices = Partition::<E>::indices(names)?;
        let pruning_boundary = Inner::<E, A>::recover_pruning_boundary(
            checkpoint.boundary_hint(),
            indices.first().copied(),
            items_per_blob,
        )?;
        if ceiling < pruning_boundary {
            return Err(Error::ItemPruned(ceiling));
        }

        // Whole discarded blobs need no valid pages or item encodings.
        let mut pending = BTreeMap::new();
        let mut discarded = Vec::new();
        for blob in indices {
            let first = first_in_blob(pruning_boundary, blob, items_per_blob)?;
            if max_size.is_some() && first >= ceiling {
                discarded.push(blob);
            } else {
                pending.insert(blob, partition.open_recovery(blob).await?);
            }
        }

        // Check the two newest blobs for interior holes before any resize. Only they can hold
        // non-durable data, and a crash during an in-flight fsync can lose an interior page while
        // later pages survive. `PagedRecovery::open` sizes a blob by its last valid page, so it
        // cannot see such a hole. An item-aligned resize can land within a page, which
        // `Recovery::truncate` must read and validate before rewriting its partial tip. The scan
        // starts at the acknowledged in-blob prefix, which a bounded open lowers to its cap. Pages
        // below that start are covered by a completed fsync, so in-model holes are impossible
        // there and any later damage surfaces lazily at read. Above it, first move the target
        // below any hole and round it down to whole items so `recover_bounds` sees only intact
        // data.
        let floor = checkpoint.watermark().unwrap_or(0).min(ceiling);
        let floor_blob = super::position_to_blob(floor, items_per_blob);
        let suspects: Vec<u64> = pending.keys().rev().take(2).copied().collect();
        for blob in suspects {
            if blob < floor_blob {
                continue;
            }

            // Retain the watermark's in-blob prefix in its containing blob and nothing above.
            let acknowledged = if blob == floor_blob {
                Inner::<E, A>::items_to_bytes(floor.saturating_sub(first_in_blob(
                    pruning_boundary,
                    blob,
                    items_per_blob,
                )?))?
            } else {
                0
            };
            let writer = pending.get(&blob).expect("suspect blob is present");
            let required_items = ceiling
                .saturating_sub(first_in_blob(pruning_boundary, blob, items_per_blob)?)
                .min(items_per_blob);

            // Reads before publication use the page cache, which refuses a non-tip page that does
            // not validate in full, so the scan runs to the end of the cap's page and repairs a
            // page valid only under its shorter durable slot, even when that slot reaches the cap.
            // The watermark advances only after its fsync and rewrites keep the durable slot
            // intact, so under the interrupted-write model that slot covers every byte below the
            // watermark and the repair keeps every acknowledged item, even with the cap below it.
            let limit = if max_size.is_some() {
                let page = u64::from(cfg.page_cache.page_size().get());
                required_items
                    .saturating_mul(Inner::<E, A>::CHUNK_SIZE_U64)
                    .div_ceil(page)
                    .saturating_mul(page)
                    .min(writer.size())
            } else {
                writer.size()
            };
            let recoverable = writer
                .recoverable_prefix_len_at_most(
                    acknowledged,
                    limit,
                    cfg.replay_buffer,
                    ReadOptions::default(),
                )
                .await?;
            let valid_items = recoverable / Inner::<E, A>::CHUNK_SIZE_U64;
            let valid = Inner::<E, A>::items_to_bytes(valid_items)?;
            if valid == writer.size()
                || (max_size.is_some() && recoverable == limit && valid_items >= required_items)
            {
                continue;
            }

            // Missing acknowledged pages surface as `valid < acknowledged`. The
            // scan clamps to the pages physically present, so it can never exceed the size.
            if valid < acknowledged {
                return Err(Error::Corruption(format!(
                    "blob {blob} no longer backs acknowledged items: well-formed prefix {valid} \
                     of size {}",
                    writer.size()
                )));
            }
            warn!(
                blob,
                valid,
                size = writer.size(),
                "truncating to recoverable item prefix"
            );
            let writer = pending.remove(&blob).expect("suspect blob is present");
            pending.insert(blob, writer.truncate(valid).await?);
        }

        let RecoveredBounds {
            size,
            recovery_watermark,
        } = Inner::<E, A>::recover_bounds(
            &pending,
            items_per_blob,
            pruning_boundary,
            checkpoint
                .watermark()
                .map(|watermark| watermark.min(ceiling)),
        )?;

        Ok(Self {
            context,
            cfg,
            checkpoint,
            partition,
            pending,
            discarded,
            bounds: pruning_boundary..size.min(ceiling),
            watermark: recovery_watermark.min(ceiling),
            bounded: max_size.is_some(),
            _marker: PhantomData,
        })
    }

    /// Complete the reset to `target` staged in `checkpoint` without opening any stored blob.
    async fn complete_clear(
        context: E,
        cfg: Config,
        checkpoint: Checkpoint<E>,
        target: u64,
    ) -> Result<Self, Error> {
        let items_per_blob = cfg.items_per_blob.get();
        let new_partition = format!("{}-blobs", cfg.partition);
        Partition::<E>::remove_all(&context, &cfg.partition).await?;
        Partition::<E>::remove_all(&context, &new_partition).await?;
        let partition = Partition::new(
            context.child("blobs"),
            new_partition,
            cfg.page_cache.clone(),
            cfg.write_buffer,
        );

        // Recreate the target tail before the checkpoint publishes reset completion.
        let tail = super::position_to_blob(target, items_per_blob);
        let mut pending = BTreeMap::new();
        pending.insert(tail, partition.open_recovery(tail).await?);
        let checkpoint = checkpoint.finish_clear(items_per_blob, target).await?;
        Ok(Self {
            context,
            cfg,
            checkpoint,
            partition,
            pending,
            discarded: Vec::new(),
            bounds: target..target,
            watermark: target,

            // finish_clear persisted `target` as the watermark, so publication has no selected end
            // to raise.
            bounded: false,
            _marker: PhantomData,
        })
    }

    /// Stage a reset to `size` in `checkpoint`, await `clear_dependents`, then complete the reset
    /// without opening any stored blob. A crash after staging leaves an intent that the next open
    /// completes.
    #[commonware_macros::stability(ALPHA)]
    pub(super) async fn open_cleared<F, Fut>(
        context: E,
        cfg: Config,
        checkpoint: Checkpoint<E>,
        size: u64,
        clear_dependents: F,
    ) -> Result<Self, Error>
    where
        F: FnOnce() -> Fut,
        Fut: Future<Output = Result<(), Error>>,
    {
        // A journal sized at `u64::MAX` can never accept an append (the successor size
        // overflows), so reject it before staging any reset intent.
        if size == u64::MAX {
            return Err(Error::SizeOverflow);
        }
        let checkpoint = checkpoint.stage_clear(size).await?;
        clear_dependents().await?;
        Self::complete_clear(context, cfg, checkpoint, size).await
    }

    /// Open recovery while completing any previously staged reset callback.
    pub(super) async fn init_cleared<F, Fut>(
        context: E,
        cfg: Config,
        max_size: Option<u64>,
        clear_dependents: F,
    ) -> Result<Self, Error>
    where
        F: FnOnce() -> Fut,
        Fut: Future<Output = Result<(), Error>>,
    {
        let checkpoint = Checkpoint::open(context.child("meta"), &cfg.partition).await?;
        if checkpoint.clear_target().is_some() {
            clear_dependents().await?;
        }
        Self::open(context, cfg, checkpoint, max_size).await
    }

    /// Positions the stored blobs may hold, from blob names and the checkpoint without opening
    /// their contents. The end is an upper bound based on the newest blob's capacity. A staged
    /// clear has not yet been applied.
    #[commonware_macros::stability(ALPHA)]
    pub(super) async fn span(
        context: &E,
        cfg: &Config,
        checkpoint: &Checkpoint<E>,
    ) -> Result<Range<u64>, Error> {
        let per_blob = cfg.items_per_blob.get();
        let (_, names) = Partition::select(context, &cfg.partition).await?;
        let indices = Partition::<E>::indices(names)?;
        let start = Inner::<E, A>::recover_pruning_boundary(
            checkpoint.boundary_hint(),
            indices.first().copied(),
            per_blob,
        )?;
        let end = match indices.last() {
            Some(&newest) => super::blob_first_position(newest, per_blob)?.saturating_add(per_blob),
            None => start,
        };
        Ok(start..end)
    }

    /// Exclusive recovered item end.
    pub(super) const fn size(&self) -> u64 {
        self.bounds.end
    }

    /// First retained item position.
    pub(super) const fn pruning_boundary(&self) -> u64 {
        self.bounds.start
    }

    /// Previously acknowledged item end, bounded by the recovery cap.
    pub(super) const fn recovery_watermark(&self) -> u64 {
        self.watermark
    }

    /// Read an item for recovery selection or validation.
    pub(super) async fn item(&self, pos: u64) -> Result<A, Error> {
        if pos < self.bounds.start {
            return Err(Error::ItemPruned(pos));
        }
        if pos >= self.bounds.end {
            return Err(Error::ItemOutOfRange(pos));
        }
        let blob = super::position_to_blob(pos, self.cfg.items_per_blob.get());
        let first = first_in_blob(self.bounds.start, blob, self.cfg.items_per_blob.get())?;
        let offset = Inner::<E, A>::items_to_bytes(pos - first)?;

        // `bounds.end` is derived by walking `pending`, and every mutation keeps `pending`
        // contiguous over `bounds`, so a position inside `bounds` always has a blob.
        let writer = self
            .pending
            .get(&blob)
            .expect("positions inside bounds map to a pending recovery blob");
        Ok(A::decode(
            writer.read_at(offset, A::SIZE).await?.coalesce(),
        )?)
    }

    /// Append entries while initialization exclusively owns the recovered blobs.
    ///
    /// The first append must follow a successful truncate or reset.
    pub(crate) async fn append(mut self: Box<Self>, item: &A) -> Result<Box<Self>, Error> {
        let pos = self.bounds.end;
        let end = pos.checked_add(1).ok_or(Error::SizeOverflow)?;
        let blob = super::position_to_blob(pos, self.cfg.items_per_blob.get());

        // Encode directly into the write buffer when the item fits.
        let appended = self
            .pending
            .get_mut(&blob)
            .is_some_and(|writer| writer.try_append_value(item).is_some());

        let completed = end.is_multiple_of(self.cfg.items_per_blob.get());

        // Take the writer only to open it, append through the owned fallback, or flush it. The
        // owned fallback handles flushing and items larger than the buffer.
        if !appended || completed {
            let mut writer = match self.pending.remove(&blob) {
                Some(writer) => writer,
                None => self.partition.open_recovery(blob).await?,
            };
            if !appended && writer.try_append_value(item).is_none() {
                (writer, _) = writer.append_owned(item.encode_mut().into()).await?;
            }

            // Completed blobs stay open until publication. Flush them to bound retained buffers,
            // without advancing the checkpoint before the recovered prefix is ready to publish.
            if completed {
                writer = writer.sync().await?;
            }
            self.pending.insert(blob, writer);
        }
        self.bounds.end = end;
        Ok(self)
    }

    /// Durably retain at most `max_size` before accepting recovery appends.
    pub(crate) async fn truncate(mut self: Box<Self>, max_size: u64) -> Result<Box<Self>, Error> {
        *self = self.repair_to(max_size).await?;
        Ok(self)
    }

    /// Durably reconcile the recovered blobs and checkpoint to a prefix no longer than
    /// `max_size`.
    async fn repair_to(mut self, max_size: u64) -> Result<Self, Error> {
        let size = self.bounds.end.min(max_size);
        if size < self.bounds.start {
            return Err(Error::ItemPruned(size));
        }
        let items_per_blob = self.cfg.items_per_blob.get();
        let tail_blob = super::position_to_blob(size, items_per_blob);
        let bytes = Inner::<E, A>::items_to_bytes(
            size - first_in_blob(self.bounds.start, tail_blob, items_per_blob)?,
        )?;

        // The checkpoint must not acknowledge a suffix once blob removal begins.
        self.watermark = self.watermark.min(size);
        let boundary_hint =
            (!self.bounds.start.is_multiple_of(items_per_blob)).then_some(self.bounds.start);
        if self.checkpoint.watermark() != Some(self.watermark)
            || self.checkpoint.boundary_hint() != boundary_hint
        {
            self.checkpoint = self
                .checkpoint
                .persist(items_per_blob, self.bounds.start, self.watermark)
                .await?;
        }
        if size == self.bounds.start
            && (!self.discarded.is_empty() || self.pending.keys().any(|&blob| blob > tail_blob))
        {
            // An empty prefix needs a durable boundary blob before its last backing blob is
            // removed.
            if let Entry::Vacant(entry) = self.pending.entry(tail_blob) {
                let writer = self.partition.open_recovery(tail_blob).await?;
                entry.insert(writer.sync().await?);
            }
            self.discarded.retain(|&blob| blob != tail_blob);
        }

        // Make the selected tail newest before changing its partial-page checksum. Remove suffix
        // blobs newest-first so every crash cut leaves a contiguous physical prefix.
        while let Some(blob) = self.discarded.pop() {
            self.partition.remove(blob).await?;
        }
        while let Some((&blob, _)) = self.pending.last_key_value() {
            if blob <= tail_blob {
                break;
            }
            let writer = self.pending.remove(&blob);
            self.partition.remove(blob).await?;
            drop(writer);
        }

        // A truncated tail is made durable by truncation. An appended tail needs an explicit
        // barrier before publication can raise the watermark.
        if let Some(writer) = self.pending.remove(&tail_blob) {
            let writer = if bytes < writer.size() {
                writer.truncate(bytes).await?
            } else {
                writer.sync().await?
            };
            self.pending.insert(tail_blob, writer);
        }
        self.bounds.end = size;
        Ok(self)
    }

    /// Make every recovered blob durable. Clean blobs stay in place. An error drops the extracted
    /// blobs.
    async fn sync_pending(&mut self) -> Result<(), Error> {
        let futures: Vec<_> = self
            .pending
            .extract_if(.., |_, writer| writer.needs_sync())
            .map(|(blob, writer)| async move { writer.sync().await.map(|writer| (blob, writer)) })
            .collect();
        let synced = try_join_all(futures).await?;
        self.pending.extend(synced);
        Ok(())
    }

    /// Flush recovery data and publish the current end as the watermark.
    pub(crate) async fn sync(mut self: Box<Self>) -> Result<Box<Self>, Error> {
        self.sync_pending().await?;
        self.watermark = self.bounds.end;
        self.checkpoint = self
            .checkpoint
            .persist(
                self.cfg.items_per_blob.get(),
                self.bounds.start,
                self.watermark,
            )
            .await?;
        Ok(self)
    }

    /// Reconcile a pruning boundary while recovery owns the blobs.
    pub(crate) async fn prune(
        mut self: Box<Self>,
        position: u64,
    ) -> Result<(Box<Self>, bool), Error> {
        // Recovery pruning advances only to the whole-blob boundary containing the requested
        // retained position.
        let per_blob = self.cfg.items_per_blob.get();
        let blob = super::position_to_blob(position.min(self.bounds.end), per_blob);
        let boundary = super::blob_first_position(blob, per_blob)?;
        if boundary <= self.bounds.start {
            return Ok((self, false));
        }

        // Preserve the retained position before removing its last backing blob. A bounded
        // recovery may have discarded the empty tail that ordinarily records this boundary.
        if let Entry::Vacant(entry) = self.pending.entry(blob) {
            entry.insert(self.partition.open_recovery(blob).await?);
        }

        // The retained anchor must be durable before any older blob is removed.
        self.sync_pending().await?;
        while let Some((&oldest, _)) = self.pending.first_key_value() {
            if oldest >= blob {
                break;
            }
            drop(self.pending.remove(&oldest));
            self.partition.remove(oldest).await?;
        }
        self.bounds.start = boundary;
        Ok((self, true))
    }

    /// Complete a reset while recovery owns the blobs.
    pub(crate) async fn clear_to_size(self: Box<Self>, size: u64) -> Result<Box<Self>, Error> {
        self.clear_to_size_cleared(size, || async { Ok(()) }).await
    }

    /// Stage one durable reset intent, run the reset callback, and replace all journal blobs.
    /// Recovery may preserve an exhausted journal at `u64::MAX`. Explicit resets reject it.
    pub(super) async fn clear_to_size_cleared<F, Fut>(
        self: Box<Self>,
        size: u64,
        clear_dependents: F,
    ) -> Result<Box<Self>, Error>
    where
        F: FnOnce() -> Fut,
        Fut: Future<Output = Result<(), Error>>,
    {
        // A journal sized at `u64::MAX` can never accept an append, so reject it before staging
        // any reset intent.
        if size == u64::MAX {
            return Err(Error::SizeOverflow);
        }
        let Self {
            context,
            cfg,
            checkpoint,
            partition,
            pending,
            ..
        } = *self;

        // Release the recovered blob handles and their partition before the reset recreates both.
        drop(pending);
        drop(partition);

        // The durable intent makes `size` authoritative across every later crash cut.
        let checkpoint = checkpoint.stage_clear(size).await?;

        // Dependent state is cleared under the same intent before either journal partition is
        // removed.
        clear_dependents().await?;
        Ok(Box::new(
            Self::complete_clear(context, cfg, checkpoint, size).await?,
        ))
    }

    /// Select a retained prefix and finish it before creating a live journal.
    pub(super) async fn publish(mut self, max_size: u64) -> Result<Inner<E, A>, Error> {
        // Complete the durable prefix selection before sealing blobs for live access.
        self = self.repair_to(max_size).await?;
        let size = self.bounds.end;
        let items_per_blob = self.cfg.items_per_blob.get();
        let tail_blob = super::position_to_blob(size, items_per_blob);
        let blobs = Writable::recover(self.partition, self.pending, tail_blob).await?;

        // A bounded open publishes the selected end after every retained blob is durable.
        if self.bounded && self.checkpoint.watermark() != Some(size) {
            self.checkpoint = self
                .checkpoint
                .persist(items_per_blob, self.bounds.start, size)
                .await?;
        }
        let metrics = Metrics::new(self.context);
        metrics.update(size, self.bounds.start, items_per_blob);
        Ok(Inner::from_blobs(
            blobs,
            self.checkpoint,
            self.bounds.start..size,
            self.cfg.items_per_blob,
            metrics,
        ))
    }
}

impl<E: Context, A: CodecFixedShared> Inner<E, A> {
    /// Size of each entry in bytes. Evaluating this rejects zero-size item types at compile
    /// time, which would otherwise divide by zero in the chunk math.
    pub const CHUNK_SIZE: NonZeroUsize = match NonZeroUsize::new(A::SIZE) {
        Some(size) => size,
        None => panic!("journal item size must be nonzero"),
    };

    /// Size of each entry in bytes (as u64).
    pub const CHUNK_SIZE_U64: u64 = Self::CHUNK_SIZE.get() as u64;

    /// Convert an item count to a byte length, failing on overflow.
    fn items_to_bytes(items: u64) -> Result<u64, Error> {
        items
            .checked_mul(Self::CHUNK_SIZE_U64)
            .ok_or(Error::OffsetOverflow)
    }

    /// Construct a journal from recovered blobs.
    fn from_blobs(
        blobs: Writable<E>,
        checkpoint: Checkpoint<E>,
        bounds: Range<u64>,
        items_per_blob: NonZeroU64,
        metrics: Metrics<E>,
    ) -> Self {
        Self {
            blobs,
            barrier: Barrier::new(
                checkpoint
                    .watermark()
                    .expect("recovery watermark must exist after init"),
            ),
            checkpoint,
            bounds,
            items_per_blob,
            metrics: Arc::new(metrics),
            _phantom: PhantomData,
        }
    }

    /// See [Journal::init].
    pub(crate) async fn init(context: E, cfg: Config) -> Result<Self, Error> {
        let checkpoint = Checkpoint::open(context.child("meta"), &cfg.partition).await?;
        Recovery::<E, A>::open(context, cfg, checkpoint, None)
            .await?
            .publish(u64::MAX)
            .await
    }

    /// Recover the journal bounds and any tail repair from the reconciled pruning boundary and
    /// blob state.
    ///
    /// Blob lengths recover the contiguous size from the supplied boundary. A watermark beyond
    /// that size is corruption. The caller persists the checkpoint before repairing the suffix.
    fn recover_bounds(
        pending: &BTreeMap<u64, PagedRecovery<E::Blob>>,
        items_per_blob: u64,
        pruning_boundary: u64,
        watermark_hint: Option<u64>,
    ) -> Result<RecoveredBounds, Error> {
        let (size, has_gap) =
            Self::recover_by_walking_lengths(pending, items_per_blob, pruning_boundary)?;

        let recovery_watermark = match watermark_hint {
            Some(watermark) if watermark > size => {
                // The dual-CRC page mechanism prevents losing previously-synced data, and
                // clear_to_size updates the watermark atomically via the staged clear intent. A
                // watermark beyond the recoverable size indicates external corruption.
                return Err(Error::Corruption(format!(
                    "recovery watermark {watermark} exceeds recoverable size {size}"
                )));
            }
            Some(watermark) => watermark,
            None if has_gap => {
                // A legacy journal with a short non-tail blob violates the old rollover-sync
                // invariant (each blob was fsynced before the next received writes).
                return Err(Error::Corruption(
                    "legacy journal has a short non-tail blob".into(),
                ));
            }
            // Legacy journals have no watermark. Under the old rollover-sync invariant, all
            // non-tail blobs are durable; only the tail may have unfsynced data.
            None => first_in_blob(
                pruning_boundary,
                super::position_to_blob(size, items_per_blob),
                items_per_blob,
            )?,
        };

        Ok(RecoveredBounds {
            size,
            recovery_watermark,
        })
    }

    /// Recover the pruning boundary from the checkpoint hint if it still matches the oldest
    /// retained blob.
    ///
    /// A missing or blob-aligned hint means the blob boundary is authoritative. A mid-blob hint
    /// is trusted only when it belongs to the current oldest blob.
    fn recover_pruning_boundary(
        boundary_hint: Option<u64>,
        oldest_blob: Option<u64>,
        items_per_blob: u64,
    ) -> Result<u64, Error> {
        let blob_boundary = match oldest_blob {
            Some(oldest) => super::blob_first_position(oldest, items_per_blob)?,
            None => 0,
        };

        let Some(boundary_hint) = boundary_hint else {
            return Ok(blob_boundary);
        };
        if boundary_hint.is_multiple_of(items_per_blob) {
            return Ok(blob_boundary);
        }

        let hint_blob = super::position_to_blob(boundary_hint, items_per_blob);
        match oldest_blob {
            Some(oldest_blob) if hint_blob == oldest_blob => Ok(boundary_hint),
            Some(oldest_blob) if hint_blob < oldest_blob => {
                warn!(
                    hint_blob,
                    oldest_blob, "crash repair: boundary hint stale, computing from blobs"
                );
                Ok(blob_boundary)
            }
            Some(oldest_blob) => {
                // A hint ahead of blob state should never arise: prune removes blobs before
                // sync persists the checkpoint, and clear_to_size stages a clear intent.
                Err(Error::Corruption(format!(
                    "boundary hint references blob {hint_blob} \
                     but oldest blob is blob {oldest_blob}"
                )))
            }
            None => {
                // A mid-blob hint with no blobs should never arise: a staged clear is completed
                // before we get here, and no other operation removes all blobs without updating
                // the checkpoint.
                Err(Error::Corruption(format!(
                    "boundary hint references blob {hint_blob} but no blobs exist"
                )))
            }
        }
    }

    /// Classify a blob's untrusted on-disk length against its capacity. A missing blob counts
    /// as zero length, surfacing as a gap.
    fn classify_fill(
        pending: &BTreeMap<u64, PagedRecovery<E::Blob>>,
        items_per_blob: u64,
        pruning_boundary: u64,
        blob: u64,
    ) -> Result<BlobFill, Error> {
        let len = pending
            .get(&blob)
            .map_or(0, |writer| writer.size() / Self::CHUNK_SIZE_U64);
        // A blob's capacity is `items_per_blob`, unless the pruning boundary falls mid-blob
        // (from `init_at_size`), in which case the skipped prefix reduces it.
        let start = super::blob_first_position(blob, items_per_blob)?;
        let skipped = pruning_boundary.saturating_sub(start).min(items_per_blob);
        let capacity = items_per_blob - skipped;
        Ok(match len.cmp(&capacity) {
            std::cmp::Ordering::Less => BlobFill::Short { len },
            std::cmp::Ordering::Equal => BlobFill::Full { len },
            std::cmp::Ordering::Greater => BlobFill::Overfull { len, capacity },
        })
    }

    /// Recover size by walking blob lengths from oldest to newest, truncating at the
    /// first short or missing non-tail blob.
    ///
    /// `pruning_boundary` is trusted (already reconciled by `recover_pruning_boundary`); blob
    /// lengths are untrusted disk state. The returned size is chunk-exact and the retained
    /// prefix is contiguous.
    fn recover_by_walking_lengths(
        pending: &BTreeMap<u64, PagedRecovery<E::Blob>>,
        items_per_blob: u64,
        pruning_boundary: u64,
    ) -> Result<(u64, bool), Error> {
        let oldest = pending.keys().next().copied();
        let newest = pending.keys().next_back().copied();

        let (Some(oldest), Some(newest)) = (oldest, newest) else {
            return Ok((pruning_boundary, false));
        };

        let mut size = pruning_boundary;
        for blob in oldest..=newest {
            let fill = Self::classify_fill(pending, items_per_blob, pruning_boundary, blob)?;
            match fill {
                // Complete: count its items and keep walking.
                BlobFill::Full { len } => {
                    size = size.checked_add(len).ok_or(Error::OffsetOverflow)?;
                }
                // The newest blob is the append frontier; short is normal.
                BlobFill::Short { len } if blob == newest => {
                    size = size.checked_add(len).ok_or(Error::OffsetOverflow)?;
                    return Ok((size, false));
                }
                // A short or missing interior blob is a gap in durable data: everything newer
                // is unreachable. Truncate here.
                BlobFill::Short { len } => {
                    size = size.checked_add(len).ok_or(Error::OffsetOverflow)?;
                    return Ok((size, true));
                }
                BlobFill::Overfull { len, capacity } => {
                    return Err(Error::Corruption(format!(
                        "blob {blob} has too many items: expected at most {capacity}, got {len}"
                    )));
                }
            }
        }

        Ok((size, false))
    }

    /// See [Journal::init_at_size].
    #[commonware_macros::stability(ALPHA)]
    pub(crate) async fn init_at_size(context: E, cfg: Config, size: u64) -> Result<Self, Error> {
        // Fail before writing intent if existing blob partitions are already inconsistent.
        Partition::select(&context, &cfg.partition).await?;

        // A journal sized at `u64::MAX` can never accept an append (the successor size
        // overflows), so reject it before opening the checkpoint.
        if size == u64::MAX {
            return Err(Error::SizeOverflow);
        }
        let checkpoint = Checkpoint::open(context.child("meta"), &cfg.partition).await?;
        Recovery::<E, A>::open_cleared(context, cfg, checkpoint, size, || async { Ok(()) })
            .await?
            .publish(u64::MAX)
            .await
    }

    /// Begin durably persisting the data blobs.
    pub(super) async fn start_data_sync(
        mut self: Box<Self>,
    ) -> Result<(Box<Self>, Handle<()>), Error> {
        let (blobs, handle) = self.blobs.start_sync().await?;
        self.blobs = blobs;
        let completion: SyncCompletion = handle.boxed().shared();
        self.barrier.record(self.bounds.end, completion.clone());
        Ok((self, Handle::from_future(completion)))
    }

    /// Begin raising the recovery watermark toward `size`, capped at the barrier.
    pub(super) async fn start_watermark_sync(
        mut self: Box<Self>,
        size: u64,
    ) -> Result<(Box<Self>, Handle<()>), Error> {
        let size = size.min(self.barrier.boundary());
        let (checkpoint, handle) = self.checkpoint.start_watermark_sync(size).await?;
        self.checkpoint = checkpoint;
        Ok((self, handle))
    }

    /// See [Journal::start_sync].
    pub(crate) async fn start_sync(self: Box<Self>) -> Result<(Box<Self>, Handle<()>), Error> {
        self.metrics.start_sync_calls.inc();

        // Start data durability, then cap watermark publication at the completed barrier.
        let (mut journal, data) = self.start_data_sync().await?;
        let size = journal.barrier.boundary();
        let (journal, watermark) = journal.start_watermark_sync(size).await?;

        // Completion requires both the data and the selected watermark to be durable.
        let handle = Handle::from_future(async move {
            data.await?;
            watermark.await
        });
        Ok((journal, handle))
    }

    /// See [Journal::commit].
    pub(crate) async fn commit(mut self: Box<Self>) -> Result<Box<Self>, Error> {
        let _timer = self.metrics.commit_timer();
        self.metrics.commit_calls.inc();
        let size = self.bounds.end;
        let (blobs, handle) = self.blobs.start_sync().await?;
        self.blobs = blobs;
        handle.await?;
        self.barrier.mark_durable(size);
        Ok(self)
    }

    /// See [Journal::sync].
    pub(crate) async fn sync(mut self: Box<Self>) -> Result<Box<Self>, Error> {
        let _timer = self.metrics.sync_timer();
        self.metrics.sync_calls.inc();

        // Complete data durability before advancing the checkpoint that acknowledges it.
        let size = self.bounds.end;
        let (blobs, handle) = self.blobs.start_sync().await?;
        self.blobs = blobs;
        handle.await?;
        self.barrier.mark_durable(size);

        // Publish the durable prefix as the recovery watermark.
        self.checkpoint = self
            .checkpoint
            .persist(self.items_per_blob.get(), self.bounds.start, size)
            .await?;
        Ok(self)
    }

    /// See [Journal::snapshot].
    pub(crate) async fn snapshot(
        mut self: Box<Self>,
    ) -> Result<(Box<Self>, Reader<'static, E, A>), Error> {
        let (blobs, snapshot) = self.blobs.snapshot().await?;
        self.blobs = blobs;
        let reader = Reader {
            blobs: snapshot,
            bounds: self.bounds.clone(),
            items_per_blob: self.items_per_blob,
            metrics: self.metrics.clone(),
            _phantom: PhantomData,
        };
        Ok((self, reader))
    }

    /// A reader borrowing the journal's live state.
    pub(super) fn reader(&self) -> Reader<'_, E, A> {
        Reader {
            blobs: self.blobs.reader(),
            bounds: self.bounds.clone(),
            items_per_blob: self.items_per_blob,
            metrics: self.metrics.clone(),
            _phantom: PhantomData,
        }
    }

    /// Return the total number of items in the journal, irrespective of pruning. The next value
    /// appended to the journal will be at this position.
    pub const fn size(&self) -> u64 {
        self.bounds.end
    }

    /// See [Journal::append].
    pub(crate) async fn append(mut self: Box<Self>, item: &A) -> Result<(Box<Self>, u64), Error> {
        let _timer = self.metrics.append_timer();
        self.metrics.append_calls.inc();
        let new_size = self.bounds.end.checked_add(1).ok_or(Error::SizeOverflow)?;
        if self.blobs.try_append_value(item).is_none() {
            return self
                .append_many_inner(Many::Flat(std::slice::from_ref(item)))
                .await;
        }
        self = self.advance_tail(new_size).await?;
        let position = self.finish_append();
        Ok((self, position))
    }

    /// See [Journal::append_many].
    pub(crate) async fn append_many(
        self: Box<Self>,
        items: Many<'_, A>,
    ) -> Result<(Box<Self>, u64), Error> {
        let _timer = self.metrics.append_many_timer();
        self.metrics.append_many_calls.inc();
        self.append_many_inner(items).await
    }

    // Shared implementation for `append` and `append_many`; public wrappers record metrics.
    async fn append_many_inner(
        self: Box<Self>,
        items: Many<'_, A>,
    ) -> Result<(Box<Self>, u64), Error> {
        let prepared = self.prepare_append(items);
        self.write_encoded(prepared).await
    }

    /// See [Journal::prepare_append].
    pub(crate) fn prepare_append(&self, items: Many<'_, A>) -> PreparedAppend<A> {
        // Encode all items into a single contiguous buffer up front.
        // Uses Write::write directly to avoid per-item Bytes allocations from Encode::encode.
        let mut buf = Vec::with_capacity(items.len() * A::SIZE);
        match items {
            Many::Flat(items) => {
                for item in items {
                    item.write(&mut buf);
                }
            }
            Many::Nested(nested_items) => {
                for items in nested_items {
                    for item in *items {
                        item.write(&mut buf);
                    }
                }
            }
        }
        PreparedAppend {
            buf,
            _marker: PhantomData,
        }
    }

    /// See [Journal::append_prepared].
    pub(crate) async fn append_prepared(
        self: Box<Self>,
        prepared: PreparedAppend<A>,
    ) -> Result<(Box<Self>, u64), Error> {
        let _timer = self.metrics.append_prepared_timer();
        self.metrics.append_prepared_calls.inc();
        self.write_encoded(prepared).await
    }

    // Write pre-encoded items; shared by all append paths. Records no call metrics.
    async fn write_encoded(
        mut self: Box<Self>,
        prepared: PreparedAppend<A>,
    ) -> Result<(Box<Self>, u64), Error> {
        let items_buf = prepared.buf;
        let items_count = items_buf.len() / A::SIZE;
        if items_count == 0 {
            return Err(Error::EmptyAppend);
        }
        let items_buf = IoBuf::from(items_buf);

        // Reject the append before writing anything if it would push the size past `u64::MAX`.
        // This keeps the in-loop size arithmetic safe.
        self.bounds
            .end
            .checked_add(items_count as u64)
            .ok_or(Error::SizeOverflow)?;

        // Split the batch at blob boundaries so each full tail can be sealed before continuing.
        let mut written = 0;
        while written < items_count {
            let batch_count = super::batch_count_to_blob_boundary(
                self.bounds.end,
                items_count - written,
                self.items_per_blob.get(),
            );
            let start = written * A::SIZE;
            let end = start + batch_count * A::SIZE;
            let new_size = self.bounds.end + batch_count as u64;

            // Buffer a batch in place when it fits. Otherwise own the writer across the flush.
            if self
                .blobs
                .try_append(&items_buf.as_ref()[start..end])
                .is_none()
            {
                (self.blobs, _) = self.blobs.append_owned(items_buf.slice(start..end)).await?;
            }
            self = self.advance_tail(new_size).await?;
            written += batch_count;
        }

        let position = self.finish_append();
        Ok((self, position))
    }

    // Advance appended bounds and rotate a full tail. Both append paths use the same ordering.
    async fn advance_tail(mut self: Box<Self>, new_size: u64) -> Result<Box<Self>, Error> {
        self.bounds.end = new_size;
        if new_size.is_multiple_of(self.items_per_blob.get()) {
            self.blobs = self.blobs.seal_tail().await?;
        }
        Ok(self)
    }

    // Record state metrics once after a successful append, including multi-blob batches.
    fn finish_append(&self) -> u64 {
        self.metrics.update(
            self.bounds.end,
            self.bounds.start,
            self.items_per_blob.get(),
        );
        self.bounds.end - 1
    }

    /// Return the location before which all items have been pruned.
    pub const fn pruning_boundary(&self) -> u64 {
        self.bounds.start
    }

    /// See [Journal::prune].
    pub(crate) async fn prune(
        mut self: Box<Self>,
        min_item_pos: u64,
    ) -> Result<(Box<Self>, bool), Error> {
        // Calculate the blob that would contain min_item_pos, capped to the tail (which is
        // guaranteed to exist by our invariant).
        let target_blob = super::position_to_blob(min_item_pos, self.items_per_blob.get());
        let tail_blob = super::position_to_blob(self.bounds.end, self.items_per_blob.get());
        let min_blob = std::cmp::min(target_blob, tail_blob);

        if min_blob <= self.blobs.oldest_blob_index() {
            return Ok((self, false));
        }

        // Make all data durable before removing any: the prune target may be justified by an
        // appended-but-unflushed item (e.g. a consumer's commit record), and removals are
        // durable, so pruning without this sync could leave a recovered journal whose
        // surviving items no longer justify its boundary. The sync also covers unsynced
        // survivors above the boundary: removal may be interrupted, and recovery truncates at
        // the first torn item, so an unsynced survivor could discard every synced blob
        // behind it.
        let (blobs, sync) = self.blobs.start_sync().await?;
        self.blobs = blobs;
        sync.await?;
        self.barrier.mark_durable(self.bounds.end);

        let new_boundary = super::blob_first_position(min_blob, self.items_per_blob.get())?;
        self.blobs = self.blobs.prune(min_blob).await?;
        self.bounds.start = new_boundary;

        self.metrics.update(
            self.bounds.end,
            self.bounds.start,
            self.items_per_blob.get(),
        );

        Ok((self, true))
    }

    /// Wait for every started sync of the data and the checkpoint to complete, surfacing a
    /// failure without changing storage.
    pub(super) async fn wait_for_sync(mut self) -> Result<Self, Error> {
        self.blobs = self.blobs.wait_for_sync().await?;
        self.checkpoint.wait_for_sync().await?;
        Ok(self)
    }

    /// See [Journal::destroy].
    pub(crate) async fn destroy(self) -> Result<(), Error> {
        // A failed checkpoint sync must surface before any data is removed, since the surviving
        // checkpoint would otherwise acknowledge items the removed blobs no longer back.
        let mut journal = self.wait_for_sync().await?;

        // Stage a clear to zero before removing anything: a destroy interrupted after the intent is
        // durable reopens as a fresh journal, and one interrupted before it reopens as a crash
        // would recover it.
        journal.checkpoint = journal.checkpoint.stage_clear(0).await?;
        journal.blobs.destroy().await?;
        journal.checkpoint.destroy().await
    }

    /// Clear all data and advance the journal to a new starting position.
    ///
    /// Unlike `destroy`, this keeps the journal alive so it can be reused. After clearing, the
    /// journal will behave as if initialized with `init_at_size(new_size)`.
    ///
    /// # Crash Safety
    ///
    /// In the event of a crash during this call, upon restart recovery will ensure the journal is
    /// either still in its prior state, or has bounds `new_size..new_size`.
    #[commonware_macros::stability(ALPHA)]
    pub(crate) async fn clear_to_size(
        mut self: Box<Self>,
        new_size: u64,
    ) -> Result<Box<Self>, Error> {
        // Live reset advances the append position. Prefix selection belongs to initialization.
        if new_size < self.bounds.end {
            return Err(Error::ItemOutOfRange(new_size));
        }

        // A journal sized at `u64::MAX` can never accept an append, matching `init_at_size`.
        if new_size == u64::MAX {
            return Err(Error::SizeOverflow);
        }

        // Durably record the intent first, so a crash mid-clear is finished on reopen.
        self.checkpoint = self.checkpoint.stage_clear(new_size).await?;

        // Remove every blob, then start fresh at the new size.
        self.blobs = self
            .blobs
            .clear(super::position_to_blob(new_size, self.items_per_blob.get()))
            .await?;
        self.bounds = new_size..new_size;
        self.barrier = Barrier::new(new_size);

        // Complete the clear in the checkpoint.
        self.checkpoint = self
            .checkpoint
            .finish_clear(self.items_per_blob.get(), new_size)
            .await?;

        self.metrics.update(
            self.bounds.end,
            self.bounds.start,
            self.items_per_blob.get(),
        );
        Ok(self)
    }

    /// Durably stage a clear to `new_size` without completing it.
    ///
    /// This records a recoverable intent so a caller can clear dependent sibling state before
    /// calling `clear_to_size` to finish. If a crash interrupts the sequence, the next `init`
    /// completes the staged clear. The follow-up `clear_to_size` re-stages the same target
    /// idempotently.
    #[commonware_macros::stability(ALPHA)]
    pub(super) async fn stage_clear_intent(
        mut self: Box<Self>,
        new_size: u64,
    ) -> Result<Box<Self>, Error> {
        // A journal sized at `u64::MAX` can never accept an append, matching `init_at_size`.
        if new_size == u64::MAX {
            return Err(Error::SizeOverflow);
        }
        if new_size < self.bounds.end {
            return Err(Error::ItemOutOfRange(new_size));
        }
        self.checkpoint = self.checkpoint.stage_clear(new_size).await?;
        Ok(self)
    }
}

/// Implementation of [super::Mutable] for fixed-size value journals.
///
/// # 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 above the recovery watermark's acknowledged prefix is considered
/// the new end of the journal (and the underlying blob is truncated to the last valid item).
/// Data at or below that prefix is never repaired: a blob that no longer backs its acknowledged
/// items fails init, and other damage there surfaces as a read error. Repair is performed during
/// init.
///
/// Mutating functions consume the journal and return it only on success: an error (or a dropped
/// future) destroys the handle.
pub struct Journal<E: Context, A>(Box<Inner<E, A>>);

impl<E: Context, A: CodecFixedShared> std::fmt::Debug for Journal<E, A> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Journal")
            .field("bounds", &super::Contiguous::bounds(self))
            .finish_non_exhaustive()
    }
}

impl<E: Context, A: CodecFixedShared> Journal<E, A> {
    /// Initialize a new `Journal` instance.
    ///
    /// All backing blobs are opened during initialization. Recovery scans the two newest blobs,
    /// skipping blobs and pages the checkpoint watermark already acknowledges, and truncates
    /// each to the whole items backed by valid pages. The `replay` method can be used to
    /// iterate over all items in the `Journal`.
    pub async fn init(context: E, cfg: Config) -> Result<Self, Error> {
        Ok(Self(Box::new(Inner::init(context, cfg).await?)))
    }

    /// Open a retained prefix containing at most `max_size` items, including pruned positions.
    ///
    /// A cap above the recovered end preserves that end. A cap below the retained start returns
    /// [Error::ItemPruned]. Successful initialization durably discards the suffix. Subsequent
    /// appends may exceed the cap.
    pub async fn init_at_most(context: E, cfg: Config, max_size: u64) -> Result<Self, Error> {
        let checkpoint = Checkpoint::open(context.child("meta"), &cfg.partition).await?;
        let recovery = Recovery::<E, A>::open(context, cfg, checkpoint, Some(max_size)).await?;
        Ok(Self(Box::new(recovery.publish(max_size).await?)))
    }

    /// Initialize a `Journal` in a fully-pruned state at `size`: existing data is cleared and the
    /// journal behaves as if `size` items were appended then pruned. It is empty (`bounds` is
    /// `size..size`) and the next `append` writes at position `size`. Used for state sync.
    ///
    /// # Crash Safety
    /// In the event of a crash during this call, upon restart recovery will ensure the journal is
    /// either still in its prior state, or has bounds `size..size`.
    #[commonware_macros::stability(ALPHA)]
    pub async fn init_at_size(context: E, cfg: Config, size: u64) -> Result<Self, Error> {
        Ok(Self(Box::new(
            Inner::init_at_size(context, cfg, size).await?,
        )))
    }

    /// Discard all items and reposition the journal at or beyond its current end.
    #[commonware_macros::stability(ALPHA)]
    pub(crate) async fn clear_to_size(mut self, new_size: u64) -> Result<Self, Error> {
        self.0 = self.0.clear_to_size(new_size).await?;
        Ok(self)
    }

    /// Durably persists the current state of the structure.
    ///
    /// Does not advance the recovery watermark, so reopen may replay entries above it. Use
    /// `sync()` to advance the watermark and to ensure that a crash after this call doesn't
    /// require any recovery.
    pub async fn commit(mut self) -> Result<Self, Error> {
        self.0 = self.0.commit().await?;
        Ok(self)
    }

    /// Begin durably persisting the current state of the structure.
    ///
    /// Awaiting the returned [Handle] guarantees state appended before this call survives a
    /// crash. Also tries to advance the recovery watermark to the previous proven durable
    /// size, bounding startup recovery. Only `sync()` guarantees a current watermark.
    ///
    /// At most one data sync and one watermark sync are in flight at a time: this call waits for
    /// the prior call's syncs before starting new ones. It does not wait for a pending rollover
    /// fsync: the returned handle joins it, so an earlier call's handle may still be pending when
    /// this call returns. Reads proceed while the returned handle is pending, and appends proceed
    /// while they fit in the write buffer.
    ///
    /// Flush errors are returned directly. Dropping the handle does not cancel the sync or lose its
    /// failure: a later operation reports it, no later than the next [Self::sync].
    pub async fn start_sync(mut self) -> Result<(Self, Handle<()>), Error> {
        let (inner, handle) = self.0.start_sync().await?;
        self.0 = inner;
        Ok((self, handle))
    }

    /// Durably persist the current state of the structure, ensuring no recovery is required in the
    /// event of a crash following this call.
    ///
    /// Advances the recovery watermark to the current size.
    pub async fn sync(mut self) -> Result<Self, Error> {
        self.0 = self.0.sync().await?;
        Ok(self)
    }

    /// Capture an owned snapshot ([`Reader`]) over the current journal. Bounds are frozen at
    /// creation, and the snapshot stays readable across concurrent appends and prunes. It keeps
    /// the journal's blobs open, so reopening a partition that still holds one of them fails
    /// while the snapshot is alive.
    pub async fn snapshot(mut self) -> Result<(Self, Reader<'static, E, A>), Error> {
        let (inner, reader) = self.0.snapshot().await?;
        self.0 = inner;
        Ok((self, reader))
    }

    /// Return the total number of items in the journal, irrespective of pruning. The next value
    /// appended to the journal will be at this position.
    pub fn size(&self) -> u64 {
        self.0.size()
    }

    /// Append a new item to the journal, returning its position.
    ///
    /// # Errors
    ///
    /// Returns an error if the underlying storage operation fails.
    pub async fn append(mut self, item: &A) -> Result<(Self, u64), Error> {
        let (inner, position) = self.0.append(item).await?;
        self.0 = inner;
        Ok((self, position))
    }

    /// Append items to the journal, returning the position of the last item appended.
    ///
    /// Returns [Error::EmptyAppend] if items is empty.
    pub async fn append_many(mut self, items: Many<'_, A>) -> Result<(Self, u64), Error> {
        let (inner, position) = self.0.append_many(items).await?;
        self.0 = inner;
        Ok((self, position))
    }

    /// Encode `items` into a buffer that can be appended later with [`Self::append_prepared`].
    ///
    /// This lets callers serialize borrowed items synchronously, release those borrows, and
    /// perform the append without holding unrelated locks across journal I/O.
    pub fn prepare_append(&self, items: Many<'_, A>) -> PreparedAppend<A> {
        self.0.prepare_append(items)
    }

    /// Append items encoded by [`Self::prepare_append`], returning the position of the last item
    /// appended.
    ///
    /// Returns [Error::EmptyAppend] if `prepared` contains no items.
    pub async fn append_prepared(
        mut self,
        prepared: PreparedAppend<A>,
    ) -> Result<(Self, u64), Error> {
        let (inner, position) = self.0.append_prepared(prepared).await?;
        self.0 = inner;
        Ok((self, position))
    }

    /// Return the location before which all items have been pruned.
    pub fn pruning_boundary(&self) -> u64 {
        self.0.pruning_boundary()
    }

    /// Allow the journal to prune items older than `min_item_pos`. The journal may not prune all
    /// such items in order to preserve blob boundaries, but the amount of such items will always be
    /// less than the configured number of items per blob. Returns true if any items were pruned.
    ///
    /// Readers holding earlier snapshots keep reading pruned blobs through their own handles;
    /// later snapshots observe [Error::ItemPruned].
    ///
    /// Note that this operation may NOT be atomic, however it's guaranteed not to leave gaps in the
    /// event of failure as items are always pruned in order from oldest to newest.
    pub async fn prune(mut self, min_item_pos: u64) -> Result<(Self, bool), Error> {
        let (inner, pruned) = self.0.prune(min_item_pos).await?;
        self.0 = inner;
        Ok((self, pruned))
    }

    /// Remove any persisted data created by the journal. A started sync that failed is reported
    /// before anything is removed.
    ///
    /// # Crash Safety
    ///
    /// An interrupted destroy leaves a journal that reopens either as a crash would recover it or
    /// as a fresh journal at position zero, and that a retried destroy removes.
    pub async fn destroy(self) -> Result<(), Error> {
        self.0.destroy().await
    }
}

/// A reader over a fixed journal.
pub struct Reader<'a, E: Context, A> {
    blobs: Blobs<'a, E::Blob>,
    bounds: Range<u64>,
    items_per_blob: NonZeroU64,
    metrics: Arc<Metrics<E>>,
    _phantom: PhantomData<A>,
}

impl<E: Context, A: CodecFixedShared> Reader<'_, E, A> {
    /// Validate a position to be read: must lie within `bounds`.
    const fn validate_readable(&self, pos: u64) -> Result<(), Error> {
        if pos >= self.bounds.end {
            return Err(Error::ItemOutOfRange(pos));
        }
        if pos < self.bounds.start {
            return Err(Error::ItemPruned(pos));
        }
        Ok(())
    }

    /// Resolve a blob-sharing group of positions to its blob number and per-position byte
    /// offsets within the blob.
    fn locate_group(&self, group: &[u64]) -> Result<(u64, Vec<u64>), Error> {
        let items_per_blob = self.items_per_blob.get();
        let blob = super::position_to_blob(group[0], items_per_blob);
        let first_position = first_in_blob(self.bounds.start, blob, items_per_blob)?;
        let mut offsets = Vec::with_capacity(group.len());
        for &pos in group {
            offsets.push(Inner::<E, A>::items_to_bytes(pos - first_position)?);
        }
        Ok((blob, offsets))
    }

    /// Shared body of [`super::Contiguous::read_many`] and the variable journal's offsets
    /// reads; the callers record the batch-read metrics, so routing them through `read_many`
    /// would count every batch twice.
    pub(super) async fn read_many_inner(&self, positions: &[u64]) -> Result<Vec<A>, Error> {
        if positions.is_empty() {
            return Ok(Vec::new());
        }
        assert!(
            positions.is_sorted_by(|a, b| a < b),
            "positions must be strictly increasing"
        );
        for &pos in positions {
            self.validate_readable(pos)?;
        }

        let items_per_blob = self.items_per_blob.get();

        // Read all positions grouped by blob. Positions are sorted, so `chunk_by` splits them into
        // maximal runs that share one blob. Each group goes through the blob's batched read,
        // which serves page-cache and tip-buffer hits under a single lock acquisition and reads only
        // true misses from the blob (concurrently).
        let mut result: Vec<A> = Vec::with_capacity(positions.len());
        let mut buf = vec![0u8; positions.len() * A::SIZE];

        // The buffer is pre-sized for every position, so each group can own a disjoint slice and
        // all groups can read concurrently.
        let mut reads = Vec::new();
        let mut remaining_buf = buf.as_mut_slice();
        for group in positions.chunk_by(|a, b| {
            super::position_to_blob(*a, items_per_blob)
                == super::position_to_blob(*b, items_per_blob)
        }) {
            let (blob_num, blob_offsets) = self.locate_group(group)?;
            let blob = self
                .blobs
                .get(blob_num)
                .expect("positions in bounds map to a retained blob");
            let (group_buf, rest) = remaining_buf.split_at_mut(group.len() * A::SIZE);
            remaining_buf = rest;
            reads.push(async move {
                blob.read_many_into(group_buf, &blob_offsets, Inner::<E, A>::CHUNK_SIZE)
                    .await
            });
        }
        let hits: u64 = try_join_all(reads)
            .await?
            .into_iter()
            .map(|group_hits| group_hits as u64)
            .sum();

        let mut bytes = Bytes::from(buf);
        for _ in positions {
            result.push(A::decode((&mut bytes).take(A::SIZE)).map_err(Error::Codec)?);
        }

        self.metrics.cache_hits.inc_by(hits);
        self.metrics
            .cache_misses
            .inc_by(positions.len() as u64 - hits);
        self.metrics.items_read.inc_by(positions.len() as u64);
        Ok(result)
    }

    /// Resolve `pos` to its blob and byte offset within the blob.
    fn locate(&self, pos: u64) -> Result<(Blob<'_, E::Blob>, u64), Error> {
        self.validate_readable(pos)?;
        let items_per_blob = self.items_per_blob.get();
        let blob = super::position_to_blob(pos, items_per_blob);
        let pos_in_blob = pos - first_in_blob(self.bounds.start, blob, items_per_blob)?;
        let offset = Inner::<E, A>::items_to_bytes(pos_in_blob)?;
        let blob = self
            .blobs
            .get(blob)
            .expect("position in bounds maps to a retained blob");
        Ok((blob, offset))
    }

    /// Read `pos` from storage, for callers that already missed the page cache.
    pub(super) async fn read_miss(&self, pos: u64) -> Result<A, Error> {
        let _timer = self.metrics.read_timer();
        let (blob, offset) = self.locate(pos)?;
        self.metrics.cache_misses.inc();
        let bufs = blob.read_at(offset, A::SIZE).await?;
        let item = A::decode(bufs.coalesce()).map_err(Error::Codec)?;
        self.metrics.items_read.inc();
        Ok(item)
    }

    /// Probe `positions` (strictly increasing) against the page cache, returning one slot per
    /// position: `Some(item)` for sync hits and `None` for positions that require I/O, fail to
    /// decode, or fall outside `bounds()`.
    pub(super) fn probe_items(&self, positions: &[u64]) -> Vec<Option<A>> {
        assert!(
            positions.is_sorted_by(|a, b| a < b),
            "positions must be strictly increasing"
        );
        let mut out: Vec<Option<A>> = (0..positions.len()).map(|_| None).collect();

        // Sorted positions put pruned ones in a prefix and out-of-range ones in a suffix, so
        // validation trims the batch instead of poisoning a blob group that also holds valid
        // positions.
        let start = positions.partition_point(|&pos| pos < self.bounds.start);
        let end = positions.partition_point(|&pos| pos < self.bounds.end);
        let valid = &positions[start..end];
        if valid.is_empty() {
            return out;
        }

        // Serve the probe from the per-thread scratch buffer. Stale bytes from a previous probe
        // are harmless: slots the cache cannot serve are reported as misses and never decoded.
        let items_per_blob = self.items_per_blob.get();
        let mut scratch =
            Cached::take(&PROBE_SCRATCH, || Ok::<_, ()>(Vec::new()), |_| Ok(())).unwrap();
        let need = valid.len() * A::SIZE;
        if scratch.len() < need {
            scratch.resize(need, 0);
        }
        let buf = &mut scratch[..need];
        let mut hits = 0u64;
        let mut group_base = start;
        for group in valid.chunk_by(|a, b| {
            super::position_to_blob(*a, items_per_blob)
                == super::position_to_blob(*b, items_per_blob)
        }) {
            let base = group_base;
            group_base += group.len();
            let Ok((blob_num, blob_offsets)) = self.locate_group(group) else {
                continue;
            };
            let Some(blob) = self.blobs.get(blob_num) else {
                continue;
            };
            let buf = &mut buf[..group.len() * A::SIZE];
            let misses =
                blob.try_read_many_sync_into(buf, &blob_offsets, Inner::<E, A>::CHUNK_SIZE);
            let mut misses = misses.into_iter().peekable();
            #[allow(unknown_lints, clippy::chunks_exact_to_as_chunks)]
            for (idx, slice) in buf.chunks_exact(A::SIZE).enumerate() {
                if misses.peek() == Some(&idx) {
                    misses.next();
                    continue;
                }
                // A decode failure declines to a miss: the async completion re-reads the
                // item and bubbles the failure as [Error::Codec], like every async read path.
                if let Ok(item) = A::decode(Copying(slice)) {
                    out[base + idx] = Some(item);
                    hits += 1;
                }
            }
        }
        self.metrics.cache_hits.inc_by(hits);
        self.metrics.items_read.inc_by(hits);
        out
    }
}

impl<E: Context, A: CodecFixedShared> super::Contiguous for Reader<'_, E, A> {
    type Item = A;

    fn bounds(&self) -> Range<u64> {
        self.bounds.clone()
    }

    async fn read(&self, pos: u64) -> Result<A, Error> {
        self.metrics.read_calls.inc();

        // Serve from the page cache synchronously when possible, avoiding the async storage path.
        if let Some(item) = self.try_read_sync(pos) {
            return Ok(item);
        }
        self.read_miss(pos).await
    }

    async fn read_many(&self, positions: &[u64]) -> Result<Vec<A>, Error> {
        if positions.is_empty() {
            return Ok(Vec::new());
        }
        let _timer = self.metrics.read_many_timer();
        self.metrics.read_many_calls.inc();
        self.read_many_inner(positions).await
    }

    fn try_read_sync(&self, pos: u64) -> Option<A> {
        let (blob, offset) = self.locate(pos).ok()?;
        let mut scratch =
            Cached::take(&PROBE_SCRATCH, || Ok::<_, ()>(Vec::new()), |_| Ok(())).unwrap();
        if scratch.len() < A::SIZE {
            scratch.resize(A::SIZE, 0);
        }
        let buf = &mut scratch[..A::SIZE];
        let item = if blob.try_read_sync_into(buf, offset) {
            A::decode(Copying(buf)).ok()
        } else {
            None
        };
        if item.is_some() {
            self.metrics.cache_hits.inc();
            self.metrics.items_read.inc();
        }
        item
    }

    fn try_read_many_sync(&self, positions: &[u64]) -> Vec<Option<A>> {
        self.probe_items(positions)
    }

    async fn replay(
        &self,
        range: Range<u64>,
        buffer: NonZeroUsize,
        read_options: ReadOptions,
    ) -> Result<impl Stream<Item = Result<Batch<A>, Error>> + Send, Error> {
        replay_stream(
            &self.blobs,
            self.bounds.clone(),
            self.items_per_blob,
            range,
            buffer,
            read_options,
        )
    }
}

impl<E: Context, A: CodecFixedShared> super::Contiguous for Journal<E, A> {
    type Item = A;

    fn bounds(&self) -> Range<u64> {
        self.0.bounds.clone()
    }

    async fn read(&self, pos: u64) -> Result<A, Error> {
        self.0.reader().read(pos).await
    }

    async fn read_many(&self, positions: &[u64]) -> Result<Vec<A>, Error> {
        self.0.reader().read_many(positions).await
    }

    fn try_read_sync(&self, pos: u64) -> Option<A> {
        self.0.reader().try_read_sync(pos)
    }

    fn try_read_many_sync(&self, positions: &[u64]) -> Vec<Option<A>> {
        self.0.reader().probe_items(positions)
    }

    async fn replay(
        &self,
        range: Range<u64>,
        buffer: NonZeroUsize,
        read_options: ReadOptions,
    ) -> Result<impl Stream<Item = Result<Batch<A>, Error>> + Send, Error> {
        let blobs = self.0.blobs.reader();
        replay_stream(
            &blobs,
            self.0.bounds.clone(),
            self.0.items_per_blob,
            range,
            buffer,
            read_options,
        )
    }
}

impl<E: Context, A: CodecFixedShared> Mutable for Journal<E, A> {
    async fn append(self, item: &Self::Item) -> Result<(Self, u64), Error> {
        Self::append(self, item).await
    }

    async fn append_many(self, items: Many<'_, Self::Item>) -> Result<(Self, u64), Error> {
        Self::append_many(self, items).await
    }

    async fn prune(self, min_position: u64) -> Result<(Self, bool), Error> {
        Self::prune(self, min_position).await
    }

    async fn start_sync(self) -> Result<(Self, Handle<()>), Error> {
        Self::start_sync(self).await
    }

    async fn commit(self) -> Result<Self, Error> {
        Self::commit(self).await
    }

    async fn sync(self) -> Result<Self, Error> {
        Self::sync(self).await
    }

    async fn destroy(self) -> Result<(), Error> {
        Self::destroy(self).await
    }
}

#[commonware_macros::stability(ALPHA)]
impl<E: Context, A: CodecFixedShared> authenticated::BackingRecovery for Recovery<E, A> {
    type Journal = Journal<E, A>;

    fn bounds(&self) -> Range<u64> {
        self.bounds.clone()
    }

    async fn read(&self, pos: u64) -> Result<A, Error> {
        Self::item(self, pos).await
    }

    async fn reset(self, size: u64) -> Result<Self, Error> {
        Ok(*Box::new(self).clear_to_size(size).await?)
    }

    async fn finish(self, size: u64) -> Result<Self::Journal, Error> {
        Journal(Box::new(Self::publish(self, size).await?))
            .commit()
            .await
    }
}

#[commonware_macros::stability(ALPHA)]
impl<E: Context, A: CodecFixedShared> authenticated::Backing<E> for Journal<E, A> {
    type Recovery = Recovery<E, A>;

    async fn recover(
        context: E,
        cfg: Self::Config,
        max_size: Option<u64>,
    ) -> Result<Self::Recovery, Error> {
        let checkpoint = Checkpoint::open(context.child("meta"), &cfg.partition).await?;
        Recovery::open(context, cfg, checkpoint, max_size).await
    }

    async fn clear(context: E, cfg: Self::Config, size: u64) -> Result<Self::Recovery, Error> {
        let checkpoint = Checkpoint::open(context.child("meta"), &cfg.partition).await?;

        // A staged clear already owns both blob partitions. Otherwise fail before writing intent
        // if they are inconsistent.
        if checkpoint.clear_target().is_none() {
            Partition::select(&context, &cfg.partition).await?;
        }
        Recovery::open_cleared(context, cfg, checkpoint, size, || async { Ok(()) }).await
    }

    async fn span(context: E, cfg: &Self::Config) -> Result<Range<u64>, Error> {
        let checkpoint = Checkpoint::open(context.child("meta"), &cfg.partition).await?;
        match checkpoint.clear_target() {
            Some(target) => Ok(target..target),
            None => Recovery::<E, A>::span(&context, cfg, &checkpoint).await,
        }
    }

    type Config = Config;
}

#[cfg(test)]
impl<E: crate::Context, A: CodecFixedShared> Inner<E, A> {
    /// Return the recovery watermark.
    pub(super) fn recovery_watermark(&self) -> u64 {
        self.checkpoint
            .watermark()
            .expect("recovery watermark must exist after init")
    }

    /// Sync, then reopen a shorter prefix for recovery/fault fixtures.
    pub(super) async fn test_truncate(self: Box<Self>, cap: u64) -> Result<Box<Self>, Error> {
        self.sync().await?.test_reopen_at_most(cap).await
    }

    /// Reopen a shorter prefix without syncing first, releasing the previous owner. Faults
    /// armed by the caller therefore reach recovery rather than the closing sync.
    pub(super) async fn test_reopen_at_most(self: Box<Self>, cap: u64) -> Result<Box<Self>, Error> {
        let (context, name, page_cache, write_buffer) = self.blobs.test_configuration();
        let cfg = Config {
            partition: name.strip_suffix("-blobs").unwrap().to_owned(),
            items_per_blob: self.items_per_blob,
            page_cache,
            write_buffer,
            replay_buffer: write_buffer,
        };
        drop(self);
        let checkpoint = Checkpoint::open(context.child("meta"), &cfg.partition).await?;
        let recovery = Recovery::<E, A>::open(context, cfg, checkpoint, Some(cap)).await?;
        Ok(Box::new(recovery.publish(cap).await?))
    }

    /// Test helper: Get the oldest blob from the blob store.
    pub(crate) const fn test_oldest_blob(&self) -> Option<u64> {
        Some(self.blobs.oldest_blob_index())
    }

    /// Test helper: Get the newest blob from the blob store.
    pub(crate) fn test_newest_blob(&self) -> Option<u64> {
        Some(self.blobs.tail_blob_index())
    }

    /// Test helper: Make one blob durable (sealed history or the tail).
    pub(crate) async fn test_sync_blob(mut self: Box<Self>, blob: u64) -> Result<Box<Self>, Error> {
        self.blobs = self.blobs.sync_blob(blob).await?;
        Ok(self)
    }

    /// Test helper: Set and persist the recovery watermark directly.
    pub(crate) async fn test_set_recovery_watermark(
        mut self: Box<Self>,
        watermark: u64,
    ) -> Result<Box<Self>, Error> {
        self.checkpoint.set_watermark(Some(watermark));
        self.checkpoint = self.checkpoint.sync().await?;
        Ok(self)
    }

    /// Test helper: Durably stage a clear intent in the journal's checkpoint.
    pub(crate) async fn test_stage_clear(
        context: E,
        partition: &str,
        target: u64,
    ) -> Result<(), Error> {
        let checkpoint = Checkpoint::open(context, partition).await?;
        checkpoint.stage_clear(target).await?;
        Ok(())
    }
}

#[cfg(test)]
impl<E: crate::Context, A: CodecFixedShared> Journal<E, A> {
    /// Read the recovery watermark persisted for `partition` without opening the journal.
    pub(crate) async fn persisted_watermark(
        context: E,
        partition: &str,
    ) -> Result<Option<u64>, Error> {
        Ok(Checkpoint::open(context, partition).await?.watermark())
    }

    /// Test helper: Get the oldest blob from the blob store.
    pub(crate) fn test_oldest_blob(&self) -> Option<u64> {
        self.0.test_oldest_blob()
    }

    /// Test helper: Get the newest blob from the blob store.
    pub(crate) fn test_newest_blob(&self) -> Option<u64> {
        self.0.test_newest_blob()
    }

    /// Test helper: Make one blob durable (sealed history or the tail).
    pub(crate) async fn test_sync_blob(mut self, blob: u64) -> Result<Self, Error> {
        self.0 = self.0.test_sync_blob(blob).await?;
        Ok(self)
    }

    /// Test helper: Set and persist the recovery watermark directly.
    pub(crate) async fn test_set_recovery_watermark(
        mut self,
        watermark: u64,
    ) -> Result<Self, Error> {
        self.0 = self.0.test_set_recovery_watermark(watermark).await?;
        Ok(self)
    }

    /// Test helper: Durably stage a clear intent in the journal's checkpoint.
    pub(crate) async fn test_stage_clear(
        context: E,
        partition: &str,
        target: u64,
    ) -> Result<(), Error> {
        Inner::<E, A>::test_stage_clear(context, partition, target).await
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{
        journal::{
            authenticated,
            contiguous::{
                Contiguous as _,
                tests::{replay_items, tear_page_rewrite},
            },
        },
        utils::codec::View,
    };
    use commonware_codec::FixedSize;
    use commonware_cryptography::{Hasher as _, Sha256, sha256::Digest};
    use commonware_macros::test_traced;
    use commonware_runtime::{
        Blob, Error as RuntimeError, Metrics as _, Runner, Spawner as _, Storage, Supervisor as _,
        WriteOptions,
        buffer::paged::{Writer, corrupt_page},
        deterministic::{self, Context},
        mocks::{
            DelayedSyncContext, PendingSyncs, RecordingContext, WriteFaultContext, WriteFaults,
            drive_pending_syncs, fail_pending_syncs, next_pending_sync, release_pending_syncs,
        },
    };
    use commonware_utils::{NZU16, NZU64, NZUsize, probability};
    use futures::{StreamExt, pin_mut};
    use std::num::NonZeroU16;

    const PAGE_SIZE: NonZeroU16 = NZU16!(44);
    const PAGE_CACHE_SIZE: NonZeroUsize = NZUsize!(3);

    /// Generate a SHA-256 digest for the given value.
    fn test_digest(value: u64) -> Digest {
        Sha256::hash(&[&value.to_be_bytes()])
    }

    fn test_cfg(items_per_blob: NonZeroU64) -> Config {
        Config {
            partition: "test-partition".into(),
            items_per_blob,
            page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
            write_buffer: NZUsize!(2048),
            replay_buffer: NZUsize!(2048),
        }
    }

    fn blob_partition(cfg: &Config) -> String {
        format!("{}-blobs", cfg.partition)
    }

    #[test]
    fn test_unbounded_fixed_repairs_hole_after_full_capacity() {
        deterministic::Runner::default().start(|context| async move {
            let cache = CacheRef::new(NZU16!(8), NZUsize!(8));
            let cfg = Config {
                partition: "initialization-extra-tail".into(),
                items_per_blob: NZU64!(2),
                page_cache: cache.clone(),
                write_buffer: NZUsize!(128),
                replay_buffer: NZUsize!(128),
            };
            let (blob, size) = context
                .open("initialization-extra-tail-blobs", &0u64.to_be_bytes())
                .await
                .unwrap();
            let mut writer = Writer::new(blob, size, 128, cache).await.unwrap();
            (writer, _) = writer.append(&[1; 32]).await.unwrap();
            writer = writer.sync().await.unwrap();
            drop(writer);
            corrupt_page(
                &context,
                "initialization-extra-tail-blobs",
                &0u64.to_be_bytes(),
                2,
                8,
            )
            .await;
            let result = Journal::<_, u64>::init(context.child("open"), cfg).await;
            assert!(
                result.is_ok(),
                "unbounded recovery must trim a hole after capacity: {result:?}"
            );

            let journal = result.unwrap();
            assert_eq!(journal.bounds(), 0..2);
            let item = u64::from_be_bytes([1; 8]);
            assert_eq!(journal.read(0).await.unwrap(), item);
            assert_eq!(journal.read(1).await.unwrap(), item);
        });
    }

    #[test]
    fn test_fixed_bounded_recovery_read_count() {
        for count in [257u64, 4097] {
            for capacity in [10_000, u64::MAX] {
                deterministic::Runner::default().start(|context| async move {
                    let (context, recordings) = RecordingContext::new(context);
                    let mut cfg = test_cfg(capacity.try_into().unwrap());
                    cfg.page_cache = CacheRef::new(NZU16!(256), NZUsize!(3));
                    cfg.replay_buffer = NZUsize!(1);
                    let mut journal = Journal::<_, u64>::init(context.child("seed"), cfg.clone())
                        .await
                        .unwrap();
                    for value in 0..count {
                        (journal, _) = journal.append(&value).await.unwrap();
                    }
                    drop(journal.sync().await.unwrap());
                    for cap in [
                        None,
                        Some(33),
                        Some(32),
                        Some(count),
                        Some(count + 1),
                        Some(u64::MAX),
                    ] {
                        let checkpoint =
                            Checkpoint::open(context.child("checkpoint"), &cfg.partition)
                                .await
                                .unwrap();
                        assert_eq!(checkpoint.watermark(), Some(count));
                        recordings.clear();
                        let recovery = Recovery::<_, u64>::open(
                            context.child("recovery"),
                            cfg.clone(),
                            checkpoint,
                            cap,
                        )
                        .await
                        .unwrap();
                        assert_eq!(recovery.size(), count.min(cap.unwrap_or(count)));
                        assert_eq!(recovery.pending.get(&0).unwrap().size(), count * 8);
                        let expected = if cap == Some(32) { 1 } else { 2 };
                        assert_eq!(
                            recordings.snapshot().reads.len(),
                            expected,
                            "count={count}, capacity={capacity}, cap={cap:?}"
                        );
                    }
                });
            }
        }
    }

    #[test]
    fn test_fixed_recovery_large_capacity() {
        for count in [0, 1] {
            for cap in [
                None,
                Some(0),
                Some(1),
                Some(u64::MAX / 8 + 1),
                Some(u64::MAX),
            ] {
                deterministic::Runner::default().start(|context| async move {
                    let cfg = test_cfg(NZU64!(u64::MAX));
                    let mut journal = Journal::<_, u64>::init(context.child("seed"), cfg.clone())
                        .await
                        .unwrap();
                    for value in 0..count {
                        (journal, _) = journal.append(&value).await.unwrap();
                    }
                    drop(journal.sync().await.unwrap());
                    let journal = match cap {
                        None => Journal::<_, u64>::init(context.child("reopen"), cfg).await,
                        Some(cap) => {
                            Journal::<_, u64>::init_at_most(context.child("reopen"), cfg, cap).await
                        }
                    }
                    .unwrap();
                    let retained = count.min(cap.unwrap_or(u64::MAX));
                    assert_eq!(journal.bounds(), 0..retained);
                    if retained != 0 {
                        assert_eq!(journal.read(0).await.unwrap(), 0);
                    }
                });
            }
        }
    }

    #[test_traced]
    fn test_fixed_append_buffer_fallback_rotation_and_snapshot() {
        for write_buffer in [NZUsize!(44), NZUsize!(2048)] {
            deterministic::Runner::default().start(|context| async move {
                let mut cfg = test_cfg(NZU64!(7));
                cfg.write_buffer = write_buffer;
                let mut journal = Journal::<_, Digest>::init(context.child("initial"), cfg.clone())
                    .await
                    .unwrap();
                for i in 0..3 {
                    let position;
                    (journal, position) = journal.append(&test_digest(i)).await.unwrap();
                    assert_eq!(position, i);
                }
                let snapshot;
                (journal, snapshot) = journal.snapshot().await.unwrap();

                // Mix direct and prepared appends across buffer and blob boundaries.
                for i in 3..20 {
                    let position;
                    if i % 3 == 0 {
                        let item = test_digest(i);
                        let prepared = journal.prepare_append(Many::Flat(&[item]));
                        (journal, position) = journal.append_prepared(prepared).await.unwrap();
                    } else {
                        (journal, position) = journal.append(&test_digest(i)).await.unwrap();
                    }
                    assert_eq!(position, i);
                    assert_eq!(journal.size(), i + 1);
                }
                for i in 0..3 {
                    assert_eq!(snapshot.read(i).await.unwrap(), test_digest(i));
                }
                journal = journal.sync().await.unwrap();
                drop(snapshot);
                drop(journal);

                let journal = Journal::<_, Digest>::init(context, cfg).await.unwrap();
                assert_eq!(journal.size(), 20);
                for i in 0..20 {
                    assert_eq!(journal.read(i).await.unwrap(), test_digest(i));
                }
                journal.destroy().await.unwrap();
            });
        }
    }

    #[test_traced]
    fn test_fixed_cached_read_preserves_owned_byte_fields() {
        deterministic::Runner::default().start(|context| async move {
            let cfg = test_cfg(NZU64!(3));
            let mut journal = Journal::init(context, cfg).await.unwrap();
            (journal, _) = journal.append(&View::new(7)).await.unwrap();
            let decoded = journal.try_read_sync(0).unwrap();
            (journal, _) = journal.append(&View::new(8)).await.unwrap();
            let next = journal.try_read_sync(1).unwrap();
            journal.destroy().await.unwrap();
            assert_eq!(decoded.bytes.as_ref(), &7u64.to_be_bytes());
            assert_eq!(next.bytes.as_ref(), &8u64.to_be_bytes());
        });
    }

    #[test_traced]
    fn test_fixed_batch_read_preserves_owned_byte_fields() {
        deterministic::Runner::default().start(|context| async move {
            let cfg = test_cfg(NZU64!(3));
            let mut journal = Journal::init(context, cfg).await.unwrap();
            for i in 0..5 {
                (journal, _) = journal.append(&View::new(i)).await.unwrap();
            }
            let decoded = journal.read_many(&[0, 2, 3, 4]).await.unwrap();
            let probed = journal.try_read_many_sync(&[0, 2, 3, 4]);
            journal.destroy().await.unwrap();
            for (value, expected) in decoded.iter().zip([0u64, 2, 3, 4]) {
                assert_eq!(value.bytes.as_ref(), &expected.to_be_bytes());
                value.assert_shared();
            }
            for (value, expected) in probed.iter().zip([0u64, 2, 3, 4]) {
                let value = value.as_ref().expect("probe should hit after read_many");
                assert_eq!(value.bytes.as_ref(), &expected.to_be_bytes());
            }
        });
    }

    #[test]
    fn test_bounded_empty_prefix_at_maximum_position() {
        deterministic::Runner::default().start(|context| async move {
            let cfg = test_cfg(NZU64!(1));
            let mut journal =
                Journal::<_, u64>::init_at_size(context.child("seed"), cfg.clone(), u64::MAX - 1)
                    .await
                    .unwrap();
            (journal, _) = journal.append(&99).await.unwrap();
            (journal, _) = journal.prune(u64::MAX).await.unwrap();
            _ = journal.sync().await.unwrap();
            let journal =
                Journal::<_, u64>::init_at_most(context.child("capped"), cfg.clone(), u64::MAX)
                    .await
                    .unwrap();
            assert_eq!(journal.bounds(), u64::MAX..u64::MAX);
            drop(journal);
            let journal = Journal::<_, u64>::init(context.child("ordinary"), cfg)
                .await
                .unwrap();
            assert_eq!(journal.bounds(), u64::MAX..u64::MAX);
        });
    }

    #[test]
    fn test_bounded_empty_prefix_interrupted_repair() {
        deterministic::Runner::default().start(|context| async move {
            let cfg = test_cfg(NZU64!(5));
            let mut journal =
                Journal::<_, Digest>::init_at_size(context.child("seed"), cfg.clone(), 7)
                    .await
                    .unwrap();
            for i in 7..13 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            _ = journal.sync().await.unwrap();
            let checkpoint = Checkpoint::open(context.child("checkpoint"), &cfg.partition)
                .await
                .unwrap();
            let recovery =
                Recovery::<_, Digest>::open(context.child("cap"), cfg.clone(), checkpoint, Some(7))
                    .await
                    .unwrap();

            // Interrupt initialization after durable repair and before publication.
            drop(recovery.repair_to(7).await.unwrap());
            let journal = Journal::<_, Digest>::init(context.child("retry"), cfg)
                .await
                .unwrap();
            assert_eq!(journal.bounds(), 7..7);
        });
    }

    #[test]
    fn test_recovery_rebuilds_pending_gap() {
        deterministic::Runner::default().start(|context| async move {
            // The two-page write buffer holds two digests, so the third digest in each blob takes
            // the owned fallback during recovery.
            let mut cfg = test_cfg(NZU64!(5));
            cfg.write_buffer = NZUsize!(88);
            let mut journal = Journal::<_, Digest>::init(context.child("seed"), cfg.clone())
                .await
                .unwrap();
            for i in 0..13 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            for blob in 0..3 {
                journal = journal.test_sync_blob(blob).await.unwrap();
            }
            drop(journal);

            // Lose an unacknowledged index prefix while newer index blobs survive.
            let (blob, _) = context
                .open(&blob_partition(&cfg), &0u64.to_be_bytes())
                .await
                .unwrap();
            blob.resize(0).await.unwrap();
            blob.sync().await.unwrap();
            drop(blob);
            let checkpoint = Checkpoint::open(context.child("checkpoint"), &cfg.partition)
                .await
                .unwrap();
            let mut recovery = Box::new(
                Recovery::<_, Digest>::open(context.child("repair"), cfg.clone(), checkpoint, None)
                    .await
                    .unwrap(),
            );
            assert_eq!(recovery.size(), 0);
            assert_eq!(recovery.recovery_watermark(), 0);
            recovery = recovery.truncate(0).await.unwrap();

            // Rebuild with values the surviving blobs never held so stale pages cannot satisfy
            // the reads below.
            for i in 0..13 {
                recovery = recovery.append(&test_digest(100 + i)).await.unwrap();
            }
            let journal = recovery.publish(13).await.unwrap();
            _ = Box::new(journal).sync().await.unwrap();
            let journal = Journal::<_, Digest>::init(context.child("retry"), cfg)
                .await
                .unwrap();
            assert_eq!(journal.bounds(), 0..13);
            for value in 0..13 {
                assert_eq!(journal.read(value).await.unwrap(), test_digest(100 + value));
            }
        });
    }

    /// A bounded publication raises the watermark to the retained end, so bytes rebuilt into
    /// the tail during recovery must be durable before that raise.
    #[test]
    fn test_bounded_publish_syncs_rebuilt_tail_before_watermark_raise() {
        let ((), checkpoint) =
            deterministic::Runner::default().start_and_recover(|context| async move {
                let cfg = test_cfg(NZU64!(5));
                let checkpoint = Checkpoint::open(context.child("checkpoint"), &cfg.partition)
                    .await
                    .unwrap();
                let mut recovery = Box::new(
                    Recovery::<_, Digest>::open(
                        context.child("rebuild"),
                        cfg.clone(),
                        checkpoint,
                        Some(7),
                    )
                    .await
                    .unwrap(),
                );
                recovery = recovery.truncate(0).await.unwrap();
                for i in 0..7 {
                    recovery = recovery.append(&test_digest(i)).await.unwrap();
                }
                let journal = recovery.publish(7).await.unwrap();
                assert_eq!(journal.size(), 7);
                assert_eq!(journal.recovery_watermark(), 7);
                drop(journal);
            });
        deterministic::Runner::from(checkpoint).start(|context| async move {
            let cfg = test_cfg(NZU64!(5));
            let journal = Journal::<_, Digest>::init(context.child("reopen"), cfg)
                .await
                .unwrap();
            assert_eq!(journal.bounds(), 0..7);
            for i in 0..7 {
                assert_eq!(journal.read(i).await.unwrap(), test_digest(i));
            }
        });
    }

    #[test]
    fn test_bounded_initialization_retries_after_storage_faults() {
        for start in [0, 7] {
            for kind in 0..3 {
                for numerator in [1, 3, 7, 10] {
                    let (succeeded, checkpoint) = deterministic::Runner::default()
                        .start_and_recover(|context| async move {
                            let cfg = test_cfg(NZU64!(5));
                            let mut journal = Journal::<_, u64>::init_at_size(
                                context.child("seed"),
                                cfg.clone(),
                                start,
                            )
                            .await
                            .unwrap();
                            for value in start..13 {
                                (journal, _) = journal.append(&value).await.unwrap();
                            }
                            _ = journal.sync().await.unwrap();
                            let rate = commonware_utils::Probability::new(numerator, 10).unwrap();
                            *context.storage_fault_config().write() = deterministic::FaultConfig {
                                sync_rate: (kind == 0).then_some(rate),
                                remove_rate: (kind == 1).then_some(rate),
                                resize_rate: (kind == 2).then_some(deterministic::ResizeConfig {
                                    failure_rate: rate,
                                    partial_rate: rate,
                                }),
                                ..Default::default()
                            };

                            // The failed initializer is consumed. Inspect only freshly opened
                            // storage.
                            Journal::<_, u64>::init_at_most(context.child("faulted"), cfg, 7)
                                .await
                                .is_ok()
                        });
                    deterministic::Runner::from(checkpoint).start(|context| async move {
                        *context.storage_fault_config().write() =
                            deterministic::FaultConfig::default();
                        let cfg = test_cfg(NZU64!(5));
                        if succeeded {
                            let journal =
                                Journal::<_, u64>::init(context.child("ordinary"), cfg.clone())
                                    .await
                                    .unwrap();
                            assert_eq!(journal.bounds(), start..7);
                            drop(journal);
                        }
                        let journal =
                            Journal::<_, u64>::init_at_most(context.child("retry"), cfg.clone(), 7)
                                .await
                                .unwrap();
                        assert_eq!(journal.bounds(), start..7);
                        for value in start..7 {
                            assert_eq!(journal.read(value).await.unwrap(), value);
                        }
                        drop(journal);
                        let mut journal =
                            Journal::<_, u64>::init(context.child("restart"), cfg.clone())
                                .await
                                .unwrap();
                        assert_eq!(journal.bounds(), start..7);
                        (journal, _) = journal.append(&99).await.unwrap();
                        _ = journal.sync().await.unwrap();
                        let journal = Journal::<_, u64>::init(context.child("verify"), cfg)
                            .await
                            .unwrap();
                        assert_eq!(journal.bounds(), start..8);
                        assert_eq!(journal.read(7).await.unwrap(), 99);
                    });
                }
            }
        }
    }

    #[test]
    fn test_init_at_most_restart_and_append() {
        for start in [0, 7] {
            for cap in [0, 1, 5, 7, 8, 10, 13, 20, u64::MAX] {
                deterministic::Runner::default().start(|context| async move {
                    let cfg = test_cfg(NZU64!(5));
                    let mut journal = Journal::<_, u64>::init_at_size(
                        context.child("storage"),
                        cfg.clone(),
                        start,
                    )
                    .await
                    .unwrap();
                    for value in start..13 {
                        (journal, _) = journal.append(&value).await.unwrap();
                    }
                    journal = journal.sync().await.unwrap();
                    drop(journal);

                    let opened =
                        Journal::<_, u64>::init_at_most(context.child("storage"), cfg.clone(), cap)
                            .await;
                    if cap < start {
                        assert!(matches!(opened, Err(Error::ItemPruned(value)) if value == cap));
                        let journal = Journal::<_, u64>::init(context.child("storage"), cfg)
                            .await
                            .unwrap();
                        assert_eq!(journal.bounds(), start..13);
                        return;
                    }
                    let end = cap.min(13);
                    let journal = opened.unwrap();
                    assert_eq!(journal.bounds(), start..end);
                    for value in start..end {
                        assert_eq!(journal.read(value).await.unwrap(), value);
                    }
                    drop(journal);

                    // Opening alone must durably establish the selected prefix.
                    let mut journal =
                        Journal::<_, u64>::init(context.child("storage"), cfg.clone())
                            .await
                            .unwrap();
                    assert_eq!(journal.bounds(), start..end);
                    for value in end..end + 8 {
                        let pos;
                        (journal, pos) = journal.append(&(value + 100)).await.unwrap();
                        assert_eq!(pos, value);
                    }
                    journal = journal.sync().await.unwrap();
                    drop(journal);
                    let journal = Journal::<_, u64>::init(context.child("storage"), cfg)
                        .await
                        .unwrap();
                    for value in start..end + 8 {
                        let expected = if value < end { value } else { value + 100 };
                        assert_eq!(journal.read(value).await.unwrap(), expected);
                    }
                });
            }
        }
    }

    #[test]
    fn test_init_at_most_skips_discarded_blobs() {
        deterministic::Runner::default().start(|context| async move {
            let cfg = test_cfg(NZU64!(5));
            let mut journal = Journal::<_, u64>::init(context.child("storage"), cfg.clone())
                .await
                .unwrap();
            for value in 0..23 {
                (journal, _) = journal.append(&value).await.unwrap();
            }
            journal = journal.sync().await.unwrap();
            drop(journal);
            let (blob, _) = context
                .open(&blob_partition(&cfg), &3u64.to_be_bytes())
                .await
                .unwrap();
            blob.resize(1).await.unwrap();
            blob.sync().await.unwrap();
            drop(blob);
            let journal = Journal::<_, u64>::init_at_most(context.child("storage"), cfg.clone(), 7)
                .await
                .unwrap();
            assert_eq!(journal.bounds(), 0..7);
            drop(journal);
            let journal = Journal::<_, u64>::init(context.child("storage"), cfg)
                .await
                .unwrap();
            assert_eq!(journal.bounds(), 0..7);
        });
    }

    #[test]
    fn test_start_sync_keeps_predecessor_sync() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(3));
            let mut journal = Box::new(Inner::<_, u64>::init(context, cfg).await.unwrap());

            // Rollover starts a predecessor sync.
            (journal, _) = journal
                .append_many(Many::Flat(&[1, 2, 3, 4]))
                .await
                .unwrap();
            assert!(journal.blobs.has_tail_predecessor_sync());

            // Handle includes predecessor; slot drains later.
            let (journal, handle) = journal.start_sync().await.unwrap();
            assert!(journal.blobs.has_tail_predecessor_sync());
            handle.await.unwrap();
            assert!(journal.blobs.has_tail_predecessor_sync());

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

    #[test_traced]
    fn test_start_sync_advances_watermark_lagged() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let pending = PendingSyncs::default();
            let cfg = test_cfg(NZU64!(100));
            let make = |pending: PendingSyncs| {
                Inner::<_, u64>::init(
                    DelayedSyncContext {
                        inner: context.child("journal"),
                        pending,
                    },
                    cfg.clone(),
                )
            };
            let mut journal = Box::new(make(pending.clone()).await.unwrap());

            // Nothing proven while the first sync is parked: the watermark must not move.
            (journal, _) = journal.append_many(Many::Flat(&[1, 2, 3])).await.unwrap();
            let (mut journal, h1) = journal.start_sync().await.unwrap();
            assert_eq!(journal.recovery_watermark(), 0);

            release_pending_syncs(&pending);
            h1.await.unwrap();

            // The first sync is proven, so the next call advances the watermark to its size,
            // one interval behind the tip.
            (journal, _) = journal.append(&4).await.unwrap();
            let (journal, h2) = journal.start_sync().await.unwrap();
            assert_eq!(journal.recovery_watermark(), 3);
            drive_pending_syncs(&pending, h2).await.unwrap();

            // A third call catches the watermark up to the second sync's size.
            let (journal, h3) = drive_pending_syncs(&pending, journal.start_sync())
                .await
                .unwrap();
            assert_eq!(journal.recovery_watermark(), 4);
            drive_pending_syncs(&pending, h3).await.unwrap();

            // The advanced watermark is durable: a reopen resumes from it.
            pending.unblock();
            drop(journal);
            let journal = make(pending.clone()).await.unwrap();
            assert_eq!(journal.recovery_watermark(), 4);
            assert_eq!(journal.bounds, 0..4);
            journal.destroy().await.unwrap();
        });
    }

    /// A failed sync never advances the watermark. A failed sealed-blob sync leaves the next
    /// `start_sync` successful, and a failed tail sync fails it.
    #[test_traced]
    fn test_start_sync_failure_blocks_watermark() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let pending = PendingSyncs::default();
            let cfg = test_cfg(NZU64!(3));
            let mut journal = Box::new(
                Inner::<_, u64>::init(
                    DelayedSyncContext {
                        inner: context.child("journal"),
                        pending: pending.clone(),
                    },
                    cfg,
                )
                .await
                .unwrap(),
            );

            // Filling the first blob seals it and starts its sync. The next start_sync syncs
            // the fresh tail.
            (journal, _) = journal.append_many(Many::Flat(&[1, 2, 3])).await.unwrap();
            let (mut journal, h1) = journal.start_sync().await.unwrap();

            // Fail the sealed blob's sync and let the tail sync land.
            next_pending_sync(&pending)
                .release
                .send(Err(RuntimeError::Io(
                    std::io::Error::other("injected sync failure").into(),
                )))
                .unwrap();
            release_pending_syncs(&pending);
            assert!(h1.await.is_err());

            // The failed sync proves nothing. The next call succeeds because the tail sync
            // landed, but it must not advance the watermark past the lost blob.
            assert_eq!(journal.barrier.boundary(), 0);
            let (mut journal, h2) = journal.start_sync().await.unwrap();
            assert_eq!(journal.recovery_watermark(), 0);
            assert!(h2.await.is_err());

            // A failed tail sync fails the next call.
            (journal, _) = journal.append(&4).await.unwrap();
            let (journal, _) = journal.start_sync().await.unwrap();
            fail_pending_syncs(&pending);
            assert!(journal.start_sync().await.is_err());
        });
    }

    #[test_traced]
    fn test_truncate_drains_parked_watermark_advance() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let pending = PendingSyncs::default();
            let cfg = test_cfg(NZU64!(100));
            let make = |pending: PendingSyncs| {
                Inner::<_, u64>::init(
                    DelayedSyncContext {
                        inner: context.child("journal"),
                        pending,
                    },
                    cfg.clone(),
                )
            };
            let mut journal = Box::new(make(pending.clone()).await.unwrap());

            (journal, _) = journal.append_many(Many::Flat(&[1, 2, 3])).await.unwrap();
            let (mut journal, h1) = journal.start_sync().await.unwrap();
            release_pending_syncs(&pending);
            h1.await.unwrap();

            // This parks the metadata sync advancing the watermark to 3.
            (journal, _) = journal.append(&4).await.unwrap();
            let (journal, h2) = journal.start_sync().await.unwrap();
            assert_eq!(journal.recovery_watermark(), 3);

            // Truncate below the in-flight advance. The lowered value must win on reopen.
            let journal = drive_pending_syncs(&pending, journal.test_truncate(2))
                .await
                .unwrap();
            assert_eq!(journal.recovery_watermark(), 2);
            drop(h2);

            // Reopen: the lowered watermark held, and no corruption is reported.
            pending.unblock();
            drop(journal);
            let journal = make(pending.clone()).await.unwrap();
            assert_eq!(journal.recovery_watermark(), 2);
            assert_eq!(journal.bounds, 0..2);
            journal.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_start_sync_watermark_advance_inline_failure() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let faults = WriteFaults::default();
            let cfg = test_cfg(NZU64!(100));
            let make = |faults: WriteFaults| {
                Inner::<_, u64>::init(
                    WriteFaultContext {
                        inner: context.child("journal"),
                        faults,
                    },
                    cfg.clone(),
                )
            };
            let mut journal = Box::new(make(faults.clone()).await.unwrap());

            // Prove three items durable (watermark 3), then one more so the next call has an
            // advance to start.
            (journal, _) = journal.append_many(Many::Flat(&[1, 2, 3])).await.unwrap();
            let (mut journal, h1) = journal.start_sync().await.unwrap();
            h1.await.unwrap();
            (journal, _) = journal.append(&4).await.unwrap();
            let journal = journal.commit().await.unwrap();

            // The advance's inline metadata writes fail: the call fails, consuming the
            // journal, even though the data is durable.
            faults.arm();
            assert!(journal.start_sync().await.is_err());
            faults.disarm();

            // The failed advance never compromised the data: a reopen recovers it all and a
            // fresh sync succeeds.
            let journal = Box::new(make(faults).await.unwrap());
            assert_eq!(journal.bounds, 0..4);
            let journal = journal.sync().await.unwrap();
            assert_eq!(journal.recovery_watermark(), 4);
            journal.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_start_sync_watermark_advance_deferred_failure() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let pending = PendingSyncs::default();
            let cfg = test_cfg(NZU64!(100));
            let mut journal = Box::new(
                Inner::<_, u64>::init(
                    DelayedSyncContext {
                        inner: context.child("journal"),
                        pending: pending.clone(),
                    },
                    cfg,
                )
                .await
                .unwrap(),
            );

            (journal, _) = journal.append_many(Many::Flat(&[1, 2, 3])).await.unwrap();
            let (journal, h1) = journal.start_sync().await.unwrap();
            release_pending_syncs(&pending);
            h1.await.unwrap();

            // Only the advance's metadata fsync is in flight: its failure surfaces on the
            // journal handle even though the data is durable.
            let (mut journal, h2) = journal.start_sync().await.unwrap();
            fail_pending_syncs(&pending);
            assert!(h2.await.is_err());

            // An advance at or below the staged value is skipped, so the next handle succeeds:
            // the failure is observed only by the next checkpoint write.
            (journal, _) = journal.append(&4).await.unwrap();
            let (journal, h3) = journal.start_sync().await.unwrap();
            drive_pending_syncs(&pending, h3).await.unwrap();

            // The failed advance's completion stays pending until observed: commit (which
            // does not write the checkpoint) still succeeds, and the next checkpoint write
            // observes the failure and fails.
            let journal = drive_pending_syncs(&pending, journal.commit())
                .await
                .unwrap();
            assert!(drive_pending_syncs(&pending, journal.sync()).await.is_err());
        });
    }

    #[test_traced]
    fn test_truncate_watermark_lowering_failure_keeps_blobs() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let faults = WriteFaults::default();

            // Two items per page make the two-item cap page aligned, so the armed write fault can
            // only reach the checkpoint rewrite that lowers the watermark.
            let cfg = Config {
                page_cache: CacheRef::new(NZU16!(16), PAGE_CACHE_SIZE),
                ..test_cfg(NZU64!(100))
            };
            let make = |faults: WriteFaults| {
                Inner::<_, u64>::init(
                    WriteFaultContext {
                        inner: context.child("journal"),
                        faults,
                    },
                    cfg.clone(),
                )
            };
            let mut journal = Box::new(make(faults.clone()).await.unwrap());
            (journal, _) = journal
                .append_many(Many::Flat(&[1, 2, 3, 4]))
                .await
                .unwrap();
            let journal = journal.sync().await.unwrap();
            assert_eq!(journal.recovery_watermark(), 4);

            // Recovery must durably lower the watermark before touching blob state. When the
            // lowering fails, the reopen fails with the blobs intact.
            faults.arm();
            assert!(journal.test_reopen_at_most(2).await.is_err());
            faults.disarm();

            let journal = make(faults).await.unwrap();
            assert_eq!(journal.recovery_watermark(), 4);
            assert_eq!(journal.bounds, 0..4);
            journal.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_truncate_truncates_durable_size() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let pending = PendingSyncs::default();
            let cfg = test_cfg(NZU64!(100));
            let mut journal = Box::new(
                Inner::<_, u64>::init(
                    DelayedSyncContext {
                        inner: context.child("journal"),
                        pending: pending.clone(),
                    },
                    cfg,
                )
                .await
                .unwrap(),
            );
            (journal, _) = journal.append_many(Many::Flat(&[1, 2, 3])).await.unwrap();
            let journal = drive_pending_syncs(&pending, journal.sync()).await.unwrap();
            assert_eq!(journal.recovery_watermark(), 3);

            // Truncate discards the proof for position 2. While re-appended data is still syncing,
            // the next call's advance must not raise the watermark past the truncate point.
            let mut journal = drive_pending_syncs(&pending, journal.test_truncate(2))
                .await
                .unwrap();
            (journal, _) = journal.append(&9).await.unwrap();
            let (journal, handle) = journal.start_sync().await.unwrap();
            assert_eq!(journal.recovery_watermark(), 2);

            pending.unblock();
            handle.await.unwrap();
            journal.destroy().await.unwrap();
        });
    }

    /// A failed flush inside `start_sync` returns an error.
    #[test_traced]
    fn test_fixed_start_sync_flush_failure_returns_err() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(3));
            let mut journal = Box::new(
                Inner::<_, u64>::init(context.child("journal"), cfg)
                    .await
                    .unwrap(),
            );

            // Buffer an item so start_sync must flush it.
            (journal, _) = journal.append(&0).await.unwrap();

            // The flush inside start_sync fails.
            *context.storage_fault_config().write() = deterministic::FaultConfig {
                write_rate: Some(deterministic::WriteConfig {
                    failure_rate: probability!(1.0),
                    retention_rate: probability!(0.0),
                    mode: deterministic::PartialWriteMode::Prefix,
                }),
                ..Default::default()
            };
            assert!(matches!(journal.start_sync().await, Err(Error::Runtime(_))));
        });
    }

    /// The operation that must observe the failed sync.
    enum Boundary {
        Rollover,
        Clear,
        Destroy,
    }

    /// The unobserved sync that fails.
    enum FailedSync {
        /// The live tail's started sync, whose handle is dropped.
        Tail,
        /// The sync started when the previous tail was sealed.
        Predecessor,
    }

    /// A failed sync nobody observed prevents data blob creation and removal.
    #[rstest::rstest]
    #[test_traced]
    fn test_fixed_dropped_failed_start_sync_blocks_blob_changes(
        #[values(FailedSync::Tail, FailedSync::Predecessor)] failed: FailedSync,
        #[values(Boundary::Rollover, Boundary::Clear, Boundary::Destroy)] boundary: Boundary,
    ) {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(3));
            let data_partition = blob_partition(&cfg);
            let mut journal = Box::new(
                Inner::<_, u64>::init(context.child("journal"), cfg)
                    .await
                    .unwrap(),
            );
            let fail_syncs = |rate| {
                *context.storage_fault_config().write() = deterministic::FaultConfig {
                    sync_rate: Some(rate),
                    ..Default::default()
                };
            };

            // Fail a sync without observing it: either the tail's started sync, whose handle is
            // dropped, or the sync started by the rollover that seals the first blob.
            let blobs = match failed {
                FailedSync::Tail => {
                    (journal, _) = journal.append(&0).await.unwrap();
                    fail_syncs(probability!(1.0));
                    let handle;
                    (journal, handle) = journal.start_sync().await.unwrap();
                    drop(handle);
                    1
                }
                FailedSync::Predecessor => {
                    (journal, _) = journal.append_many(Many::Flat(&[0, 1])).await.unwrap();
                    fail_syncs(probability!(1.0));
                    (journal, _) = journal.append(&2).await.unwrap();
                    2
                }
            };
            fail_syncs(probability!(0.0));

            // The failure must surface before the operation changes the data blob namespace.
            // Clearing may persist its checkpoint intent before reaching this boundary.
            let size = journal.size();
            let result = match boundary {
                Boundary::Rollover => journal
                    .append_many(Many::Flat(&[size, size + 1, size + 2]))
                    .await
                    .map(|_| ()),
                Boundary::Clear => journal.clear_to_size(size).await.map(|_| ()),
                Boundary::Destroy => journal.destroy().await,
            };
            assert!(matches!(result, Err(Error::Runtime(_))));
            let mut names = scan_partition(&context, &data_partition).await;
            names.sort();
            let expected: Vec<_> = (0..blobs)
                .map(|blob: u64| blob.to_be_bytes().to_vec())
                .collect();
            assert_eq!(names, expected);
        });
    }

    /// A failed tail sync whose handle was dropped surfaces on the next in-blob flush.
    #[test_traced]
    fn test_fixed_dropped_failed_start_sync_surfaces_before_rollover() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(1000));
            let mut journal = Box::new(
                Inner::<_, u64>::init(context.child("journal"), cfg)
                    .await
                    .unwrap(),
            );

            // Prove one item durable, then buffer more so the next start_sync rewrites the
            // tail page. Fail the sync it starts and drop the returned handle unobserved.
            (journal, _) = journal.append(&0).await.unwrap();
            journal = journal.sync().await.unwrap();
            (journal, _) = journal.append_many(Many::Flat(&[1, 2, 3])).await.unwrap();
            *context.storage_fault_config().write() = deterministic::FaultConfig {
                sync_rate: Some(probability!(1.0)),
                ..Default::default()
            };
            let (journal, handle) = journal.start_sync().await.unwrap();
            drop(handle);
            *context.storage_fault_config().write() = deterministic::FaultConfig::default();

            // The flush must observe the failed sync before reusing a checksum slot that still
            // protects the durable prefix.
            let overflow = vec![0u64; 300];
            assert!(matches!(
                journal.append_many(Many::Flat(&overflow[..])).await,
                Err(Error::Runtime(_))
            ));
        });
    }

    /// Extract a metric counter's value from encoded metrics output.
    fn counter(buffer: &str, name: &str) -> u64 {
        buffer
            .lines()
            .find(|l| l.contains(name) && !l.starts_with('#'))
            .and_then(|l| l.split_whitespace().last())
            .and_then(|v| v.parse().ok())
            .expect("counter missing")
    }

    async fn scan_partition(context: &Context, partition: &str) -> Vec<Vec<u8>> {
        match context.scan(partition).await {
            Ok(blobs) => blobs,
            Err(RuntimeError::PartitionMissing(_)) => Vec::new(),
            Err(err) => panic!("Failed to scan partition {partition}: {err}"),
        }
    }

    #[test_traced]
    fn test_fixed_journal_init_conflicting_partitions() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(2));
            let legacy_partition = cfg.partition.clone();
            let blobs_partition = blob_partition(&cfg);

            let (legacy_blob, _) = context
                .open(&legacy_partition, &0u64.to_be_bytes())
                .await
                .expect("Failed to open legacy blob");
            legacy_blob
                .write_at(0, vec![0u8; 1], WriteOptions::SYNC)
                .await
                .expect("Failed to write legacy blob");

            let (new_blob, _) = context
                .open(&blobs_partition, &0u64.to_be_bytes())
                .await
                .expect("Failed to open new blob");
            new_blob
                .write_at(0, vec![0u8; 1], WriteOptions::SYNC)
                .await
                .expect("Failed to write new blob");

            let result = Journal::<_, Digest>::init(context.child("second"), cfg.clone()).await;
            assert!(matches!(result, Err(Error::Corruption(_))));
        });
    }

    #[test_traced]
    fn test_fixed_journal_init_prefers_legacy_partition() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(2));
            let legacy_partition = cfg.partition.clone();
            let blobs_partition = blob_partition(&cfg);

            // Seed legacy partition so it is selected.
            let (legacy_blob, _) = context
                .open(&legacy_partition, &0u64.to_be_bytes())
                .await
                .expect("Failed to open legacy blob");
            legacy_blob
                .write_at(0, vec![0u8; 1], WriteOptions::SYNC)
                .await
                .expect("Failed to write legacy blob");

            drop(legacy_blob);
            let mut journal = Journal::<_, Digest>::init(context.child("first"), cfg.clone())
                .await
                .expect("failed to initialize journal");
            (journal, _) = journal.append(&test_digest(1)).await.unwrap();
            let journal = journal.sync().await.unwrap();
            drop(journal);

            let legacy_blobs = scan_partition(&context, &legacy_partition).await;
            let new_blobs = scan_partition(&context, &blobs_partition).await;
            assert!(!legacy_blobs.is_empty());
            assert!(new_blobs.is_empty());
        });
    }

    #[test_traced]
    fn test_fixed_journal_init_defaults_to_blobs_partition() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(2));
            let legacy_partition = cfg.partition.clone();
            let blobs_partition = blob_partition(&cfg);

            let mut journal = Journal::<_, Digest>::init(context.child("first"), cfg.clone())
                .await
                .expect("failed to initialize journal");
            (journal, _) = journal.append(&test_digest(1)).await.unwrap();
            let journal = journal.sync().await.unwrap();
            drop(journal);

            let legacy_blobs = scan_partition(&context, &legacy_partition).await;
            let new_blobs = scan_partition(&context, &blobs_partition).await;
            assert!(legacy_blobs.is_empty());
            assert!(!new_blobs.is_empty());
        });
    }

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

        // Start the test within the executor
        executor.start(|context| async move {
            // Initialize the journal, allowing a max of 2 items per blob.
            let cfg = test_cfg(NZU64!(2));
            let mut journal = Journal::init(context.child("first"), cfg.clone())
                .await
                .expect("failed to initialize journal");

            // Append an item to the journal
            let mut pos;
            (journal, pos) = journal
                .append(&test_digest(0))
                .await
                .expect("failed to append data 0");
            assert_eq!(pos, 0);

            // Drop the journal and re-initialize it to simulate a restart
            let journal = journal.sync().await.expect("Failed to sync journal");
            drop(journal);

            let cfg = test_cfg(NZU64!(2));
            let mut journal = Journal::init(context.child("second"), cfg.clone())
                .await
                .expect("failed to re-initialize journal");
            assert_eq!(journal.size(), 1);

            // Append two more items to the journal to trigger a new blob creation
            (journal, pos) = journal
                .append(&test_digest(1))
                .await
                .expect("failed to append data 1");
            assert_eq!(pos, 1);
            (journal, pos) = journal
                .append(&test_digest(2))
                .await
                .expect("failed to append data 2");
            assert_eq!(pos, 2);

            // Read the items back
            let item0 = journal.read(0).await.expect("failed to read data 0");
            assert_eq!(item0, test_digest(0));
            let item1 = journal.read(1).await.expect("failed to read data 1");
            assert_eq!(item1, test_digest(1));
            let item2 = journal.read(2).await.expect("failed to read data 2");
            assert_eq!(item2, test_digest(2));
            let err = journal.read(3).await.expect_err("expected read to fail");
            assert!(matches!(err, Error::ItemOutOfRange(3)));

            // Sync the journal
            journal = journal.sync().await.expect("failed to sync journal");

            // Pruning to 1 should be a no-op because there's no blob with only older items.
            (journal, _) = journal.prune(1).await.expect("failed to prune journal 1");

            // Pruning to 2 should allow the first blob to be pruned.
            (journal, _) = journal.prune(2).await.expect("failed to prune journal 2");
            assert_eq!(journal.bounds().start, 2);

            // Reading from the first blob should fail since it's now pruned
            let result0 = journal.read(0).await;
            assert!(matches!(result0, Err(Error::ItemPruned(0))));
            let result1 = journal.read(1).await;
            assert!(matches!(result1, Err(Error::ItemPruned(1))));

            // Third item should still be readable
            let result2 = journal.read(2).await.unwrap();
            assert_eq!(result2, test_digest(2));

            // Should be able to continue to append items
            for i in 3..10 {
                let pos;
                (journal, pos) = journal
                    .append(&test_digest(i))
                    .await
                    .expect("failed to append data");
                assert_eq!(pos, i);
            }

            // Check no-op pruning
            (journal, _) = journal.prune(0).await.expect("no-op pruning failed");
            assert_eq!(journal.test_oldest_blob(), Some(1));
            assert_eq!(journal.test_newest_blob(), Some(5));
            assert_eq!(journal.bounds().start, 2);

            // Prune first 3 blobs (6 items)
            (journal, _) = journal
                .prune(3 * cfg.items_per_blob.get())
                .await
                .expect("failed to prune journal 2");
            assert_eq!(journal.test_oldest_blob(), Some(3));
            assert_eq!(journal.test_newest_blob(), Some(5));
            assert_eq!(journal.bounds().start, 6);

            // Try pruning (more than) everything in the journal.
            (journal, _) = journal
                .prune(10000)
                .await
                .expect("failed to max-prune journal");
            let size = journal.size();
            assert_eq!(size, 10);
            assert_eq!(journal.test_oldest_blob(), Some(5));
            assert_eq!(journal.test_newest_blob(), Some(5));
            // Since the size of the journal is currently a multiple of items_per_blob, the newest blob
            // will be empty, and there will be no retained items.
            let bounds = journal.bounds();
            assert!(bounds.is_empty());
            // bounds.start should equal bounds.end when empty.
            assert_eq!(bounds.start, size);

            // Replaying from 0 should fail since all items before bounds.start are pruned
            {
                let reader;
                (journal, reader) = journal.snapshot().await.unwrap();
                let result = replay_items(
                    &reader,
                    0..reader.bounds().end,
                    NZUsize!(1024),
                    ReadOptions::default(),
                )
                .await;
                assert!(matches!(result, Err(Error::ItemPruned(0))));
            }

            // Replaying from pruning_boundary should return empty stream
            {
                let reader;
                (journal, reader) = journal.snapshot().await.unwrap();
                let res = replay_items(
                    &reader,
                    0..reader.bounds().end,
                    NZUsize!(1024),
                    ReadOptions::default(),
                )
                .await;
                assert!(matches!(res, Err(Error::ItemPruned(_))));

                let reader;
                (journal, reader) = journal.snapshot().await.unwrap();
                let stream = replay_items(
                    &reader,
                    journal.bounds().start..reader.bounds().end,
                    NZUsize!(1024),
                    ReadOptions::default(),
                )
                .await
                .expect("failed to replay journal from pruning boundary");
                pin_mut!(stream);
                let mut items = Vec::new();
                while let Some(result) = stream.next().await {
                    match result {
                        Ok((pos, item)) => {
                            assert_eq!(test_digest(pos), item);
                            items.push(pos);
                        }
                        Err(err) => panic!("Failed to read item: {err}"),
                    }
                }
                assert_eq!(items, Vec::<u64>::new());
            }

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

    /// Append a lot of data to make sure we exercise page cache paging boundaries.
    #[test_traced]
    fn test_fixed_journal_append_a_lot_of_data() {
        // Initialize the deterministic context
        let executor = deterministic::Runner::default();
        const ITEMS_PER_BLOB: NonZeroU64 = NZU64!(10000);
        executor.start(|context| async move {
            let cfg = test_cfg(ITEMS_PER_BLOB);
            let mut journal = Journal::init(context.child("first"), cfg.clone())
                .await
                .expect("failed to initialize journal");
            // Append 2 blobs worth of items.
            for i in 0u64..ITEMS_PER_BLOB.get() * 2 - 1 {
                (journal, _) = journal
                    .append(&test_digest(i))
                    .await
                    .expect("failed to append data");
            }
            // Sync, reopen, then read back.
            journal.sync().await.expect("failed to sync journal");
            let journal = Journal::init(context.child("second"), cfg.clone())
                .await
                .expect("failed to re-initialize journal");
            for i in 0u64..10000 {
                let item: Digest = journal.read(i).await.expect("failed to read data");
                assert_eq!(item, test_digest(i));
            }
            journal.destroy().await.expect("failed to destroy journal");
        });
    }

    #[test_traced]
    fn test_fixed_journal_replay() {
        const ITEMS_PER_BLOB: NonZeroU64 = NZU64!(7);
        // Initialize the deterministic context
        let executor = deterministic::Runner::default();

        // Start the test within the executor
        executor.start(|context| async move {
            // Initialize the journal, allowing a max of 7 items per blob.
            let cfg = test_cfg(ITEMS_PER_BLOB);
            let mut journal = Journal::init(context.child("first"), cfg.clone())
                .await
                .expect("failed to initialize journal");

            // Append many items, filling 100 blobs and part of the 101st
            for i in 0u64..(ITEMS_PER_BLOB.get() * 100 + ITEMS_PER_BLOB.get() / 2) {
                let pos;
                (journal, pos) = journal
                    .append(&test_digest(i))
                    .await
                    .expect("failed to append data");
                assert_eq!(pos, i);
            }

            // Read them back the usual way.
            for i in 0u64..(ITEMS_PER_BLOB.get() * 100 + ITEMS_PER_BLOB.get() / 2) {
                let item: Digest = journal.read(i).await.expect("failed to read data");
                assert_eq!(item, test_digest(i), "i={i}");
            }

            // Replay should return all items
            {
                let reader;
                (journal, reader) = journal.snapshot().await.unwrap();
                let stream = replay_items(
                    &reader,
                    0..reader.bounds().end,
                    NZUsize!(1024),
                    ReadOptions::default(),
                )
                .await
                .expect("failed to replay journal");
                let mut items = Vec::new();
                pin_mut!(stream);
                while let Some(result) = stream.next().await {
                    match result {
                        Ok((pos, item)) => {
                            assert_eq!(test_digest(pos), item, "pos={pos}, item={item:?}");
                            items.push(pos);
                        }
                        Err(err) => panic!("Failed to read item: {err}"),
                    }
                }

                // Make sure all items were replayed
                assert_eq!(
                    items.len(),
                    ITEMS_PER_BLOB.get() as usize * 100 + ITEMS_PER_BLOB.get() as usize / 2
                );
                items.sort();
                for (i, pos) in items.iter().enumerate() {
                    assert_eq!(i as u64, *pos);
                }
            }

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

            // Corrupt one of the bytes and make sure it's detected.
            let (blob, _) = context
                .open(&blob_partition(&cfg), &40u64.to_be_bytes())
                .await
                .expect("Failed to open blob");
            // Write junk bytes.
            let bad_bytes = 123456789u32;
            blob.write_at(1, bad_bytes.to_be_bytes().to_vec(), WriteOptions::SYNC)
                .await
                .expect("Failed to write bad bytes");

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

            // Make sure reading an item that resides in the corrupted page fails.
            let err = journal
                .read(40 * ITEMS_PER_BLOB.get() + 1)
                .await
                .unwrap_err();
            assert!(matches!(err, Error::Runtime(_)));

            // Replay all items.
            {
                let mut error_found = false;
                let (_journal, reader) = journal.snapshot().await.unwrap();
                let stream = replay_items(
                    &reader,
                    0..reader.bounds().end,
                    NZUsize!(1024),
                    ReadOptions::default(),
                )
                .await
                .expect("failed to replay journal");
                let mut items = Vec::new();
                pin_mut!(stream);
                while let Some(result) = stream.next().await {
                    match result {
                        Ok((pos, item)) => {
                            assert_eq!(test_digest(pos), item);
                            items.push(pos);
                        }
                        Err(err) => {
                            error_found = true;
                            assert!(matches!(err, Error::Runtime(_)));
                            assert!(stream.next().await.is_none());
                            break;
                        }
                    }
                }
                assert!(error_found); // error should abort replay
            }
        });
    }

    #[test_traced]
    fn test_replay_read_options_reach_history_and_tip() {
        const ITEMS_PER_BLOB: NonZeroU64 = NZU64!(6);

        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let (context, recordings) = RecordingContext::new(context);
            let cfg = test_cfg(ITEMS_PER_BLOB);
            let page_cache = cfg.page_cache.clone();
            let mut journal = Journal::init(context.child("journal"), cfg)
                .await
                .expect("failed to initialize journal");

            // The tip blob holds three items across two full pages and a partial page.
            for i in 0..9 {
                (journal, _) = journal
                    .append(&test_digest(i))
                    .await
                    .expect("failed to append");
            }
            journal = journal.sync().await.expect("failed to sync journal");

            // Sealed history receives the replay operation's read policy directly.
            page_cache.clear();
            recordings.clear();
            {
                let stream = replay_items(
                    &journal,
                    0..journal.bounds().end,
                    NZUsize!(56),
                    ReadOptions::DONT_CACHE,
                )
                .await
                .expect("failed to replay sealed history");
                pin_mut!(stream);
                let (position, item) = stream
                    .next()
                    .await
                    .expect("missing sealed replay item")
                    .expect("failed to replay sealed item");
                assert_eq!(position, 0);
                assert_eq!(item, test_digest(0));

                let reads = recordings.snapshot().reads;
                assert!(!reads.is_empty());
                assert!(
                    reads
                        .iter()
                        .all(|options| *options == ReadOptions::DONT_CACHE)
                );
            }

            // A synced writable tip has no unwritten full page, so it is replayed like sealed
            // history and its full pages arrive in one read that carries the caller's policy.
            for options in [ReadOptions::default(), ReadOptions::DONT_CACHE] {
                page_cache.clear();
                recordings.clear();
                {
                    let stream =
                        replay_items(&journal, 6..journal.bounds().end, NZUsize!(2048), options)
                            .await
                            .expect("failed to replay writable tip");
                    pin_mut!(stream);
                    for i in 6..9 {
                        let (position, item) = stream
                            .next()
                            .await
                            .expect("missing writable replay item")
                            .expect("failed to replay writable item");
                        assert_eq!(position, i);
                        assert_eq!(item, test_digest(i));
                    }
                    assert!(stream.next().await.is_none());
                    assert_eq!(recordings.snapshot().reads, vec![options]);
                }
                recordings.clear();
                assert_eq!(journal.read(6).await.unwrap(), test_digest(6));
                assert!(!recordings.snapshot().reads.is_empty());
                recordings.clear();
                assert_eq!(journal.read(6).await.unwrap(), test_digest(6));
                assert!(recordings.snapshot().reads.is_empty());
            }

            // Once the tip buffers an unwritten full page (without filling the blob, which would
            // seal it), replay goes through the page cache, which faults each missing page with
            // DONT_CACHE regardless of the caller's policy.
            for i in 9..11 {
                (journal, _) = journal
                    .append(&test_digest(i))
                    .await
                    .expect("failed to append");
            }
            page_cache.clear();
            recordings.clear();
            {
                let stream = replay_items(
                    &journal,
                    6..journal.bounds().end,
                    NZUsize!(2048),
                    ReadOptions::default(),
                )
                .await
                .expect("failed to replay buffered tip");
                pin_mut!(stream);
                let (position, item) = stream
                    .next()
                    .await
                    .expect("missing buffered replay item")
                    .expect("failed to replay buffered item");
                assert_eq!(position, 6);
                assert_eq!(item, test_digest(6));
                assert_eq!(
                    recordings.snapshot().reads,
                    vec![ReadOptions::DONT_CACHE, ReadOptions::DONT_CACHE]
                );
            }

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

    #[test_traced]
    fn test_fixed_replay_stops_after_error() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(10));
            let mut journal = Journal::init(context.child("first"), cfg.clone())
                .await
                .unwrap();

            for i in 0u64..30 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            let journal = journal.sync().await.unwrap();
            drop(journal);

            let (blob, _) = context
                .open(&blob_partition(&cfg), &1u64.to_be_bytes())
                .await
                .unwrap();
            blob.write_at(1, 123456789u32.to_be_bytes().to_vec(), WriteOptions::SYNC)
                .await
                .unwrap();

            drop(blob);
            let mut journal = Journal::<_, Digest>::init(context.child("second"), cfg.clone())
                .await
                .unwrap();
            let reader;
            (journal, reader) = journal.snapshot().await.unwrap();
            let stream = replay_items(
                &reader,
                0..reader.bounds().end,
                NZUsize!(1024),
                ReadOptions::default(),
            )
            .await
            .unwrap();
            pin_mut!(stream);

            for i in 0u64..10 {
                let (pos, item) = stream.next().await.unwrap().unwrap();
                assert_eq!(pos, i);
                assert_eq!(item, test_digest(i));
            }
            assert!(matches!(
                stream.next().await.unwrap(),
                Err(Error::Runtime(_))
            ));
            assert!(stream.next().await.is_none());

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

    #[test_traced]
    fn test_fixed_journal_replay_with_missing_historical_blob() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(2));
            let mut journal = Journal::init(context.child("first"), cfg.clone())
                .await
                .unwrap();
            for i in 0u64..5 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            let journal = journal.sync().await.unwrap();
            drop(journal);

            // Delete a middle blob (external corruption). The watermark (5) now exceeds the
            // recoverable contiguous prefix, which is corruption.
            context
                .remove(&blob_partition(&cfg), Some(&1u64.to_be_bytes()))
                .await
                .unwrap();

            let result = Journal::<_, Digest>::init(context.child("second"), cfg.clone()).await;
            assert!(matches!(result, Err(Error::Corruption(_))));
        });
    }

    #[test_traced]
    fn test_fixed_journal_partial_replay() {
        const ITEMS_PER_BLOB: NonZeroU64 = NZU64!(7);
        // 53 % 7 = 4, which will trigger a non-trivial seek in the starting blob to reach the
        // starting position.
        const START_POS: u64 = 53;

        // Initialize the deterministic context
        let executor = deterministic::Runner::default();
        // Start the test within the executor
        executor.start(|context| async move {
            // Initialize the journal, allowing a max of 7 items per blob.
            let cfg = test_cfg(ITEMS_PER_BLOB);
            let mut journal = Journal::init(context.child("storage"), cfg.clone())
                .await
                .expect("failed to initialize journal");

            // Append many items, filling 100 blobs and part of the 101st
            for i in 0u64..(ITEMS_PER_BLOB.get() * 100 + ITEMS_PER_BLOB.get() / 2) {
                let pos;
                (journal, pos) = journal
                    .append(&test_digest(i))
                    .await
                    .expect("failed to append data");
                assert_eq!(pos, i);
            }

            // Replay should return all items except the first `START_POS`.
            {
                let reader;
                (journal, reader) = journal.snapshot().await.unwrap();
                let stream = replay_items(
                    &reader,
                    START_POS..reader.bounds().end,
                    NZUsize!(1024),
                    ReadOptions::default(),
                )
                .await
                .expect("failed to replay journal");
                let mut items = Vec::new();
                pin_mut!(stream);
                while let Some(result) = stream.next().await {
                    match result {
                        Ok((pos, item)) => {
                            assert!(pos >= START_POS, "pos={pos}, expected >= {START_POS}");
                            assert_eq!(
                                test_digest(pos),
                                item,
                                "Item at position {pos} did not match expected digest"
                            );
                            items.push(pos);
                        }
                        Err(err) => panic!("Failed to read item: {err}"),
                    }
                }

                // Make sure all items were replayed
                assert_eq!(
                    items.len(),
                    ITEMS_PER_BLOB.get() as usize * 100 + ITEMS_PER_BLOB.get() as usize / 2
                        - START_POS as usize
                );
                items.sort();
                for (i, pos) in items.iter().enumerate() {
                    assert_eq!(i as u64, *pos - START_POS);
                }
            }

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

    /// Replay tiles a range in position order without crossing blob boundaries, through
    /// sealed blobs and the unsynced tail, at budgets of one item, part of a blob, and the whole
    /// journal.
    #[test_traced]
    fn test_fixed_journal_replay_batches() {
        const ITEMS_PER_BLOB: NonZeroU64 = NZU64!(7);

        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(ITEMS_PER_BLOB);
            let mut journal = Journal::init(context.child("storage"), cfg.clone())
                .await
                .expect("failed to initialize journal");

            // Fill ten blobs and part of the eleventh.
            let total = ITEMS_PER_BLOB.get() * 10 + 3;
            for i in 0u64..total {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }

            for buffer in [Digest::SIZE, 100, 2048] {
                for range in [0..total, 12..40, 20..21, 30..30, 70..total] {
                    crate::journal::contiguous::tests::check_replay_batches(
                        &journal,
                        range,
                        buffer,
                        ITEMS_PER_BLOB.get(),
                        Digest::SIZE,
                        test_digest,
                    )
                    .await;
                }
            }

            // A synced tail is replayed through its sealed pages.
            let journal = journal.sync().await.unwrap();
            crate::journal::contiguous::tests::check_replay_batches(
                &journal,
                65..total,
                100,
                ITEMS_PER_BLOB.get(),
                Digest::SIZE,
                test_digest,
            )
            .await;

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

    #[test]
    fn test_fixed_replay_bulk_retains_original_allocation() {
        deterministic::Runner::default().start(|context| async move {
            let data: Vec<u8> = (0u64..16).flat_map(u64::to_be_bytes).collect();
            let cache = CacheRef::new(NZU16!(64), NZUsize!(4));
            let (blob, size) = context
                .open("fixed-replay-sharing", b"synced")
                .await
                .unwrap();
            let writer = Writer::new(blob, size, 256, cache.clone()).await.unwrap();
            let (writer, _) = writer.append(&data).await.unwrap();
            let (writer, snapshot) = writer.snapshot().await.unwrap();
            let (blob, size) = context
                .open("fixed-replay-sharing", b"pending")
                .await
                .unwrap();
            let pending = Writer::new(blob, size, 256, cache).await.unwrap();
            let (pending, _) = pending.append(&data).await.unwrap();
            assert!(pending.try_snapshot().is_none());

            for source in [
                super::super::blobs::Blob::Sealed(snapshot),
                super::super::blobs::Blob::Writer(&writer),
                super::super::blobs::Blob::Writer(&pending),
            ] {
                let mut state = FixedReplayState::<_, View> {
                    replay: source
                        .replay_from(0, NZUsize!(256), ReadOptions::DONT_CACHE)
                        .unwrap(),
                    pos: 0,
                    end_pos: 16,
                    items_per_batch: 32,
                    _marker: PhantomData,
                };
                assert!(state.replay.ensure(View::SIZE).await.unwrap());
                let first_chunk = state.replay.chunk();
                let source = first_chunk.as_ptr() as usize;
                let count = first_chunk.len() / View::SIZE;
                let batch = {
                    let stream = super::super::replay_stream_from_states(vec![state]);
                    pin_mut!(stream);
                    stream.next().await.unwrap().unwrap()
                };
                assert!(!batch.items.is_empty());
                for (index, item) in batch.items.iter().enumerate() {
                    assert_eq!(item.bytes.as_ref(), &(index as u64).to_be_bytes());
                    if index < count {
                        assert_eq!(item.bytes.as_ptr() as usize, source + index * View::SIZE);
                    }
                    item.assert_shared();
                }
            }
        });
    }

    /// A page with a bad checksum inside a batch read ahead of the decoder ends the stream after
    /// the batches before it, with one checksum error, whether the page starts its batch or sits
    /// inside it, at buffers of 8 and 64 physical pages. Valid pages ahead of the bad one in the
    /// same batch are not yielded, since a fill validates its whole batch.
    #[test_traced]
    fn test_bad_page_in_batch_read_ahead_ends_replay() {
        const PAGE: usize = 64;
        const ITEMS: u64 = 1280;
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let physical_page = PAGE + commonware_runtime::buffer::paged::CHECKSUM_SIZE as usize;
            let items_per_page = PAGE / u64::SIZE;
            for (buffer_pages, within_batch) in [(8, 0), (8, 1), (64, 0), (64, 1)] {
                // The buffer splits into four batches. Corrupt a page of the third one.
                let batch_pages = buffer_pages / 4;
                let bad_page = 2 * batch_pages + within_batch;
                let expected_items = (2 * batch_pages * items_per_page) as u64;
                let cfg = Config {
                    partition: format!("bad-page-{buffer_pages}-{within_batch}"),
                    items_per_blob: NZU64!(ITEMS),
                    page_cache: CacheRef::new(NZU16!(PAGE as u16), NZUsize!(4)),
                    write_buffer: NZUsize!(2048),
                    replay_buffer: NZUsize!(2048),
                };
                let mut journal = Journal::<_, u64>::init(context.child("write"), cfg.clone())
                    .await
                    .unwrap();
                for item in 0..ITEMS {
                    (journal, _) = journal.append(&item).await.unwrap();
                }
                drop(journal.sync().await.unwrap());
                corrupt_page(
                    &context,
                    &format!("{}-blobs", cfg.partition),
                    &0u64.to_be_bytes(),
                    bad_page as u64,
                    PAGE as u64,
                )
                .await;
                cfg.page_cache.clear();

                let journal = Journal::<_, u64>::init(context.child("replay"), cfg)
                    .await
                    .unwrap();
                let (journal, reader) = journal.snapshot().await.unwrap();
                let stream = reader
                    .replay(
                        0..ITEMS,
                        NZUsize!(buffer_pages * physical_page),
                        ReadOptions::DONT_CACHE,
                    )
                    .await
                    .unwrap();
                pin_mut!(stream);
                let mut next = 0u64;
                loop {
                    match stream.next().await.expect("checksum error before end") {
                        Ok(batch) => {
                            for (position, item) in batch {
                                assert_eq!(position, next);
                                assert_eq!(item, next);
                                next += 1;
                            }
                        }
                        Err(error) => {
                            assert!(matches!(
                                error,
                                Error::Runtime(RuntimeError::InvalidChecksum)
                            ));
                            break;
                        }
                    }
                }
                assert_eq!(
                    next, expected_items,
                    "{buffer_pages} pages, page {bad_page}"
                );
                assert!(stream.next().await.is_none());
                drop(journal);
            }
        });
    }

    /// A codec error ends the batch stream after the items decoded before it, whether the failing
    /// item is decoded in bulk from the first page, through the per-item path because it straddles
    /// a page boundary, or in bulk from a later page.
    #[test_traced]
    fn test_fixed_journal_replay_batches_stop_at_codec_error() {
        const ITEMS: u64 = 100;

        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // With three-byte items on 44-byte logical pages, items 0..14 fit in page 0, item 14
            // straddles pages 0 and 1, and item 20 lies inside page 1.
            for invalid in [5u64, 14, 20] {
                let mut cfg = test_cfg(NZU64!(1000));
                cfg.partition = format!("codec-error-{invalid}");

                // Write byte triples that decode as `[bool; 3]` except at `invalid`.
                let mut journal = Journal::<_, [u8; 3]>::init(context.child("writer"), cfg.clone())
                    .await
                    .unwrap();
                for i in 0..ITEMS {
                    let item = if i == invalid { [0, 2, 0] } else { [0, 1, 0] };
                    (journal, _) = journal.append(&item).await.unwrap();
                }
                let journal = journal.sync().await.unwrap();
                drop(journal);

                // Replay the same bytes as `[bool; 3]` to get the valid prefix and then the error.
                let journal = Journal::<_, [bool; 3]>::init(context.child("reader"), cfg)
                    .await
                    .unwrap();
                {
                    let stream = journal
                        .replay(0..ITEMS, NZUsize!(2048), ReadOptions::default())
                        .await
                        .unwrap();
                    pin_mut!(stream);
                    let mut decoded = 0u64;
                    loop {
                        match stream.next().await.unwrap() {
                            Ok(batch) => {
                                assert_eq!(batch.start, decoded, "invalid item {invalid}");
                                assert!(
                                    batch.items.iter().all(|item| *item == [false, true, false])
                                );
                                decoded += batch.items.len() as u64;
                            }
                            Err(Error::Codec(_)) => break,
                            Err(err) => panic!("unexpected error for item {invalid}: {err}"),
                        }
                    }
                    assert_eq!(decoded, invalid, "invalid item {invalid}");
                    assert!(stream.next().await.is_none());
                }

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

    /// Replay yields exactly the requested positions, spanning blob boundaries, and
    /// validates the range against `bounds()`.
    #[test_traced]
    fn test_fixed_journal_replay_range() {
        const ITEMS_PER_BLOB: NonZeroU64 = NZU64!(7);

        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(ITEMS_PER_BLOB);
            let mut journal = Journal::init(context.child("storage"), cfg.clone())
                .await
                .expect("failed to initialize journal");

            // Fill ten blobs and part of the eleventh.
            let total = ITEMS_PER_BLOB.get() * 10 + 3;
            for i in 0u64..total {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }

            // A mid-journal range crossing several blob boundaries yields exactly [start, end).
            let (start, end) = (12u64, 40u64);
            {
                let stream =
                    replay_items(&journal, start..end, NZUsize!(1024), ReadOptions::default())
                        .await
                        .expect("failed to replay range");
                pin_mut!(stream);
                let mut next = start;
                while let Some(result) = stream.next().await {
                    let (pos, item) = result.expect("failed to read item");
                    assert_eq!(pos, next);
                    assert_eq!(item, test_digest(pos));
                    next += 1;
                }
                assert_eq!(next, end);
            }

            // An empty range yields an empty stream.
            {
                let stream = replay_items(&journal, 5..5, NZUsize!(1024), ReadOptions::default())
                    .await
                    .expect("failed to replay empty range");
                pin_mut!(stream);
                assert!(stream.next().await.is_none());
            }

            // A range past the journal's end is rejected.
            let err = replay_items(
                &journal,
                0..total + 1,
                NZUsize!(1024),
                ReadOptions::default(),
            )
            .await
            .map(|_| ())
            .unwrap_err();
            assert!(matches!(err, Error::ItemOutOfRange(pos) if pos == total + 1));

            // An inverted range is rejected.
            #[allow(clippy::reversed_empty_ranges)]
            let err = replay_items(&journal, 10..5, NZUsize!(1024), ReadOptions::default())
                .await
                .map(|_| ())
                .unwrap_err();
            assert!(matches!(err, Error::ItemOutOfRange(10)));

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

    #[test_traced]
    fn test_fixed_journal_rejects_corrupted_tail_blob() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(3));
            let mut journal = Journal::init(context.child("first"), cfg.clone())
                .await
                .unwrap();
            for i in 0..5 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            let journal = journal.sync().await.unwrap();
            drop(journal);

            // Truncate the tail blob by 1 byte (external corruption). The watermark (5) now
            // exceeds the recoverable size, which is corruption.
            let (blob, size) = context
                .open(&blob_partition(&cfg), &1u64.to_be_bytes())
                .await
                .unwrap();
            blob.resize(size - 1).await.unwrap();
            blob.sync().await.unwrap();

            drop(blob);
            let result = Journal::<_, Digest>::init(context.child("second"), cfg.clone()).await;
            assert!(matches!(result, Err(Error::Corruption(_))));
        });
    }

    /// A destroy interrupted at any point leaves a journal that reopens as a crash would recover
    /// it or as a fresh journal, and that a retried destroy removes.
    #[test_traced]
    fn test_interrupted_destroy_reopens() {
        let (mut unchanged, mut emptied) = (0, 0);
        for seed in 0..32u64 {
            let executor = deterministic::Runner::seeded(seed);
            let outcome = executor.start(|context| async move {
                // Start mid-blob, where the checkpoint's boundary hint is load-bearing, and append
                // six items across four blobs.
                let cfg = test_cfg(NZU64!(2));
                let mut journal =
                    Journal::<_, Digest>::init_at_size(context.child("first"), cfg.clone(), 5)
                        .await
                        .unwrap();
                for i in 5..11 {
                    (journal, _) = journal.append(&test_digest(i)).await.unwrap();
                }
                let journal = journal.sync().await.unwrap();

                // Fail a sample of syncs and removals so destroy stops at varying depths, from the
                // staged clear through the checkpoint's own blobs.
                context.storage_fault_config().write().sync_rate = Some(probability!(0.3));
                context.storage_fault_config().write().remove_rate = Some(probability!(0.1));
                let result = journal.destroy().await;
                context.storage_fault_config().write().sync_rate = None;
                context.storage_fault_config().write().remove_rate = None;
                if result.is_ok() {
                    return 0;
                }

                // The reopened journal is the synced one or a fresh one.
                let journal = Journal::<_, Digest>::init(context.child("second"), cfg.clone())
                    .await
                    .unwrap();
                let bounds = journal.bounds();
                let outcome = if bounds == (5..11) {
                    for pos in bounds {
                        assert_eq!(journal.read(pos).await.unwrap(), test_digest(pos));
                    }
                    1
                } else {
                    assert_eq!(bounds, 0..0, "seed {seed} reopened with {bounds:?}");
                    2
                };

                // A retried destroy completes and leaves nothing behind.
                journal.destroy().await.unwrap();
                let journal = Journal::<_, Digest>::init(context.child("third"), cfg)
                    .await
                    .unwrap();
                assert_eq!(journal.bounds(), 0..0);
                outcome
            });
            match outcome {
                1 => unchanged += 1,
                2 => emptied += 1,
                _ => {}
            }
        }
        assert!(
            unchanged > 0 && emptied > 0,
            "unchanged {unchanged} emptied {emptied}"
        );
    }

    /// Simulate a crash after recovery persists metadata but before the truncate repair completes.
    /// The stale blobs beyond the repair point still exist. The next init must succeed: it
    /// re-derives the same size from blob lengths, and the persisted watermark is still within
    /// the recovered size.
    #[test_traced]
    fn test_fixed_journal_crash_during_recovery_repair() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));
            let mut journal = Journal::<_, Digest>::init(context.child("first"), cfg.clone())
                .await
                .unwrap();

            // Fill 3 blobs (0..15), sync everything.
            for i in 0..15u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            let mut journal = journal.sync().await.unwrap();
            assert_eq!(journal.0.recovery_watermark(), 15);

            // Persist the recovered metadata (watermark=9) as init does before applying the
            // truncate repair. This simulates a crash after metadata sync but before the repair
            // removes stale blobs.
            journal.0.checkpoint = journal
                .0
                .checkpoint
                .persist(cfg.items_per_blob.get(), 0, 9)
                .await
                .unwrap();
            drop(journal);

            // Shorten blob 1 to simulate a short non-tail blob. Recovery will compute
            // size=9 (blob 0 full + 4 items in blob 1) and generate a repair.
            {
                let cache_ref = CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE);
                let (blob, blob_size) = context
                    .open(&blob_partition(&cfg), &1u64.to_be_bytes())
                    .await
                    .expect("failed to open blob 1");
                let mut append = PagedRecovery::open(blob, blob_size, 2048, cache_ref)
                    .await
                    .expect("failed to wrap blob 1");
                append = append
                    .truncate(4 * Digest::SIZE as u64)
                    .await
                    .expect("failed to shorten blob 1");
                append
                    .sync()
                    .await
                    .expect("failed to sync shortened blob 1");
            }

            // Blobs 2 (and the empty tail at 3) still exist. Init must succeed and the
            // truncate must remove the stale blobs.
            let journal = Journal::<_, Digest>::init(context.child("second"), cfg.clone())
                .await
                .expect("init should succeed after crash during recovery repair");
            assert_eq!(journal.bounds(), 0..9);
            assert_eq!(journal.0.recovery_watermark(), 9);
            assert_eq!(journal.read(8).await.unwrap(), test_digest(8));
            assert!(matches!(
                journal.read(9).await,
                Err(Error::ItemOutOfRange(9))
            ));
            assert_eq!(
                journal.test_newest_blob(),
                Some(1),
                "stale blobs beyond the repair point should be removed"
            );

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

    #[test_traced]
    fn test_fixed_journal_recover_accepts_clean_short_tail() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));

            // Set up via the public API: 5 items in blob 0 (full) + 2 items in blob 1
            // (partial), then sync and drop.
            let mut journal = Journal::<_, Digest>::init(context.child("first"), cfg.clone())
                .await
                .unwrap();
            for i in 0..7 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            journal.sync().await.unwrap();

            // Reopen and verify the size is exactly 7 with no repair (a clean short tail).
            let journal = Journal::<_, Digest>::init(context.child("second"), cfg.clone())
                .await
                .unwrap();
            assert_eq!(journal.size(), 7);
            // Blobs 0 and 1 exist and we can read every position.
            for i in 0..7u64 {
                assert_eq!(journal.read(i).await.unwrap(), test_digest(i));
            }
            journal.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_fixed_journal_recover_accepts_clean_empty_tail() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));

            // Set up via the public API: 5 items in blob 0 (full); rolling over implicitly
            // creates an empty blob 1 as the tail. Sync and drop.
            let mut journal = Journal::<_, Digest>::init(context.child("first"), cfg.clone())
                .await
                .unwrap();
            for i in 0..5 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            journal.sync().await.unwrap();

            // Reopen: blob 0 is full, blob 1 is the empty tail. Size = 5, no repair.
            let journal = Journal::<_, Digest>::init(context.child("second"), cfg.clone())
                .await
                .unwrap();
            assert_eq!(journal.size(), 5);
            for i in 0..5u64 {
                assert_eq!(journal.read(i).await.unwrap(), test_digest(i));
            }
            assert_eq!(journal.test_newest_blob(), Some(1));
            journal.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_fixed_journal_recover_sparse_blob_ids_repairs_at_gap() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(1));
            let blob_partition = blob_partition(&cfg);

            let mut journal = Journal::<_, Digest>::init(context.child("first"), cfg.clone())
                .await
                .unwrap();
            (journal, _) = journal.append(&test_digest(0)).await.unwrap();
            let journal = journal.sync().await.unwrap();
            drop(journal);

            // Add a far-future blob directly. Recovery should inspect actual blob ids and
            // repair at the first missing boundary instead of walking the entire numeric range.
            let cache_ref = CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE);
            let (blob, blob_size) = context
                .open(&blob_partition, &u64::MAX.to_be_bytes())
                .await
                .unwrap();
            let mut append = Writer::new(blob, blob_size, 2048, cache_ref).await.unwrap();
            let extra = test_digest(999);
            (append, _) = append.append(extra.as_ref()).await.unwrap();
            append = append.sync().await.unwrap();
            drop(append);

            let journal = Journal::<_, Digest>::init(context.child("second"), cfg.clone())
                .await
                .unwrap();
            assert_eq!(journal.bounds(), 0..1);
            assert_eq!(journal.read(0).await.unwrap(), test_digest(0));
            assert!(matches!(
                journal.read(1).await,
                Err(Error::ItemOutOfRange(1))
            ));
            assert_eq!(journal.test_newest_blob(), Some(1));

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

    #[test_traced]
    fn test_fixed_journal_recover_fallback_truncates_after_short_oldest_blob() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));
            let mut journal =
                Journal::<_, Digest>::init_at_size(context.child("first"), cfg.clone(), 7)
                    .await
                    .expect("failed to initialize journal at size");

            for i in 0..8u64 {
                (journal, _) = journal
                    .append(&test_digest(100 + i))
                    .await
                    .expect("failed to append data");
            }
            let journal = journal.sync().await.expect("failed to sync journal");
            assert_eq!(journal.bounds(), 7..15);

            let journal = journal
                .test_set_recovery_watermark(6)
                .await
                .expect("failed to sync lower recovery watermark");
            drop(journal);

            let (blob, size) = context
                .open(&blob_partition(&cfg), &1u64.to_be_bytes())
                .await
                .expect("failed to open oldest blob");
            blob.resize(size - 1).await.expect("failed to corrupt blob");
            blob.sync().await.expect("failed to sync blob");

            drop(blob);
            let journal = Journal::<_, Digest>::init(context.child("second"), cfg.clone())
                .await
                .expect("failed to recover journal");
            assert_eq!(journal.bounds(), 7..9);
            assert_eq!(journal.read(7).await.unwrap(), test_digest(100));
            assert_eq!(journal.read(8).await.unwrap(), test_digest(101));
            assert!(matches!(
                journal.read(9).await,
                Err(Error::ItemOutOfRange(9))
            ));
            assert_eq!(journal.test_oldest_blob(), Some(1));
            assert_eq!(journal.test_newest_blob(), Some(1));

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

    #[test_traced]
    fn test_fixed_journal_stale_pruning_metadata_preserves_watermark() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));
            let mut journal =
                Journal::<_, Digest>::init_at_size(context.child("first"), cfg.clone(), 7)
                    .await
                    .expect("failed to initialize journal at size");

            for i in 0..10u64 {
                (journal, _) = journal
                    .append(&test_digest(i))
                    .await
                    .expect("failed to append data");
            }
            let journal = journal.sync().await.expect("failed to sync journal");
            assert_eq!(journal.bounds(), 7..17);

            // Stage the stale forward-looking watermark while the journal is alive (so we go
            // through the public metadata path), then drop and corrupt the underlying blob.
            let journal = journal
                .test_set_recovery_watermark(12)
                .await
                .expect("failed to sync recovery watermark");
            drop(journal);

            // Shorten blob 2 to two items through paged recovery so the on-disk logical view
            // matches the staged watermark of 12.
            {
                let cache_ref = CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE);
                let (blob, blob_size) = context
                    .open(&blob_partition(&cfg), &2u64.to_be_bytes())
                    .await
                    .expect("failed to open blob 2");
                let mut append = PagedRecovery::open(blob, blob_size, 2048, cache_ref)
                    .await
                    .expect("failed to wrap blob 2");
                append = append
                    .truncate(2 * Digest::SIZE as u64)
                    .await
                    .expect("failed to shorten anchored blob");
                append.sync().await.expect("failed to sync blob 2");
            }

            // Remove the checkpoint's oldest blob so the boundary hint of 7 is stale. The
            // watermark is preserved because length-based recovery ends at the same point.
            context
                .remove(&blob_partition(&cfg), Some(&1u64.to_be_bytes()))
                .await
                .expect("failed to remove stale oldest blob");

            let journal = Journal::<_, Digest>::init(context.child("second"), cfg.clone())
                .await
                .expect("failed to recover journal");
            assert_eq!(journal.bounds(), 10..12);
            assert_eq!(journal.0.recovery_watermark(), 12);
            assert_eq!(journal.read(10).await.unwrap(), test_digest(3));
            assert_eq!(journal.read(11).await.unwrap(), test_digest(4));
            assert!(matches!(
                journal.read(12).await,
                Err(Error::ItemOutOfRange(12))
            ));

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

    #[test_traced]
    fn test_fixed_journal_stale_pruning_metadata_without_watermark_walks_lengths() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));
            let mut journal =
                Journal::<_, Digest>::init_at_size(context.child("first"), cfg.clone(), 7)
                    .await
                    .expect("failed to initialize journal at size");

            for i in 0..10u64 {
                (journal, _) = journal
                    .append(&test_digest(i))
                    .await
                    .expect("failed to append data");
            }
            let mut journal = journal.sync().await.expect("failed to sync journal");
            assert_eq!(journal.bounds(), 7..17);

            {
                journal.0.checkpoint.set_watermark(None);
                journal.0.checkpoint = journal
                    .0
                    .checkpoint
                    .sync()
                    .await
                    .expect("failed to remove recovery watermark");
            }
            drop(journal);

            // Remove the checkpoint's oldest blob so the boundary hint of 7 is stale. Without a
            // recovery watermark, recovery must still walk lengths from the recovered blob boundary.
            context
                .remove(&blob_partition(&cfg), Some(&1u64.to_be_bytes()))
                .await
                .expect("failed to remove stale oldest blob");

            let mut journal = Journal::<_, Digest>::init(context.child("second"), cfg.clone())
                .await
                .expect("failed to recover journal");
            assert_eq!(journal.bounds(), 10..17);
            // No watermark: watermark at the tail blob start, not size.
            assert_eq!(journal.0.recovery_watermark(), 15);
            assert_eq!(journal.read(10).await.unwrap(), test_digest(3));
            assert_eq!(journal.read(16).await.unwrap(), test_digest(9));

            // After sync, watermark advances to the full recovered size.
            journal = journal.sync().await.expect("failed to sync");
            assert_eq!(journal.0.recovery_watermark(), 17);

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

    /// A boundary hint ahead of the oldest blob is not a reachable crash state: prune removes
    /// blobs before sync persists the checkpoint, and clear_to_size stages a clear intent for
    /// atomicity. Verify it is rejected as corruption.
    #[test_traced]
    fn test_fixed_journal_boundary_hint_ahead_of_blobs_is_corruption() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));
            let mut journal =
                Journal::<_, Digest>::init_at_size(context.child("first"), cfg.clone(), 3)
                    .await
                    .unwrap();

            // Append 12 items (positions 3..15) spanning blobs 0, 1, 2.
            for i in 0..12u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            let mut journal = journal.sync().await.unwrap();
            assert_eq!(journal.bounds(), 3..15);

            // Set the boundary hint to 8 (blob 1) and lower the watermark so it won't
            // independently trigger the watermark > size corruption check. Then remove blob 1's
            // blob so blob 0 is the oldest. The boundary hint now references a blob ahead
            // of the oldest blob, which is the corruption we're testing.
            {
                journal.0.checkpoint.set_boundary_hint(8);
                journal.0.checkpoint.set_watermark(Some(3));
                journal.0.checkpoint = journal.0.checkpoint.sync().await.unwrap();
            }
            drop(journal);

            context
                .remove(&blob_partition(&cfg), Some(&1u64.to_be_bytes()))
                .await
                .unwrap();

            let result = Journal::<_, Digest>::init(context.child("second"), cfg.clone()).await;
            assert!(matches!(result, Err(Error::Corruption(_))));
        });
    }

    /// A mid-blob boundary hint with no blobs is not a reachable crash state (see comment in
    /// `recover_bounds`). Verify it is rejected as corruption rather than silently recovering empty.
    #[test_traced]
    fn test_fixed_journal_boundary_hint_with_no_blobs_is_corruption() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));
            let mut journal =
                Journal::<_, Digest>::init_at_size(context.child("first"), cfg.clone(), 7)
                    .await
                    .unwrap();

            for i in 0..3u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            let journal = journal.sync().await.unwrap();
            drop(journal);

            // Remove all blobs but leave the checkpoint (with a boundary hint of 7) intact.
            for name in scan_partition(&context, &blob_partition(&cfg)).await {
                context
                    .remove(&blob_partition(&cfg), Some(&name))
                    .await
                    .unwrap();
            }

            let result = Journal::<_, Digest>::init(context.child("second"), cfg.clone()).await;
            assert!(matches!(result, Err(Error::Corruption(_))));
        });
    }

    #[test_traced]
    fn test_fixed_journal_legacy_recovery_installs_watermark() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));
            let mut journal = Journal::<_, Digest>::init(context.child("first"), cfg.clone())
                .await
                .expect("failed to initialize journal");

            for i in 0..12u64 {
                (journal, _) = journal
                    .append(&test_digest(i))
                    .await
                    .expect("failed to append data");
            }
            let mut journal = journal.sync().await.expect("failed to sync journal");

            {
                journal.0.checkpoint.set_watermark(None);
                journal.0.checkpoint = journal
                    .0
                    .checkpoint
                    .sync()
                    .await
                    .expect("failed to remove recovery watermark");
            }
            drop(journal);

            // Legacy recovery sets watermark to the tail blob start, not size, so sync must fsync
            // the tail before advancing the watermark.
            let mut journal = Journal::<_, Digest>::init(context.child("second"), cfg.clone())
                .await
                .expect("failed to recover legacy journal");
            assert_eq!(journal.bounds(), 0..12);
            assert_eq!(journal.0.recovery_watermark(), 10);

            // After sync, the watermark advances to the full size.
            journal = journal
                .sync()
                .await
                .expect("failed to sync after legacy recovery");
            assert_eq!(journal.0.recovery_watermark(), 12);

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

    #[test_traced]
    fn test_fixed_journal_commit_does_not_advance_recovery_watermark() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));
            let mut journal = Journal::<_, Digest>::init(context.child("journal"), cfg.clone())
                .await
                .unwrap();

            (journal, _) = journal.append(&test_digest(0)).await.unwrap();
            let mut journal = journal.sync().await.unwrap();
            assert_eq!(journal.0.recovery_watermark(), 1);

            (journal, _) = journal.append(&test_digest(1)).await.unwrap();
            let mut journal = journal.commit().await.unwrap();
            assert_eq!(
                journal.0.recovery_watermark(),
                1,
                "commit must make new data durable without advancing the recovery watermark",
            );

            journal = journal.sync().await.unwrap();
            assert_eq!(journal.0.recovery_watermark(), 2);
            journal.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_fixed_journal_prune_to_blob_boundary_removes_pruning_metadata() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));
            let mut journal =
                Journal::<_, Digest>::init_at_size(context.child("first"), cfg.clone(), 7)
                    .await
                    .expect("failed to initialize journal at size");

            for i in 0..8u64 {
                (journal, _) = journal
                    .append(&test_digest(i))
                    .await
                    .expect("failed to append data");
            }
            let mut journal = journal.sync().await.expect("failed to sync journal");
            assert_eq!(journal.bounds(), 7..15);

            (journal, _) = journal.prune(10).await.expect("failed to prune journal");
            let journal = journal.sync().await.expect("failed to sync pruned journal");
            assert_eq!(journal.bounds(), 10..15);
            drop(journal);

            let checkpoint = Checkpoint::open(context.child("metadata"), &cfg.partition)
                .await
                .expect("failed to reopen checkpoint");
            assert!(checkpoint.boundary_hint().is_none());
            drop(checkpoint);

            let journal = Journal::<_, Digest>::init(context.child("second"), cfg.clone())
                .await
                .expect("failed to reopen journal");
            assert_eq!(journal.bounds(), 10..15);
            assert_eq!(journal.read(10).await.unwrap(), test_digest(3));
            journal.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_fixed_journal_recover_rejects_overlong_blob() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));
            let mut journal = Journal::<_, Digest>::init(context.child("first"), cfg.clone())
                .await
                .expect("failed to initialize journal");

            for i in 0..5u64 {
                (journal, _) = journal
                    .append(&test_digest(i))
                    .await
                    .expect("failed to append data");
            }
            let journal = journal.sync().await.expect("failed to sync journal");
            drop(journal);

            // Inject an extra item into blob 0 at the blob level so its length exceeds
            // items_per_blob, which `recover_bounds` validates and rejects as Corruption.
            {
                let extra = test_digest(99);
                let cache_ref = CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE);
                let (blob, blob_size) = context
                    .open(&blob_partition(&cfg), &0u64.to_be_bytes())
                    .await
                    .expect("failed to open blob 0");
                let mut append = Writer::new(blob, blob_size, 2048, cache_ref)
                    .await
                    .expect("failed to wrap blob 0");
                (append, _) = append
                    .append(extra.as_ref())
                    .await
                    .expect("failed to append extra item");
                append.sync().await.expect("failed to sync corrupted blob");
            }

            let result = Journal::<_, Digest>::init(context.child("second"), cfg.clone()).await;
            assert!(matches!(result, Err(Error::Corruption(_))));

            // A bound does not excuse the overlong blob, whether it cuts inside the blob or at
            // its boundary.
            for cap in [3, 5] {
                let result =
                    Journal::<_, Digest>::init_at_most(context.child("capped"), cfg.clone(), cap)
                        .await;
                assert!(matches!(result, Err(Error::Corruption(_))), "cap {cap}");
            }
        });
    }

    #[test_traced("DEBUG")]
    fn test_fixed_journal_recover_from_unwritten_data() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Initialize the journal, allowing a max of 10 items per blob.
            let cfg = test_cfg(NZU64!(10));
            let mut journal = Journal::init(context.child("first"), cfg.clone())
                .await
                .expect("failed to initialize journal");

            // Add only a single item
            (journal, _) = journal
                .append(&test_digest(0))
                .await
                .expect("failed to append data");
            assert_eq!(journal.size(), 1);
            let journal = journal.sync().await.expect("Failed to sync journal");
            drop(journal);

            // Manually extend the blob to simulate a failure where the file was extended, but no
            // bytes were written due to failure.
            let (blob, size) = context
                .open(&blob_partition(&cfg), &0u64.to_be_bytes())
                .await
                .expect("Failed to open blob");
            blob.write_at(
                size,
                vec![0u8; PAGE_SIZE.get() as usize * 3],
                WriteOptions::SYNC,
            )
            .await
            .expect("Failed to extend blob");

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

            // The zero-filled pages are detected as invalid (bad checksum) and truncated.
            // No items should be lost since we called sync before the corruption.
            assert_eq!(journal.size(), 1);

            // Make sure journal still works for appending.
            (journal, _) = journal
                .append(&test_digest(1))
                .await
                .expect("failed to append data");

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

    #[test_traced]
    fn test_fixed_journal_init_at_most() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Initialize the journal, allowing a max of 2 items per blob.
            let cfg = test_cfg(NZU64!(2));
            let journal: Journal<_, Digest> = Journal::init(context.child("first"), cfg.clone())
                .await
                .expect("failed to initialize journal");
            let journal = {
                _ = journal.sync().await.unwrap();
                Journal::<_, Digest>::init_at_most(context.child("cap"), cfg.clone(), 0).await
            }
            .unwrap();
            drop(journal);
            let journal =
                Journal::<_, Digest>::init_at_most(context.child("above"), cfg.clone(), 1)
                    .await
                    .unwrap();
            assert_eq!(journal.size(), 0);
            drop(journal);
            let mut journal: Journal<_, Digest> =
                Journal::init(context.child("reopen"), cfg.clone())
                    .await
                    .expect("failed to re-initialize journal");

            // Append an item to the journal
            (journal, _) = journal
                .append(&test_digest(0))
                .await
                .expect("failed to append data 0");
            assert_eq!(journal.size(), 1);
            journal = {
                _ = journal.sync().await.unwrap();
                Journal::<_, Digest>::init_at_most(context.child("cap"), cfg.clone(), 1).await
            }
            .unwrap(); // should be no-op
            journal = {
                _ = journal.sync().await.unwrap();
                Journal::<_, Digest>::init_at_most(context.child("cap"), cfg.clone(), 0).await
            }
            .unwrap();
            assert_eq!(journal.size(), 0);

            // append 7 items
            for i in 0..7 {
                let pos;
                (journal, pos) = journal
                    .append(&test_digest(i))
                    .await
                    .expect("failed to append data");
                assert_eq!(pos, i);
            }
            assert_eq!(journal.size(), 7);

            // Open the prefix ending at item 4.
            journal = {
                _ = journal.sync().await.unwrap();
                Journal::<_, Digest>::init_at_most(context.child("cap"), cfg.clone(), 4).await
            }
            .unwrap();
            assert_eq!(journal.size(), 4);

            // Open the empty prefix.
            journal = {
                _ = journal.sync().await.unwrap();
                Journal::<_, Digest>::init_at_most(context.child("cap"), cfg.clone(), 0).await
            }
            .unwrap();
            assert_eq!(journal.size(), 0);

            // Repeatedly append 100 items and reopen 49 items earlier.
            for _ in 0..10 {
                for i in 0..100 {
                    (journal, _) = journal
                        .append(&test_digest(i))
                        .await
                        .expect("failed to append data");
                }
                let size = journal.size();
                journal = {
                    _ = journal.sync().await.unwrap();
                    Journal::<_, Digest>::init_at_most(context.child("cap"), cfg.clone(), size - 49)
                        .await
                }
                .unwrap();
            }
            const ITEMS_REMAINING: u64 = 10 * (100 - 49);
            assert_eq!(journal.size(), ITEMS_REMAINING);

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

            // Repeat with a different blob size (3 items per blob)
            let mut cfg = test_cfg(NZU64!(3));
            cfg.partition = "test-partition-2".into();
            let mut journal = Journal::init(context.child("second"), cfg.clone())
                .await
                .expect("failed to initialize journal");
            for _ in 0..10 {
                for i in 0..100 {
                    (journal, _) = journal
                        .append(&test_digest(i))
                        .await
                        .expect("failed to append data");
                }
                let size = journal.size();
                journal = {
                    _ = journal.sync().await.unwrap();
                    Journal::<_, Digest>::init_at_most(context.child("cap"), cfg.clone(), size - 49)
                        .await
                }
                .unwrap();
            }
            assert_eq!(journal.size(), ITEMS_REMAINING);

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

            // Make sure re-opened journal is as expected
            let mut journal: Journal<_, Digest> =
                Journal::init(context.child("third"), cfg.clone())
                    .await
                    .expect("failed to re-initialize journal");
            assert_eq!(journal.size(), 10 * (100 - 49));

            // Make sure truncating works after pruning
            (journal, _) = journal.prune(300).await.expect("pruning failed");
            assert_eq!(journal.size(), ITEMS_REMAINING);

            // Truncating to the prune point should work. Always remain in the journal.
            journal = {
                _ = journal.sync().await.unwrap();
                Journal::<_, Digest>::init_at_most(context.child("cap"), cfg.clone(), 300).await
            }
            .unwrap();
            let bounds = journal.bounds();
            assert_eq!(bounds.end, 300);
            assert!(bounds.is_empty());

            // Truncating prior to our prune point should fail.
            assert!(matches!(
                {
                    _ = journal.sync().await.unwrap();
                    Journal::<_, Digest>::init_at_most(context.child("cap"), cfg.clone(), 299).await
                },
                Err(Error::ItemPruned(299))
            ));
        });
    }

    #[test_traced]
    fn test_fixed_journal_init_at_most_commit_reopen() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));
            let mut journal = Journal::<_, Digest>::init(context.child("first"), cfg.clone())
                .await
                .expect("failed to initialize journal");

            for i in 0..12u64 {
                (journal, _) = journal
                    .append(&test_digest(i))
                    .await
                    .expect("failed to append data");
            }
            let journal = journal.sync().await.expect("failed to sync journal");

            let journal = {
                _ = journal.sync().await.unwrap();
                Journal::<_, Digest>::init_at_most(context.child("cap"), cfg.clone(), 7).await
            }
            .expect("failed to initialize journal prefix");
            journal.commit().await.expect("failed to commit journal");

            let journal = Journal::<_, Digest>::init(context.child("second"), cfg.clone())
                .await
                .expect("failed to re-initialize journal");
            assert_eq!(journal.bounds(), 0..7);
            for i in 0..7u64 {
                assert_eq!(journal.read(i).await.unwrap(), test_digest(i));
            }
            assert!(matches!(
                journal.read(7).await,
                Err(Error::ItemOutOfRange(7))
            ));

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

    #[test_traced]
    fn test_fixed_journal_init_at_most_persists_lower_watermark() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));
            let mut journal = Journal::<_, Digest>::init(context.child("first"), cfg.clone())
                .await
                .expect("failed to initialize journal");

            for i in 0..12u64 {
                (journal, _) = journal
                    .append(&test_digest(i))
                    .await
                    .expect("failed to append data");
            }
            let journal = journal.sync().await.expect("failed to sync journal");
            {
                _ = journal.sync().await.unwrap();
                Journal::<_, Digest>::init_at_most(context.child("cap"), cfg.clone(), 7).await
            }
            .expect("failed to truncate journal");

            let checkpoint = Checkpoint::open(context.child("metadata"), &cfg.partition)
                .await
                .expect("failed to reopen checkpoint");
            let persisted_watermark = checkpoint
                .watermark()
                .expect("missing recovery watermark after truncate");
            assert_eq!(persisted_watermark, 7);
            drop(checkpoint);

            let journal = Journal::<_, Digest>::init(context.child("second"), cfg.clone())
                .await
                .expect("failed to re-initialize journal");
            journal.destroy().await.unwrap();
        });
    }

    /// Reopen at a page-aligned bound, append over the freed pages, then crash with the appends
    /// retained and any unsynced resize lost. Recovery must not join the new pages with the
    /// discarded suffix.
    #[test_traced]
    fn test_fixed_journal_init_at_most_truncation_survives_crash() {
        // Two digests per page make the four-item bound page aligned. The two-page write buffer
        // floor sends a five-item append down the direct path: four items land in two full pages
        // and the fifth stays buffered, so the old pages behind them are never rewritten.
        const PAGE_SIZE: NonZeroU16 = NZU16!(64);
        const BOUND: u64 = 4;

        fn cfg() -> Config {
            Config {
                partition: "init-at-most-truncation-crash".into(),
                items_per_blob: NZU64!(100),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1),
                replay_buffer: NZUsize!(2048),
            }
        }
        let appended: Vec<Digest> = (100..105u64).map(test_digest).collect();

        let executor = deterministic::Runner::default();
        let (appended, checkpoint) = executor.start_and_recover(|context| async move {
            let mut journal = Journal::<_, Digest>::init(context.child("first"), cfg())
                .await
                .unwrap();
            for i in 0..12u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            let journal = journal.sync().await.unwrap();
            drop(journal);

            // Keep unsynced writes and drop unsynced resizes at the crash.
            *context.storage_fault_config().write() = deterministic::FaultConfig {
                write_rate: Some(deterministic::WriteConfig {
                    failure_rate: probability!(0.0),
                    retention_rate: probability!(1.0),
                    mode: deterministic::PartialWriteMode::Prefix,
                }),
                resize_rate: Some(deterministic::ResizeConfig {
                    failure_rate: probability!(0.0),
                    partial_rate: probability!(0.0),
                }),
                ..Default::default()
            };
            let journal = Journal::<_, Digest>::init_at_most(context.child("cap"), cfg(), BOUND)
                .await
                .unwrap();
            assert_eq!(journal.bounds(), 0..BOUND);
            let (journal, _) = journal.append_many(Many::Flat(&appended)).await.unwrap();
            drop(journal);
            appended
        });

        deterministic::Runner::from(checkpoint).start(|context| async move {
            *context.storage_fault_config().write() = deterministic::FaultConfig::default();
            let journal = Journal::<_, Digest>::init(context.child("recover"), cfg())
                .await
                .unwrap();
            let size = journal.size();
            assert!(
                (BOUND..=BOUND + appended.len() as u64).contains(&size),
                "recovered size {size} is not a prefix of the new history"
            );
            for pos in 0..BOUND {
                assert_eq!(journal.read(pos).await.unwrap(), test_digest(pos));
            }
            for pos in BOUND..size {
                assert_eq!(
                    journal.read(pos).await.unwrap(),
                    appended[(pos - BOUND) as usize],
                    "content at retained position {pos} was never written there"
                );
            }
            journal.destroy().await.unwrap();
        });
    }

    /// Repair a torn tail at a page-aligned item boundary during init, append over the freed
    /// pages, then crash with the appends retained and any unsynced resize lost. Recovery must
    /// not join the new pages with the discarded suffix. The default fault arm drops the appends
    /// too, so recovery must land exactly on the repaired size.
    #[test_traced]
    fn test_fixed_journal_repair_truncation_survives_crash() {
        // Two digests per page make the four-item repair target page aligned. The two-page write
        // buffer floor sends a five-item append down the direct path: four items land in two full
        // pages and the fifth stays buffered, so the old pages behind them are never rewritten.
        const PAGE_SIZE: NonZeroU16 = NZU16!(64);
        const REPAIRED: u64 = 4;

        fn cfg() -> Config {
            Config {
                partition: "repair-truncation-crash".into(),
                items_per_blob: NZU64!(100),
                page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
                write_buffer: NZUsize!(1),
                replay_buffer: NZUsize!(2048),
            }
        }

        for retained in [true, false] {
            let appended: Vec<Digest> = (100..105u64).map(test_digest).collect();

            let executor = deterministic::Runner::default();
            let (appended, checkpoint) = executor.start_and_recover(|context| async move {
                let cfg = cfg();
                let mut journal = Journal::<_, Digest>::init(context.child("first"), cfg.clone())
                    .await
                    .unwrap();
                for i in 0..12u64 {
                    (journal, _) = journal.append(&test_digest(i)).await.unwrap();
                }

                // Commit leaves the watermark behind, so init scans the whole blob.
                let journal = journal.commit().await.unwrap();
                drop(journal);

                // Tear page 2 of the six-page blob. Init stops at the hole and repairs the blob
                // to the four items on pages 0 and 1, leaving pages 3 to 5 behind the target.
                corrupt_page(
                    &context,
                    &blob_partition(&cfg),
                    &0u64.to_be_bytes(),
                    2,
                    PAGE_SIZE.get() as u64,
                )
                .await;

                // Keep unsynced writes and drop unsynced resizes at the crash. The default
                // drops both.
                *context.storage_fault_config().write() = if retained {
                    deterministic::FaultConfig {
                        write_rate: Some(deterministic::WriteConfig {
                            failure_rate: probability!(0.0),
                            retention_rate: probability!(1.0),
                            mode: deterministic::PartialWriteMode::Prefix,
                        }),
                        resize_rate: Some(deterministic::ResizeConfig {
                            failure_rate: probability!(0.0),
                            partial_rate: probability!(0.0),
                        }),
                        ..Default::default()
                    }
                } else {
                    deterministic::FaultConfig::default()
                };
                let journal = Journal::<_, Digest>::init(context.child("repair"), cfg)
                    .await
                    .unwrap();
                assert_eq!(journal.bounds(), 0..REPAIRED);
                let (journal, _) = journal.append_many(Many::Flat(&appended)).await.unwrap();
                drop(journal);
                appended
            });

            deterministic::Runner::from(checkpoint).start(|context| async move {
                *context.storage_fault_config().write() = deterministic::FaultConfig::default();
                let journal = Journal::<_, Digest>::init(context.child("recover"), cfg())
                    .await
                    .unwrap();
                let size = journal.size();
                let kept = if retained { appended.len() as u64 } else { 0 };
                assert!(
                    (REPAIRED..=REPAIRED + kept).contains(&size),
                    "recovered size {size} is not a prefix of the new history"
                );
                for pos in 0..REPAIRED {
                    assert_eq!(journal.read(pos).await.unwrap(), test_digest(pos));
                }
                for pos in REPAIRED..size {
                    assert_eq!(
                        journal.read(pos).await.unwrap(),
                        appended[(pos - REPAIRED) as usize],
                        "content at retained position {pos} was never written there"
                    );
                }
                journal.destroy().await.unwrap();
            });
        }
    }

    #[test_traced]
    fn test_fixed_journal_recover_after_watermark_lowered_before_truncate() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));
            let mut journal = Journal::<_, Digest>::init(context.child("first"), cfg.clone())
                .await
                .expect("failed to initialize journal");

            for i in 0..12u64 {
                (journal, _) = journal
                    .append(&test_digest(i))
                    .await
                    .expect("failed to append data");
            }
            let journal = journal.sync().await.expect("failed to sync journal");

            let journal = journal
                .test_set_recovery_watermark(7)
                .await
                .expect("failed to lower recovery watermark");
            drop(journal);

            let journal = Journal::<_, Digest>::init(context.child("second"), cfg.clone())
                .await
                .expect("failed to recover journal");
            assert_eq!(journal.bounds(), 0..12);
            assert_eq!(journal.0.recovery_watermark(), 7);
            assert_eq!(journal.read(11).await.unwrap(), test_digest(11));
            journal.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_fixed_journal_truncate_append_commit_reopen() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));
            let mut journal = Journal::<_, Digest>::init(context.child("first"), cfg.clone())
                .await
                .expect("failed to initialize journal");

            for i in 0..12u64 {
                (journal, _) = journal
                    .append(&test_digest(i))
                    .await
                    .expect("failed to append data");
            }
            let journal = journal.sync().await.expect("failed to sync journal");

            let mut journal = {
                _ = journal.sync().await.unwrap();
                Journal::<_, Digest>::init_at_most(context.child("cap"), cfg.clone(), 7).await
            }
            .expect("failed to truncate journal");
            for i in 0..3u64 {
                (journal, _) = journal
                    .append(&test_digest(100 + i))
                    .await
                    .expect("failed to append data");
            }
            journal.commit().await.expect("failed to commit journal");

            let journal = Journal::<_, Digest>::init(context.child("second"), cfg.clone())
                .await
                .expect("failed to re-initialize journal");
            assert_eq!(journal.bounds(), 0..10);
            assert_eq!(journal.0.recovery_watermark(), 7);
            for i in 0..7u64 {
                assert_eq!(journal.read(i).await.unwrap(), test_digest(i));
            }
            for i in 0..3u64 {
                assert_eq!(journal.read(7 + i).await.unwrap(), test_digest(100 + i));
            }
            assert!(matches!(
                journal.read(10).await,
                Err(Error::ItemOutOfRange(10))
            ));

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

    #[test_traced]
    fn test_fixed_recovery_preserves_rolled_predecessors_without_commit() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(1));
            let mut journal = Journal::<_, Digest>::init(context.child("journal"), cfg.clone())
                .await
                .unwrap();

            // Persist a prefix, then append across multiple blob boundaries without calling
            // commit/sync. Rollover starts predecessor syncs, so the filled predecessor blobs are
            // recoverable even though the recovery watermark is not advanced.
            let appended;
            (journal, appended) = journal.append(&test_digest(10)).await.unwrap();
            assert_eq!(appended, 0);
            journal = journal.sync().await.unwrap();
            let appended;
            (journal, appended) = journal.append(&test_digest(20)).await.unwrap();
            assert_eq!(appended, 1);
            let appended;
            (journal, appended) = journal.append(&test_digest(30)).await.unwrap();
            assert_eq!(appended, 2);
            drop(journal);

            let blobs = scan_partition(&context, &blob_partition(&cfg)).await;
            assert!(
                blobs.len() > 2,
                "expected multiple empty trailing blobs, got {}",
                blobs.len()
            );

            let journal = Journal::<_, Digest>::init(context.child("recovered"), cfg.clone())
                .await
                .unwrap();
            assert_eq!(journal.bounds(), 0..3);
            assert_eq!(journal.read(0).await.unwrap(), test_digest(10));
            assert_eq!(journal.read(1).await.unwrap(), test_digest(20));
            assert_eq!(journal.read(2).await.unwrap(), test_digest(30));
            drop(journal);

            // Recovery should preserve the filled predecessors plus the empty tail.
            let blobs = scan_partition(&context, &blob_partition(&cfg)).await;
            assert_eq!(blobs.len(), 4);

            let mut journal = Journal::<_, Digest>::init(context.child("recovered"), cfg.clone())
                .await
                .unwrap();
            let appended;
            (journal, appended) = journal.append(&test_digest(42)).await.unwrap();
            assert_eq!(appended, 3);
            assert_eq!(journal.read(3).await.unwrap(), test_digest(42));
            journal.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_fixed_recovery_preserves_first_rollover_without_commit() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(1));
            let mut journal = Journal::<_, Digest>::init(context.child("journal"), cfg.clone())
                .await
                .unwrap();

            // Append across multiple blob boundaries without ever calling commit/sync. Rollover
            // starts predecessor syncs, so filled predecessors remain recoverable.
            let appended;
            (journal, appended) = journal.append(&test_digest(10)).await.unwrap();
            assert_eq!(appended, 0);
            let appended;
            (journal, appended) = journal.append(&test_digest(20)).await.unwrap();
            assert_eq!(appended, 1);
            drop(journal);

            let blobs = scan_partition(&context, &blob_partition(&cfg)).await;
            assert!(
                blobs.len() > 1,
                "expected multiple empty blobs, got {}",
                blobs.len()
            );

            let journal = Journal::<_, Digest>::init(context.child("recovered"), cfg.clone())
                .await
                .unwrap();
            assert_eq!(journal.bounds(), 0..2);
            drop(journal);

            // Recovery should preserve the filled predecessors plus the empty tail.
            let blobs = scan_partition(&context, &blob_partition(&cfg)).await;
            assert_eq!(blobs.len(), 3);

            let mut journal = Journal::<_, Digest>::init(context.child("recovered"), cfg.clone())
                .await
                .unwrap();
            let appended;
            (journal, appended) = journal.append(&test_digest(42)).await.unwrap();
            assert_eq!(appended, 2);
            assert_eq!(journal.read(2).await.unwrap(), test_digest(42));
            journal.destroy().await.unwrap();
        });
    }

    /// Recovery establishes a contiguous valid page prefix before truncating to whole items.
    #[test_traced]
    fn test_fixed_recovery_validates_pages_before_trailing_bytes() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            const LOGICAL_PAGE_SIZE: u64 = 5;

            let cfg = Config {
                partition: "fixed-validate-before-tail-trim".into(),
                items_per_blob: NZU64!(10),
                page_cache: CacheRef::new(NZU16!(LOGICAL_PAGE_SIZE as u16), NZUsize!(4)),
                write_buffer: NZUsize!(128),
                replay_buffer: NZUsize!(128),
            };
            let partition = blob_partition(&cfg);
            let (blob, size) = context.open(&partition, &0u64.to_be_bytes()).await.unwrap();
            let mut writer = PagedRecovery::open(blob, size, 128, cfg.page_cache.clone())
                .await
                .unwrap();
            let values = [11u64, 22, 33, 44];
            let mut bytes = Vec::new();
            for value in values {
                bytes.extend_from_slice(&value.to_be_bytes());
            }
            (writer, _) = writer.append(&bytes).await.unwrap();
            writer = writer.truncate(30).await.unwrap();
            writer = writer.sync().await.unwrap();
            drop(writer);

            // Five-byte integrity pages crossed by eight-byte journal items:
            //
            // pages: [0..5) [5..10) [10..15) [15..20) [20..25) [25..30)
            // state:    ok      ok       ok       ok       torn      ok
            // items: [0......8) [8.......16) [16......24) [24..30 tail)
            //
            // Backward sizing stops at valid page 5 and reports 30 logical bytes. Rounding that
            // size to a whole item selects 24, inside torn page 4. `Recovery::truncate(24)` must
            // read page 4 to preserve bytes 20..24 and rewrite its partial-page checksum, so it
            // cannot perform that truncation. Forward validation instead stops at 20, rounds down
            // to 16, and safely resizes within valid page 3.
            corrupt_page(
                &context,
                &partition,
                &0u64.to_be_bytes(),
                4,
                LOGICAL_PAGE_SIZE,
            )
            .await;

            let journal = Journal::<_, u64>::init(context.child("recover"), cfg)
                .await
                .unwrap();
            assert_eq!(journal.bounds(), 0..2);
            assert_eq!(journal.read(0).await.unwrap(), 11);
            assert_eq!(journal.read(1).await.unwrap(), 22);
            journal.destroy().await.unwrap();
        });
    }

    /// The oldest retained blob can begin after its natural start. Watermark validation must
    /// measure acknowledged bytes from that retained start before repairing a partial item.
    #[test_traced]
    fn test_fixed_recovery_watermark_uses_retained_blob_start() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(10));
            let mut journal =
                Journal::<_, u64>::init_at_size(context.child("first"), cfg.clone(), 7)
                    .await
                    .unwrap();
            for (offset, value) in [11u64, 22].into_iter().enumerate() {
                let position;
                (journal, position) = journal.append(&value).await.unwrap();
                assert_eq!(position, 7 + offset as u64);
            }
            journal = journal.sync().await.unwrap();
            drop(journal);

            let partition = blob_partition(&cfg);
            let (blob, size) = context.open(&partition, &0u64.to_be_bytes()).await.unwrap();
            let mut writer =
                PagedRecovery::open(blob, size, cfg.write_buffer.get(), cfg.page_cache.clone())
                    .await
                    .unwrap();
            assert_eq!(writer.size(), 16);
            (writer, _) = writer.append(&[0; 4]).await.unwrap();
            assert_eq!(writer.size(), 20);
            writer = writer.sync().await.unwrap();
            drop(writer);

            let journal = Journal::<_, u64>::init(context.child("recover"), cfg)
                .await
                .unwrap();
            assert_eq!(journal.bounds(), 7..9);
            assert_eq!(journal.read(7).await.unwrap(), 11);
            assert_eq!(journal.read(8).await.unwrap(), 22);
            journal.destroy().await.unwrap();
        });
    }

    /// A torn item write can persist a partial item on the tail with every page valid. The
    /// suspect scan owns the whole-item floor: recovery must truncate the fragment and leave
    /// the tail aligned for future appends.
    #[test_traced]
    fn test_fixed_recovery_truncates_partial_item_tail() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(10));
            let mut journal = Journal::<_, u64>::init(context.child("first"), cfg.clone())
                .await
                .unwrap();
            for value in [11u64, 22, 33] {
                (journal, _) = journal.append(&value).await.unwrap();
            }
            journal = journal.sync().await.unwrap();
            drop(journal);

            // Persist three bytes of a fourth item directly on the tail blob: every page stays
            // valid, only the item boundary is torn.
            let partition = blob_partition(&cfg);
            let (blob, size) = context.open(&partition, &0u64.to_be_bytes()).await.unwrap();
            let mut writer =
                Writer::new(blob, size, cfg.write_buffer.get(), cfg.page_cache.clone())
                    .await
                    .unwrap();
            assert_eq!(writer.size(), 24);
            (writer, _) = writer.append(&44u64.to_be_bytes()[..3]).await.unwrap();
            writer = writer.sync().await.unwrap();
            drop(writer);

            // Recovery floors to whole items and the journal appends aligned afterward.
            let mut journal = Journal::<_, u64>::init(context.child("recover"), cfg)
                .await
                .unwrap();
            assert_eq!(journal.bounds(), 0..3);
            assert_eq!(journal.read(2).await.unwrap(), 33);
            let position;
            (journal, position) = journal.append(&44u64).await.unwrap();
            assert_eq!(position, 3);
            let journal = journal.sync().await.unwrap();
            assert_eq!(journal.read(3).await.unwrap(), 44);
            journal.destroy().await.unwrap();
        });
    }

    /// A hole strictly between a mid-blob watermark and the blob's end truncates to whole
    /// items above the acknowledged prefix: acknowledged < recoverable floor < size, so the
    /// scan must start at the watermark rather than zero or the blob's end.
    #[test_traced]
    fn test_fixed_recovery_truncates_above_mid_blob_watermark() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            const LOGICAL_PAGE_SIZE: u64 = 5;
            let cfg = Config {
                partition: "fixed-truncate-above-watermark".into(),
                items_per_blob: NZU64!(10),
                page_cache: CacheRef::new(NZU16!(LOGICAL_PAGE_SIZE as u16), NZUsize!(4)),
                write_buffer: NZUsize!(128),
                replay_buffer: NZUsize!(128),
            };
            let mut journal = Journal::<_, u64>::init(context.child("first"), cfg.clone())
                .await
                .unwrap();
            for value in [11u64, 22] {
                (journal, _) = journal.append(&value).await.unwrap();
            }
            journal.sync().await.unwrap();

            // Persist two more items past the watermark, then tear one page beneath the
            // acknowledged prefix (the proof must skip it) and one strictly above it:
            // acknowledged (16) < recoverable floor (24) < size (32).
            let partition = blob_partition(&cfg);
            let (blob, size) = context.open(&partition, &0u64.to_be_bytes()).await.unwrap();
            let mut writer = Writer::new(blob, size, 128, cfg.page_cache.clone())
                .await
                .unwrap();
            assert_eq!(writer.size(), 16);
            let mut bytes = Vec::new();
            for value in [33u64, 44] {
                bytes.extend_from_slice(&value.to_be_bytes());
            }
            (writer, _) = writer.append(&bytes).await.unwrap();
            writer = writer.sync().await.unwrap();
            drop(writer);
            corrupt_page(
                &context,
                &partition,
                &0u64.to_be_bytes(),
                1,
                LOGICAL_PAGE_SIZE,
            )
            .await;
            corrupt_page(
                &context,
                &partition,
                &0u64.to_be_bytes(),
                5,
                LOGICAL_PAGE_SIZE,
            )
            .await;

            // The acknowledged tear is adopted (its items fail lazily at read) while the hole
            // above the watermark truncates the unacknowledged fourth item.
            let journal = Journal::<_, u64>::init(context.child("recover"), cfg)
                .await
                .unwrap();
            assert_eq!(journal.bounds(), 0..3);
            assert!(journal.read(0).await.is_err());
            assert!(journal.read(1).await.is_err());
            assert_eq!(journal.read(2).await.unwrap(), 33);
            journal.destroy().await.unwrap();
        });
    }

    /// A crash during the rollover fsync can persist a valid last page above a lost interior
    /// page, which `Writer::new`'s backward scan cannot see. Recovery must forward-validate the
    /// suspect blob and truncate at the hole.
    #[test_traced]
    fn test_fixed_recovery_truncates_torn_interior_page() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(10));
            let mut journal = Journal::<_, Digest>::init(context.child("first"), cfg.clone())
                .await
                .unwrap();
            for i in 0..15u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            journal.commit().await.unwrap();

            // Blob 0 holds 10 items (320 bytes) across 8 pages; tear page 3.
            corrupt_page(
                &context,
                &blob_partition(&cfg),
                &0u64.to_be_bytes(),
                3,
                PAGE_SIZE.get() as u64,
            )
            .await;

            // Pages 0-2 hold 132 bytes = 4 chunk-aligned items; the gap makes blob 1 unreachable.
            let mut journal = Journal::<_, Digest>::init(context.child("second"), cfg)
                .await
                .unwrap();
            assert_eq!(journal.bounds(), 0..4);
            for i in 0..4u64 {
                assert_eq!(journal.read(i).await.unwrap(), test_digest(i));
            }
            let appended;
            (journal, appended) = journal.append(&test_digest(42)).await.unwrap();
            assert_eq!(appended, 4);
            journal.destroy().await.unwrap();
        });
    }

    /// A torn interior page in the tail blob truncates the append frontier without disturbing
    /// its full predecessors.
    #[test_traced]
    fn test_fixed_recovery_truncates_torn_interior_page_in_tail() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(10));
            let mut journal = Journal::<_, Digest>::init(context.child("first"), cfg.clone())
                .await
                .unwrap();
            for i in 0..15u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            journal.commit().await.unwrap();

            // Blob 1 holds 5 items (160 bytes) across 4 pages; tear page 1.
            corrupt_page(
                &context,
                &blob_partition(&cfg),
                &1u64.to_be_bytes(),
                1,
                PAGE_SIZE.get() as u64,
            )
            .await;

            // Blob 1 keeps page 0 only: 44 bytes = 1 chunk-aligned item after blob 0's 10.
            let mut journal = Journal::<_, Digest>::init(context.child("second"), cfg)
                .await
                .unwrap();
            assert_eq!(journal.bounds(), 0..11);
            for i in 0..11u64 {
                assert_eq!(journal.read(i).await.unwrap(), test_digest(i));
            }
            let appended;
            (journal, appended) = journal.append(&test_digest(42)).await.unwrap();
            assert_eq!(appended, 11);
            journal.destroy().await.unwrap();
        });
    }

    /// A bounded open whose cap lies inside a torn page. One byte of the page's rewrite by an
    /// unsynced flush was lost while the pages after it landed, so only its shorter durable
    /// checksum slot validates. Recovery must truncate at the hole before serving the retained
    /// prefix, since reads before publication go through the page cache, which refuses a short
    /// interior page.
    #[test_traced]
    fn test_fixed_bounded_recovery_repairs_torn_rewrite_at_cap() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            const PAGE: u64 = 100;
            let cfg = Config {
                partition: "torn-rewrite-at-cap".into(),
                items_per_blob: NZU64!(64),
                page_cache: CacheRef::new(NZU16!(PAGE as u16), NZUsize!(3)),
                write_buffer: NZUsize!(2048),
                replay_buffer: NZUsize!(2048),
            };
            let mut journal = Journal::<_, Digest>::init(context.child("first"), cfg.clone())
                .await
                .unwrap();
            for i in 0..33u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }

            // Page 10 holds the acknowledged prefix's last 56 bytes, and the watermark is 33.
            journal = journal.sync().await.unwrap();
            for i in 33..38u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }

            // Write pages 10..12 without advancing the watermark, then lose one byte of page 10's
            // rewrite beyond its durable prefix.
            journal.commit().await.unwrap();
            tear_page_rewrite(
                &context,
                &blob_partition(&cfg),
                &0u64.to_be_bytes(),
                10,
                60,
                PAGE,
            )
            .await;

            // The cap's last item ends inside page 10's durable prefix.
            let checkpoint = Checkpoint::open(context.child("meta"), &cfg.partition)
                .await
                .unwrap();
            let recovery = Recovery::<_, Digest>::open(
                context.child("capped"),
                cfg.clone(),
                checkpoint,
                Some(32),
            )
            .await
            .unwrap();
            assert_eq!(recovery.bounds, 0..32);
            assert_eq!(recovery.item(31).await.unwrap(), test_digest(31));
            assert_eq!(
                authenticated::BackingRecovery::last_matching(&recovery, 32, |_| true)
                    .await
                    .unwrap(),
                32
            );
            let journal = Journal(Box::new(recovery.publish(32).await.unwrap()));
            assert_eq!(journal.bounds(), 0..32);
            assert_eq!(journal.read(31).await.unwrap(), test_digest(31));
            journal.destroy().await.unwrap();
        });
    }

    /// A torn page beneath the recovery watermark is external corruption, not a crash artifact,
    /// since the watermark only advances after the covering fsync completes. An unbounded open
    /// never re-reads acknowledged pages, so it adopts the journal unchanged and the damage
    /// surfaces as a read error on the affected items.
    #[test_traced]
    fn test_fixed_recovery_adopts_torn_page_below_watermark() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(10));
            let mut journal = Journal::<_, Digest>::init(context.child("first"), cfg.clone())
                .await
                .unwrap();
            for i in 0..15u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            journal.sync().await.unwrap();

            corrupt_page(
                &context,
                &blob_partition(&cfg),
                &0u64.to_be_bytes(),
                3,
                PAGE_SIZE.get() as u64,
            )
            .await;
            let size_before = context
                .logical_blob(&blob_partition(&cfg), &0u64.to_be_bytes())
                .unwrap()
                .len();

            let journal = Journal::<_, Digest>::init(context.child("second"), cfg.clone())
                .await
                .expect("acknowledged damage must not fail recovery");

            // Adoption must not mutate the torn blob. Items on the torn page fail lazily at
            // read while every item beyond the damaged blob remains readable.
            let size_after = context
                .logical_blob(&blob_partition(&cfg), &0u64.to_be_bytes())
                .unwrap()
                .len();
            assert_eq!(
                size_after, size_before,
                "adoption must preserve the evidence"
            );
            let mut damaged = 0;
            for i in 0..10u64 {
                match journal.read(i).await {
                    Ok(item) => assert_eq!(item, test_digest(i)),
                    Err(_) => damaged += 1,
                }
            }
            assert!(damaged > 0, "the torn page must surface as read errors");
            for i in 10..15u64 {
                assert_eq!(journal.read(i).await.unwrap(), test_digest(i));
            }
            drop(journal);

            // A retry adopts the same state without mutating it.
            let _ = Journal::<_, Digest>::init(context.child("third"), cfg.clone())
                .await
                .unwrap();
            let size_retry = context
                .logical_blob(&blob_partition(&cfg), &0u64.to_be_bytes())
                .unwrap()
                .len();
            assert_eq!(size_retry, size_before);
        });
    }

    /// Blobs wholly below the floor's blob are skipped by the interior-hole scan: a torn
    /// page in a fully acknowledged blob is adopted at init and surfaces as read errors on
    /// the affected items. The floor blob's covered prefix is pinned separately by the
    /// mid-blob adoption test.
    #[test_traced]
    fn test_fixed_recovery_skips_watermark_covered_blobs() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(10));
            let mut journal = Journal::<_, Digest>::init(context.child("first"), cfg.clone())
                .await
                .unwrap();
            for i in 0..15u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            journal.sync().await.unwrap();

            // Tear an interior page of the fully acknowledged blob 0: recovery must adopt the
            // blob without scanning it, or it would truncate acknowledged items.
            corrupt_page(
                &context,
                &blob_partition(&cfg),
                &0u64.to_be_bytes(),
                2,
                u64::from(PAGE_SIZE.get()),
            )
            .await;

            let journal = Journal::<_, Digest>::init(context.child("second"), cfg)
                .await
                .unwrap();
            assert_eq!(journal.bounds(), 0..15);
            assert_eq!(journal.read(0).await.unwrap(), test_digest(0));
            assert!(journal.read(2).await.is_err());
            assert_eq!(journal.read(14).await.unwrap(), test_digest(14));
            journal.destroy().await.unwrap();
        });
    }

    /// A torn page below a mid-blob watermark is adopted without truncating the blob's
    /// acknowledged prefix, and the damage surfaces as read errors on the affected items.
    #[test_traced]
    fn test_fixed_recovery_adopts_torn_page_below_mid_blob_watermark() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(10));
            let mut journal = Journal::<_, Digest>::init(context.child("first"), cfg.clone())
                .await
                .unwrap();
            for i in 0..15u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            // The watermark (15) sits mid-blob: blob 1 holds 5 items (160 bytes) across 4 pages.
            journal.sync().await.unwrap();

            // Tear page 1 of blob 1, beneath the acknowledged bytes ending at 160.
            corrupt_page(
                &context,
                &blob_partition(&cfg),
                &1u64.to_be_bytes(),
                1,
                PAGE_SIZE.get() as u64,
            )
            .await;
            let (_, size_before) = context
                .open(&blob_partition(&cfg), &1u64.to_be_bytes())
                .await
                .unwrap();

            for child in ["second", "retry"] {
                let journal = Journal::<_, Digest>::init(context.child(child), cfg.clone())
                    .await
                    .expect("acknowledged damage must not fail recovery");
                let mut damaged = 0;
                for i in 10..15u64 {
                    match journal.read(i).await {
                        Ok(item) => assert_eq!(item, test_digest(i)),
                        Err(_) => damaged += 1,
                    }
                }
                assert!(damaged > 0, "the torn page must surface as read errors");
                for i in 0..10u64 {
                    assert_eq!(journal.read(i).await.unwrap(), test_digest(i));
                }
            }

            // Adoption must not truncate the blob's acknowledged prefix.
            let (_, size_after) = context
                .open(&blob_partition(&cfg), &1u64.to_be_bytes())
                .await
                .unwrap();
            assert_eq!(size_after, size_before);
        });
    }

    /// A torn page containing the watermark's acknowledged boundary is the one sub-watermark
    /// shape recovery still reads: the blob no longer backs its acknowledged items, so init
    /// fails loudly. Opening may first trim the torn page as an ordinary crash tail, but
    /// retries fail identically without mutating further.
    #[test_traced]
    fn test_fixed_recovery_rejects_torn_boundary_page() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(10));
            let mut journal = Journal::<_, Digest>::init(context.child("first"), cfg.clone())
                .await
                .unwrap();
            for i in 0..15u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }

            // The watermark (15) acknowledges 160 bytes of blob 1, ending inside page 3.
            journal.sync().await.unwrap();

            // Tear page 3 of blob 1: the boundary page recovery must still validate.
            let physical_page_size = PAGE_SIZE.get() as u64 + 12;
            let offset = 3 * physical_page_size + 5;
            let (blob, _) = context
                .open(&blob_partition(&cfg), &1u64.to_be_bytes())
                .await
                .unwrap();
            let byte = blob
                .read_at(offset, 1, commonware_runtime::ReadOptions::default())
                .await
                .unwrap()
                .coalesce();
            blob.write_at(
                offset,
                vec![byte.as_ref()[0] ^ 0xFF],
                WriteOptions::default(),
            )
            .await
            .unwrap();
            blob.sync().await.unwrap();
            drop(blob);

            let result = Journal::<_, Digest>::init(context.child("second"), cfg.clone()).await;
            assert!(matches!(result, Err(Error::Corruption(_))));
            let (_, size_mid) = context
                .open(&blob_partition(&cfg), &1u64.to_be_bytes())
                .await
                .unwrap();

            // A retry fails identically and mutates nothing further.
            let result = Journal::<_, Digest>::init(context.child("retry"), cfg.clone()).await;
            assert!(matches!(result, Err(Error::Corruption(_))));
            let (_, size_after) = context
                .open(&blob_partition(&cfg), &1u64.to_be_bytes())
                .await
                .unwrap();
            assert_eq!(size_after, size_mid);
        });
    }

    /// Test that recovery preserves exactly the durable contiguous prefix when the tail blob's
    /// items were never synced: blobs 0 and 1 are durable, blob 2's items were only buffered.
    #[test_traced]
    fn test_fixed_recovery_unsynced_tail_keeps_contiguous_prefix() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(10));
            let mut journal = Journal::<_, Digest>::init(context.child("first"), cfg.clone())
                .await
                .unwrap();

            // Fill blobs 0 and 1 and partially fill blob 2 (positions 20..25). Rollover has
            // already made blobs 0 and 1 durable; blob 2's items sit in the write buffer only.
            for i in 0..25u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }

            // Wait for blobs 0 and 1 to become durable, then drop without flushing
            // blob 2's buffered items.
            {
                journal = journal.test_sync_blob(0).await.unwrap();
                journal = journal.test_sync_blob(1).await.unwrap();
            }
            drop(journal);

            // The durable data is exactly the contiguous prefix: blobs 0 and 1 hold items and
            // blob 2 is an empty trailing blob.
            let names = scan_partition(&context, &blob_partition(&cfg)).await;
            assert_eq!(names.len(), 3);
            for (blob, name) in names.iter().enumerate() {
                let (_, size) = context.open(&blob_partition(&cfg), name).await.unwrap();
                if blob < 2 {
                    assert!(size > 0, "blob {blob} should be durable");
                } else {
                    assert_eq!(size, 0, "blob {blob} should be empty");
                }
            }

            // Recovery preserves exactly the contiguous prefix 0..20.
            let mut journal = Journal::<_, Digest>::init(context.child("second"), cfg.clone())
                .await
                .unwrap();
            assert_eq!(journal.bounds(), 0..20);
            for i in 0..20u64 {
                assert_eq!(journal.read(i).await.unwrap(), test_digest(i));
            }
            assert!(matches!(
                journal.read(20).await,
                Err(Error::ItemOutOfRange(20))
            ));

            // Appends resume cleanly from the recovered boundary.
            let appended;
            (journal, appended) = journal.append(&test_digest(999)).await.unwrap();
            assert_eq!(appended, 20);
            assert_eq!(journal.read(20).await.unwrap(), test_digest(999));

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

    /// Test that a durable blob above the sync watermark, sitting beyond an empty intermediate
    /// blob, is rolled back to the contiguous boundary during recovery.
    ///
    /// Since #3790 removed the append-time sync when crossing blob boundaries, a process crash can
    /// leave a later blob incidentally durable while an earlier blob stayed buffered and was
    /// lost, producing a physical gap. Length-based recovery walks blobs from oldest and
    /// truncates at the first short non-tail blob, so the post-gap blob is discarded and only
    /// the synced prefix survives.
    #[test_traced]
    fn test_fixed_recovery_rolls_back_durable_blob_after_gap() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(10));
            let mut journal = Journal::<_, Digest>::init(context.child("first"), cfg.clone())
                .await
                .unwrap();

            // Durably commit blob 0 (positions 0..10), advancing the recovery watermark to 10.
            for i in 0..10u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            journal = journal.sync().await.unwrap();

            // Append blob 1 and part of blob 2 without committing. Then corrupt blob 1 back to
            // empty while keeping blob 2 durable, creating an external gap recovery must reject.
            for i in 10..28u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            {
                journal = journal.test_sync_blob(2).await.unwrap();
            }
            drop(journal);
            let (blob, _) = context
                .open(&blob_partition(&cfg), &1u64.to_be_bytes())
                .await
                .unwrap();
            blob.resize(0).await.unwrap();
            blob.sync().await.unwrap();
            drop(blob);

            // Durable state: blob 0 (10 items), blob 1 (empty gap), blob 2 (8 items).
            let names = scan_partition(&context, &blob_partition(&cfg)).await;
            assert_eq!(names.len(), 3);
            let mut sizes = Vec::new();
            for name in &names {
                let (_, size) = context.open(&blob_partition(&cfg), name).await.unwrap();
                sizes.push(size);
            }
            assert!(sizes[0] > 0, "blob 0 should be durable");
            assert_eq!(sizes[1], 0, "blob 1 should be the gap");
            assert!(sizes[2] > 0, "blob 2 should be incidentally durable");

            // Recovery rolls back to the watermark boundary: only the synced prefix survives and the
            // gapped blob 2 is truncated away.
            let mut journal = Journal::<_, Digest>::init(context.child("second"), cfg.clone())
                .await
                .unwrap();
            assert_eq!(journal.bounds(), 0..10);
            for i in 0..10u64 {
                assert_eq!(journal.read(i).await.unwrap(), test_digest(i));
            }
            assert!(matches!(
                journal.read(10).await,
                Err(Error::ItemOutOfRange(10))
            ));

            // The orphaned blob 2 is gone; the truncated blob 1 remains as the recovered tail.
            let names = scan_partition(&context, &blob_partition(&cfg)).await;
            assert_eq!(names.len(), 2);

            // Appends resume cleanly from the recovered boundary.
            let appended;
            (journal, appended) = journal.append(&test_digest(999)).await.unwrap();
            assert_eq!(appended, 10);
            assert_eq!(journal.read(10).await.unwrap(), test_digest(999));

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

    /// A crash right after pruning must not lose retained items that were appended but never
    /// synced. Blob removal is durable, so prune makes all data durable first: otherwise the
    /// unsynced tail would vanish with the crash and recovery would truncate the journal to
    /// empty even though the removal survived.
    #[test_traced]
    fn test_fixed_recovery_prune_crash_retains_unsynced_tail() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(10));
            let mut journal = Journal::<_, Digest>::init(context.child("first"), cfg.clone())
                .await
                .unwrap();

            // Durably persist blob 0 (positions 0..10), then append positions 10..25 across
            // blobs 1 and 2 without syncing.
            for i in 0..10u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            journal = journal.sync().await.unwrap();
            for i in 10..25u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }

            // Prune away blob 0, then crash before any sync.
            let (journal, pruned) = journal.prune(10).await.unwrap();
            assert!(pruned);
            drop(journal);

            let journal = Journal::<_, Digest>::init(context.child("second"), cfg.clone())
                .await
                .unwrap();
            assert_eq!(journal.bounds(), 10..25);
            for i in 10..25u64 {
                assert_eq!(journal.read(i).await.unwrap(), test_digest(i));
            }
            journal.destroy().await.unwrap();
        });
    }

    /// Test recovery when the oldest blob is empty but a newer blob still holds durable items.
    ///
    /// This is the fixed-journal analog of the variable-journal empty-oldest-blob gap bug. A
    /// contiguous journal can only populate a later blob after filling the earlier one, so an
    /// empty oldest blob with a populated newer blob is an orphaned gap. Length-based recovery
    /// walks from the oldest blob, finds it short (empty), and truncates everything from there,
    /// aligning the journal to empty without panicking.
    #[test_traced]
    fn test_fixed_recovery_empty_oldest_blob_orphaned_newer_blob() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(10));

            // Durably persist blobs 0 and 1 (positions 0..20).
            let mut journal = Journal::<_, Digest>::init(context.child("first"), cfg.clone())
                .await
                .unwrap();
            for i in 0..20u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            let journal = journal.sync().await.unwrap();
            drop(journal);

            // Empty the oldest blob (external corruption). The watermark (20) now exceeds the
            // recoverable size (0), which is corruption.
            let (blob0, size0) = context
                .open(&blob_partition(&cfg), &0u64.to_be_bytes())
                .await
                .unwrap();
            assert!(size0 > 0);
            blob0.resize(0).await.unwrap();
            blob0.sync().await.unwrap();

            drop(blob0);
            let result = Journal::<_, Digest>::init(context.child("second"), cfg.clone()).await;
            assert!(matches!(result, Err(Error::Corruption(_))));
        });
    }

    /// A config that stores one item per blob under the given partition.
    fn single_item_per_blob_config() -> Config {
        Config {
            partition: "single-item-per-blob".into(),
            items_per_blob: NZU64!(1),
            page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
            write_buffer: NZUsize!(2048),
            replay_buffer: NZUsize!(2048),
        }
    }

    /// With one item per blob, the tail blob is empty after every sync. Appends, tail reads,
    /// pruning, appends past the prune, and a restart all keep every retained position readable
    /// and the bounds intact.
    #[test_traced]
    fn test_single_item_per_blob() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = single_item_per_blob_config();

            // A fresh journal starts empty, and its first synced item is readable at the tail.
            let mut journal = Journal::init(context.child("first"), cfg.clone())
                .await
                .expect("failed to initialize journal");

            // Verify empty state
            let bounds = journal.bounds();
            assert_eq!(bounds.end, 0);
            assert!(bounds.is_empty());

            // Append 1 item
            let pos;
            (journal, pos) = journal
                .append(&test_digest(0))
                .await
                .expect("failed to append");
            assert_eq!(pos, 0);
            assert_eq!(journal.size(), 1);

            // Sync
            journal = journal.sync().await.expect("failed to sync");

            // Read from size() - 1
            let value = journal
                .read(journal.size() - 1)
                .await
                .expect("failed to read");
            assert_eq!(value, test_digest(0));

            // Each further append fills its own blob, and the tail stays readable at size() - 1.
            for i in 1..10u64 {
                let pos;
                (journal, pos) = journal
                    .append(&test_digest(i))
                    .await
                    .expect("failed to append");
                assert_eq!(pos, i);
                assert_eq!(journal.size(), i + 1);

                // Verify we can read the just-appended item at size() - 1
                let value = journal
                    .read(journal.size() - 1)
                    .await
                    .expect("failed to read");
                assert_eq!(value, test_digest(i));
            }

            // Verify all items can be read
            for i in 0..10u64 {
                assert_eq!(journal.read(i).await.unwrap(), test_digest(i));
            }

            // Pruning the first five positions keeps the size but moves the start. Reads below the
            // start report ItemPruned while the tail stays readable.
            journal = journal.sync().await.expect("failed to sync");
            (journal, _) = journal.prune(5).await.expect("failed to prune");

            // Size should still be 10
            assert_eq!(journal.size(), 10);

            // bounds.start should be 5
            assert_eq!(journal.bounds().start, 5);

            // Reading from size() - 1 (position 9) should still work
            let value = journal
                .read(journal.size() - 1)
                .await
                .expect("failed to read");
            assert_eq!(value, test_digest(9));

            // Reading from pruned positions should return ItemPruned
            for i in 0..5 {
                assert!(matches!(journal.read(i).await, Err(Error::ItemPruned(_))));
            }

            // Reading from retained positions should work
            for i in 5..10u64 {
                assert_eq!(journal.read(i).await.unwrap(), test_digest(i));
            }

            // Appends after the prune continue at the old size, each readable at the tail.
            for i in 10..15u64 {
                let pos;
                (journal, pos) = journal
                    .append(&test_digest(i))
                    .await
                    .expect("failed to append");
                assert_eq!(pos, i);

                // Verify we can read from size() - 1
                let value = journal
                    .read(journal.size() - 1)
                    .await
                    .expect("failed to read");
                assert_eq!(value, test_digest(i));
            }

            // A reopen recovers the size and pruned start, so every retained position reads back.
            journal.sync().await.expect("failed to sync");
            let journal = Journal::<_, Digest>::init(context.child("second"), cfg.clone())
                .await
                .expect("failed to re-initialize journal");

            // Verify size is preserved
            assert_eq!(journal.size(), 15);

            // Verify bounds.start is preserved
            assert_eq!(journal.bounds().start, 5);

            // Reading from size() - 1 should work after restart
            let value = journal
                .read(journal.size() - 1)
                .await
                .expect("failed to read");
            assert_eq!(value, test_digest(14));

            // Reading all retained positions should work
            for i in 5..15u64 {
                assert_eq!(journal.read(i).await.unwrap(), test_digest(i));
            }

            journal.destroy().await.expect("failed to destroy journal");

            // Destroy removes the journal state, so the same config reopens empty and accepts new
            // writes.
            let mut journal = Journal::<_, Digest>::init(context.child("after_destroy"), cfg)
                .await
                .expect("failed to re-initialize journal");
            assert!(journal.bounds().is_empty());
            assert_eq!(journal.size(), 0);
            for i in 0..10u64 {
                (journal, _) = journal.append(&test_digest(i + 1000)).await.unwrap();
            }
            journal = journal.sync().await.unwrap();
            assert_eq!(journal.bounds().end, 10);
            for i in 0..10u64 {
                assert_eq!(journal.read(i).await.unwrap(), test_digest(i + 1000));
            }
            journal.destroy().await.expect("failed to destroy journal");
        });
    }

    /// A one-item-per-blob journal that is pruned and reopened keeps its bounds and reads every
    /// retained position, including the tail.
    #[test_traced]
    fn test_single_item_per_blob_restart_after_prune() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = single_item_per_blob_config();

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

            // Append 10 items (positions 0-9)
            for i in 0..10u64 {
                (journal, _) = journal.append(&test_digest(i + 100)).await.unwrap();
            }

            // Prune to position 5 (removes positions 0-4)
            (journal, _) = journal.prune(5).await.unwrap();
            let bounds = journal.bounds();
            assert_eq!(bounds.end, 10);
            assert_eq!(bounds.start, 5);

            // Sync and restart
            journal.sync().await.unwrap();

            // Re-open journal
            let journal = Journal::<_, Digest>::init(context.child("fourth"), cfg.clone())
                .await
                .expect("failed to re-initialize journal");

            // Verify state after restart
            let bounds = journal.bounds();
            assert_eq!(bounds.end, 10);
            assert_eq!(bounds.start, 5);

            // Reading from size() - 1 (position 9) should work
            let value = journal.read(journal.size() - 1).await.unwrap();
            assert_eq!(value, test_digest(109));

            // Verify all retained positions (5-9) work
            for i in 5..10u64 {
                assert_eq!(journal.read(i).await.unwrap(), test_digest(i + 100));
            }

            journal.destroy().await.expect("failed to destroy journal");

            // Destroy removes the journal state, so the same config reopens empty and accepts new
            // writes.
            let mut journal = Journal::<_, Digest>::init(context.child("after_destroy"), cfg)
                .await
                .expect("failed to re-initialize journal");
            assert!(journal.bounds().is_empty());
            assert_eq!(journal.size(), 0);
            for i in 0..10u64 {
                (journal, _) = journal.append(&test_digest(i + 1000)).await.unwrap();
            }
            journal = journal.sync().await.unwrap();
            assert_eq!(journal.bounds().end, 10);
            for i in 0..10u64 {
                assert_eq!(journal.read(i).await.unwrap(), test_digest(i + 1000));
            }
            journal.destroy().await.expect("failed to destroy journal");
        });
    }

    /// Pruning every item of a one-item-per-blob journal keeps its size, empties its bounds, and
    /// reports the tail position as pruned until a new append.
    #[test_traced]
    fn test_single_item_per_blob_prune_all() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = single_item_per_blob_config();

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

            for i in 0..5u64 {
                (journal, _) = journal.append(&test_digest(i + 200)).await.unwrap();
            }
            journal = journal.sync().await.unwrap();

            // Prune all items
            (journal, _) = journal.prune(5).await.unwrap();
            let bounds = journal.bounds();
            assert_eq!(bounds.end, 5); // Size unchanged
            assert!(bounds.is_empty()); // All pruned

            // size() - 1 = 4, but position 4 is pruned
            let result = journal.read(journal.size() - 1).await;
            assert!(matches!(result, Err(Error::ItemPruned(4))));

            // After appending, reading works again
            (journal, _) = journal.append(&test_digest(205)).await.unwrap();
            assert_eq!(journal.bounds().start, 5);
            assert_eq!(
                journal.read(journal.size() - 1).await.unwrap(),
                test_digest(205)
            );

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

    #[test_traced]
    fn test_fixed_journal_init_at_size_zero() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));
            let mut journal =
                Journal::<_, Digest>::init_at_size(context.child("storage"), cfg.clone(), 0)
                    .await
                    .unwrap();

            let bounds = journal.bounds();
            assert_eq!(bounds.end, 0);
            assert!(bounds.is_empty());

            // Next append should get position 0
            let pos;
            (journal, pos) = journal.append(&test_digest(100)).await.unwrap();
            assert_eq!(pos, 0);
            assert_eq!(journal.size(), 1);
            assert_eq!(journal.read(0).await.unwrap(), test_digest(100));

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

    #[test_traced]
    fn test_fixed_journal_init_at_max_size_rejected() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let mut cfg = test_cfg(NZU64!(1));
            cfg.partition = "max-size-rejected".into();

            // A journal sized at `u64::MAX` could never accept an append, so init rejects it.
            assert!(matches!(
                Journal::<_, Digest>::init_at_size(context.child("storage"), cfg, u64::MAX).await,
                Err(Error::SizeOverflow)
            ));
        });
    }

    #[test_traced]
    fn test_fixed_journal_append_size_overflow() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let mut cfg = test_cfg(NZU64!(1));
            cfg.partition = "append-size-overflow".into();

            // Initialize one item shy of the maximum size.
            let mut journal =
                Journal::<_, Digest>::init_at_size(context.child("near_max"), cfg, u64::MAX - 1)
                    .await
                    .unwrap();

            // The first append fills the last representable position.
            let appended;
            (journal, appended) = journal.append(&test_digest(7)).await.unwrap();
            assert_eq!(appended, u64::MAX - 1);
            assert_eq!(journal.size(), u64::MAX);

            // The next append would overflow the size; it must return an error rather than
            // panicking.
            assert!(matches!(
                journal.append(&test_digest(8)).await,
                Err(Error::SizeOverflow)
            ));
        });
    }

    #[test_traced]
    fn test_fixed_journal_replay_near_max_size() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let mut cfg = test_cfg(NZU64!(10));
            cfg.partition = "replay-near-max-size".into();

            let mut journal =
                Journal::<_, Digest>::init_at_size(context.child("near_max"), cfg, u64::MAX - 1)
                    .await
                    .unwrap();
            let expected = test_digest(7);
            let appended;
            (journal, appended) = journal.append(&expected).await.unwrap();
            assert_eq!(appended, u64::MAX - 1);

            {
                let reader;
                (journal, reader) = journal.snapshot().await.unwrap();
                let stream = replay_items(
                    &reader,
                    u64::MAX - 1..reader.bounds().end,
                    NZUsize!(1024),
                    ReadOptions::default(),
                )
                .await
                .unwrap();
                pin_mut!(stream);
                let (pos, item) = stream.next().await.unwrap().unwrap();
                assert_eq!(pos, u64::MAX - 1);
                assert_eq!(item, expected);
                assert!(stream.next().await.is_none());
            }

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

    #[test_traced]
    fn test_fixed_journal_init_at_size_blob_boundary() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));

            // Initialize at position 10 (exactly at blob 2 boundary with items_per_blob=5)
            let mut journal =
                Journal::<_, Digest>::init_at_size(context.child("storage"), cfg.clone(), 10)
                    .await
                    .unwrap();

            let bounds = journal.bounds();
            assert_eq!(bounds.end, 10);
            assert!(bounds.is_empty());

            // Next append should get position 10
            let pos;
            (journal, pos) = journal.append(&test_digest(1000)).await.unwrap();
            assert_eq!(pos, 10);
            assert_eq!(journal.size(), 11);
            assert_eq!(journal.read(10).await.unwrap(), test_digest(1000));

            // Can continue appending
            let pos;
            (journal, pos) = journal.append(&test_digest(1001)).await.unwrap();
            assert_eq!(pos, 11);
            assert_eq!(journal.read(11).await.unwrap(), test_digest(1001));

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

    #[test_traced]
    fn test_fixed_journal_init_at_size_mid_blob() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));

            // Initialize at position 7 (middle of blob 1 with items_per_blob=5)
            let mut journal =
                Journal::<_, Digest>::init_at_size(context.child("storage"), cfg.clone(), 7)
                    .await
                    .unwrap();

            let bounds = journal.bounds();
            assert_eq!(bounds.end, 7);
            // No data exists yet after init_at_size
            assert!(bounds.is_empty());

            // Reading before bounds.start should return ItemPruned
            assert!(matches!(journal.read(5).await, Err(Error::ItemPruned(5))));
            assert!(matches!(journal.read(6).await, Err(Error::ItemPruned(6))));

            // Next append should get position 7
            let pos;
            (journal, pos) = journal.append(&test_digest(700)).await.unwrap();
            assert_eq!(pos, 7);
            assert_eq!(journal.size(), 8);
            assert_eq!(journal.read(7).await.unwrap(), test_digest(700));
            // Now bounds.start should be 7 (first data position)
            assert_eq!(journal.bounds().start, 7);

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

    #[test_traced]
    fn test_fixed_journal_append_many_after_mid_blob_start() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(100));
            let mut journal =
                Journal::<_, Digest>::init_at_size(context.child("first"), cfg.clone(), 150)
                    .await
                    .unwrap();

            let items: Vec<_> = (0..100u64).map(|i| test_digest(1500 + i)).collect();
            let last;
            (journal, last) = journal.append_many(Many::Flat(&items)).await.unwrap();
            assert_eq!(last, 249);
            assert_eq!(journal.bounds(), 150..250);

            for (position, index) in [(150, 0), (199, 49), (200, 50), (249, 99)] {
                assert_eq!(
                    journal.read(position).await.unwrap(),
                    items[index],
                    "item at position {position} did not match"
                );
            }

            journal.sync().await.unwrap();

            let journal = Journal::<_, Digest>::init(context.child("second"), cfg.clone())
                .await
                .unwrap();
            assert_eq!(journal.bounds(), 150..250);
            for (position, index) in [(150, 0), (199, 49), (200, 50), (249, 99)] {
                assert_eq!(
                    journal.read(position).await.unwrap(),
                    items[index],
                    "item at position {position} did not match after reopen"
                );
            }

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

    #[test_traced]
    fn test_fixed_journal_init_at_size_persistence() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));

            // Initialize at position 15
            let mut journal =
                Journal::<_, Digest>::init_at_size(context.child("first"), cfg.clone(), 15)
                    .await
                    .unwrap();

            // Append some items
            for i in 0..5u64 {
                let pos;
                (journal, pos) = journal.append(&test_digest(1500 + i)).await.unwrap();
                assert_eq!(pos, 15 + i);
            }

            assert_eq!(journal.size(), 20);

            // Sync and reopen
            journal.sync().await.unwrap();

            let mut journal = Journal::<_, Digest>::init(context.child("second"), cfg.clone())
                .await
                .unwrap();

            // Size and data should be preserved
            let bounds = journal.bounds();
            assert_eq!(bounds.end, 20);
            assert_eq!(bounds.start, 15);

            // Verify data
            for i in 0..5u64 {
                assert_eq!(journal.read(15 + i).await.unwrap(), test_digest(1500 + i));
            }

            // Can continue appending
            let pos;
            (journal, pos) = journal.append(&test_digest(9999)).await.unwrap();
            assert_eq!(pos, 20);
            assert_eq!(journal.read(20).await.unwrap(), test_digest(9999));

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

    #[test_traced]
    fn test_fixed_journal_init_at_size_persistence_without_data() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));

            // Initialize at position 15
            let journal =
                Journal::<_, Digest>::init_at_size(context.child("first"), cfg.clone(), 15)
                    .await
                    .unwrap();

            let bounds = journal.bounds();
            assert_eq!(bounds.end, 15);
            assert!(bounds.is_empty());

            // Drop without writing any data
            drop(journal);

            // Reopen and verify size persisted
            let mut journal = Journal::<_, Digest>::init(context.child("second"), cfg.clone())
                .await
                .unwrap();

            let bounds = journal.bounds();
            assert_eq!(bounds.end, 15);
            assert!(bounds.is_empty());

            // Can append starting at position 15
            let pos;
            (journal, pos) = journal.append(&test_digest(1500)).await.unwrap();
            assert_eq!(pos, 15);
            assert_eq!(journal.read(15).await.unwrap(), test_digest(1500));

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

    #[test_traced]
    fn test_fixed_journal_init_at_size_large_offset() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));

            // Initialize at a large position (position 1000)
            let mut journal =
                Journal::<_, Digest>::init_at_size(context.child("storage"), cfg.clone(), 1000)
                    .await
                    .unwrap();

            let bounds = journal.bounds();
            assert_eq!(bounds.end, 1000);
            assert!(bounds.is_empty());

            // Next append should get position 1000
            let pos;
            (journal, pos) = journal.append(&test_digest(100000)).await.unwrap();
            assert_eq!(pos, 1000);
            assert_eq!(journal.read(1000).await.unwrap(), test_digest(100000));

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

    #[test_traced]
    fn test_fixed_journal_init_at_size_prune_and_append() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));

            // Initialize at position 20
            let mut journal =
                Journal::<_, Digest>::init_at_size(context.child("storage"), cfg.clone(), 20)
                    .await
                    .unwrap();

            // Append items 20-29
            for i in 0..10u64 {
                (journal, _) = journal.append(&test_digest(2000 + i)).await.unwrap();
            }

            assert_eq!(journal.size(), 30);

            // Prune to position 25
            (journal, _) = journal.prune(25).await.unwrap();

            let bounds = journal.bounds();
            assert_eq!(bounds.end, 30);
            assert_eq!(bounds.start, 25);

            // Verify remaining items are readable
            for i in 25..30u64 {
                assert_eq!(journal.read(i).await.unwrap(), test_digest(2000 + (i - 20)));
            }

            // Continue appending
            let pos;
            (journal, pos) = journal.append(&test_digest(3000)).await.unwrap();
            assert_eq!(pos, 30);

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

    #[test_traced]
    fn test_fixed_journal_clear_to_size() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(10));
            let mut journal = Journal::init(context.child("journal"), cfg.clone())
                .await
                .expect("failed to initialize journal");

            // Append 25 items (positions 0-24, spanning 3 blobs)
            for i in 0..25u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            assert_eq!(journal.size(), 25);
            journal = journal.sync().await.unwrap();

            // Clear to position 100, effectively resetting the journal
            journal.0 = journal.0.clear_to_size(100).await.unwrap();
            assert_eq!(journal.size(), 100);

            // Old positions should fail
            for i in 0..25 {
                assert!(matches!(journal.read(i).await, Err(Error::ItemPruned(_))));
            }

            // Verify size persists after restart without writing any data
            drop(journal);
            let mut journal =
                Journal::<_, Digest>::init(context.child("journal_after_clear"), cfg.clone())
                    .await
                    .expect("failed to re-initialize journal after clear");
            assert_eq!(journal.size(), 100);

            // Append new data starting at position 100
            for i in 100..105u64 {
                let pos;
                (journal, pos) = journal.append(&test_digest(i)).await.unwrap();
                assert_eq!(pos, i);
            }
            assert_eq!(journal.size(), 105);

            // New positions should be readable
            for i in 100..105u64 {
                assert_eq!(journal.read(i).await.unwrap(), test_digest(i));
            }

            // Sync and re-init to verify persistence
            journal.sync().await.unwrap();

            let journal = Journal::<_, Digest>::init(context.child("journal_reopened"), cfg)
                .await
                .expect("failed to re-initialize journal");

            assert_eq!(journal.size(), 105);
            for i in 100..105u64 {
                assert_eq!(journal.read(i).await.unwrap(), test_digest(i));
            }

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

    #[test_traced]
    fn test_fixed_journal_clear_to_size_rejects_max() {
        // `clear_to_size` must reject `u64::MAX` like `init_at_size`: such a journal could never
        // accept an append.
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(10));
            let journal = Journal::<_, Digest>::init(context.child("journal"), cfg.clone())
                .await
                .unwrap();
            assert!(matches!(
                journal.0.clear_to_size(u64::MAX).await,
                Err(Error::SizeOverflow)
            ));
            // The failed clear consumed the journal. Reopen to exercise the staging path.
            let journal = Journal::<_, Digest>::init(context.child("journal_intent"), cfg)
                .await
                .unwrap();
            assert!(matches!(
                journal.0.stage_clear_intent(u64::MAX).await,
                Err(Error::SizeOverflow)
            ));
        });
    }

    #[test_traced]
    fn test_fixed_journal_clear_to_size_rejects_backward_without_mutation() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(10));
            let mut journal = Journal::<_, Digest>::init(context.child("first"), cfg.clone())
                .await
                .unwrap();
            for i in 0..2 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            let journal = journal.sync().await.unwrap();

            assert!(matches!(
                journal.0.clear_to_size(1).await,
                Err(Error::ItemOutOfRange(1))
            ));

            let journal = Journal::<_, Digest>::init(context.child("after_clear"), cfg.clone())
                .await
                .unwrap();
            assert_eq!(journal.bounds(), 0..2);
            assert_eq!(journal.read(0).await.unwrap(), test_digest(0));
            assert_eq!(journal.read(1).await.unwrap(), test_digest(1));
            assert!(matches!(
                journal.0.stage_clear_intent(1).await,
                Err(Error::ItemOutOfRange(1))
            ));

            let journal = Journal::<_, Digest>::init(context.child("after_intent"), cfg)
                .await
                .unwrap();
            assert_eq!(journal.bounds(), 0..2);
            assert_eq!(journal.read(0).await.unwrap(), test_digest(0));
            assert_eq!(journal.read(1).await.unwrap(), test_digest(1));
            journal.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_fixed_journal_sync_crash_meta_none_boundary_aligned() {
        // Old meta = None (aligned), new boundary = aligned.
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));
            let mut journal = Journal::<_, Digest>::init(context.child("first"), cfg.clone())
                .await
                .unwrap();

            for i in 0..5u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            journal.commit().await.unwrap();

            let journal = Journal::<_, Digest>::init(context.child("second"), cfg.clone())
                .await
                .unwrap();
            let bounds = journal.bounds();
            assert_eq!(bounds.start, 0);
            assert_eq!(bounds.end, 5);
            journal.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_fixed_journal_missing_metadata_with_short_blob_is_corruption() {
        // Clearing all metadata leaves no watermark. Recovery falls back to the blob boundary
        // and finds a short non-tail blob, violating the legacy rollover-sync invariant.
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));
            let mut journal =
                Journal::<_, Digest>::init_at_size(context.child("first"), cfg.clone(), 7)
                    .await
                    .unwrap();
            for i in 0..3u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            let mut journal = journal.sync().await.unwrap();

            // Simulate metadata deletion (corruption).
            journal.0.checkpoint.clear();
            journal.0.checkpoint = journal.0.checkpoint.sync().await.unwrap();
            drop(journal);

            let result = Journal::<_, Digest>::init(context.child("second"), cfg.clone()).await;
            assert!(matches!(result, Err(Error::Corruption(_))));
        });
    }

    #[test_traced]
    fn test_fixed_journal_sync_crash_meta_mid_boundary_unchanged() {
        // Old meta = Some(mid), new boundary = mid-blob (same value).
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));
            let mut journal =
                Journal::<_, Digest>::init_at_size(context.child("first"), cfg.clone(), 7)
                    .await
                    .unwrap();
            for i in 0..3u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            journal.commit().await.unwrap();

            let journal = Journal::<_, Digest>::init(context.child("second"), cfg.clone())
                .await
                .unwrap();
            let bounds = journal.bounds();
            assert_eq!(bounds.start, 7);
            assert_eq!(bounds.end, 10);
            journal.destroy().await.unwrap();
        });
    }
    #[test_traced]
    fn test_fixed_journal_sync_crash_meta_mid_to_aligned_becomes_stale() {
        // Old meta = Some(mid), new boundary = aligned.
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));
            let mut journal =
                Journal::<_, Digest>::init_at_size(context.child("first"), cfg.clone(), 7)
                    .await
                    .unwrap();
            for i in 0..10u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            assert_eq!(journal.size(), 17);
            (journal, _) = journal.prune(10).await.unwrap();

            journal.commit().await.unwrap();

            let journal = Journal::<_, Digest>::init(context.child("second"), cfg.clone())
                .await
                .unwrap();
            let bounds = journal.bounds();
            assert_eq!(bounds.start, 10);
            assert_eq!(bounds.end, 17);
            journal.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_fixed_journal_prune_does_not_move_boundary_backwards() {
        // Pruning to a position earlier than pruning_boundary (within the same blob)
        // should not move the boundary backwards.
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));
            // init_at_size(7) sets pruning_boundary = 7 (mid-blob in blob 1)
            let mut journal =
                Journal::<_, Digest>::init_at_size(context.child("first"), cfg.clone(), 7)
                    .await
                    .unwrap();
            // Append 5 items at positions 7-11, filling blob 1 and part of blob 2
            for i in 0..5u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            // Prune to position 5 (blob 1 start) should NOT move boundary back from 7 to 5
            let (journal, _) = journal.prune(5).await.unwrap();
            assert_eq!(journal.bounds().start, 7);
            journal.destroy().await.unwrap();
        });
    }

    /// A prune that returns true has made every pre-prune item durable: a crash immediately
    /// after must recover the full pre-prune size even when every unsynced write is lost.
    #[test_traced]
    fn test_fixed_journal_prune_durability_survives_crash() {
        let executor = deterministic::Runner::default();
        let (_, checkpoint) = executor.start_and_recover(|context| async move {
            let cfg = test_cfg(NZU64!(3));
            let mut journal = Journal::<_, Digest>::init(context.child("first"), cfg)
                .await
                .unwrap();

            // Fill two blobs plus an unsynced tail, then prune into blob 1. The crash
            // drops every write not covered by a completed sync, so the prune's internal
            // sync is the only durability point covering these items.
            for i in 0..8u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            let (journal, pruned) = journal.prune(3).await.unwrap();
            assert!(pruned);
            drop(journal);
        });

        deterministic::Runner::from(checkpoint).start(|context| async move {
            let cfg = test_cfg(NZU64!(3));
            let journal = Journal::<_, Digest>::init(context.child("recover"), cfg)
                .await
                .unwrap();
            assert_eq!(
                journal.bounds(),
                3..8,
                "pruned journal lost acknowledged items"
            );
            for i in 3..8u64 {
                assert_eq!(journal.read(i).await.unwrap(), test_digest(i));
            }
            journal.destroy().await.unwrap();
        });
    }

    /// Prune makes all data durable before removing blobs, so a commit issued right after must
    /// not attempt to sync the removed blobs.
    #[test_traced]
    fn test_fixed_journal_commit_after_prune() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));
            let mut journal = Journal::<_, Digest>::init(context.child("journal"), cfg.clone())
                .await
                .unwrap();

            for i in 0..12 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }

            let (mut journal, _) = journal.prune(5).await.unwrap();
            journal = journal
                .commit()
                .await
                .expect("commit should not try to sync pruned blobs");
            assert_eq!(journal.bounds(), 5..12);
            journal.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_fixed_journal_replay_after_init_at_size_spanning_blobs() {
        // Test replay when first blob begins mid-blob: init_at_size creates a journal
        // where pruning_boundary is mid-blob, then we append across multiple blobs.
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));

            // Initialize at position 7 (mid-blob with items_per_blob=5)
            // Blob 1 (positions 5-9) begins mid-blob: only positions 7, 8, 9 have data
            let mut journal =
                Journal::<_, Digest>::init_at_size(context.child("storage"), cfg.clone(), 7)
                    .await
                    .unwrap();

            // Append 13 items (positions 7-19), spanning blobs 1, 2, 3
            for i in 0..13u64 {
                let pos;
                (journal, pos) = journal.append(&test_digest(100 + i)).await.unwrap();
                assert_eq!(pos, 7 + i);
            }
            assert_eq!(journal.size(), 20);
            journal = journal.sync().await.unwrap();

            // Replay from pruning_boundary
            {
                let reader;
                (journal, reader) = journal.snapshot().await.unwrap();
                let stream = replay_items(
                    &reader,
                    7..reader.bounds().end,
                    NZUsize!(1024),
                    ReadOptions::default(),
                )
                .await
                .expect("failed to replay");
                pin_mut!(stream);
                let mut items: Vec<(u64, Digest)> = Vec::new();
                while let Some(result) = stream.next().await {
                    items.push(result.expect("replay item failed"));
                }

                // Should get all 13 items with correct logical positions
                assert_eq!(items.len(), 13);
                for (i, (pos, item)) in items.iter().enumerate() {
                    assert_eq!(*pos, 7 + i as u64);
                    assert_eq!(*item, test_digest(100 + i as u64));
                }
            }

            // Replay from mid-stream (position 12)
            {
                let reader;
                (journal, reader) = journal.snapshot().await.unwrap();
                let stream = replay_items(
                    &reader,
                    12..reader.bounds().end,
                    NZUsize!(1024),
                    ReadOptions::default(),
                )
                .await
                .expect("failed to replay from mid-stream");
                pin_mut!(stream);
                let mut items: Vec<(u64, Digest)> = Vec::new();
                while let Some(result) = stream.next().await {
                    items.push(result.expect("replay item failed"));
                }

                // Should get items from position 12 onwards
                assert_eq!(items.len(), 8);
                for (i, (pos, item)) in items.iter().enumerate() {
                    assert_eq!(*pos, 12 + i as u64);
                    assert_eq!(*item, test_digest(100 + 5 + i as u64));
                }
            }

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

    #[test_traced]
    fn test_fixed_journal_init_at_most_before_retained_start() {
        // A bound below the retained start must be rejected.
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));

            let mut journal =
                Journal::<_, Digest>::init_at_size(context.child("storage"), cfg.clone(), 10)
                    .await
                    .unwrap();

            // Append a few items (positions 10, 11, 12)
            for i in 0..3u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            assert_eq!(journal.size(), 13);

            // A bound of 11 retains one item.
            journal = {
                _ = journal.sync().await.unwrap();
                Journal::<_, Digest>::init_at_most(context.child("cap"), cfg.clone(), 11).await
            }
            .unwrap();
            assert_eq!(journal.size(), 11);

            // The retained start is a valid empty prefix.
            journal = {
                _ = journal.sync().await.unwrap();
                Journal::<_, Digest>::init_at_most(context.child("cap"), cfg.clone(), 10).await
            }
            .unwrap();
            assert_eq!(journal.size(), 10);

            // Truncate to before pruning_boundary should fail
            let result = {
                _ = journal.sync().await.unwrap();
                Journal::<_, Digest>::init_at_most(context.child("cap"), cfg.clone(), 9).await
            };
            assert!(matches!(result, Err(Error::ItemPruned(9))));
        });
    }

    #[test_traced]
    fn test_fixed_journal_init_at_size_crash_scenarios() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));

            // Setup: Create a journal with some data and mid-blob metadata
            let mut journal =
                Journal::<_, Digest>::init_at_size(context.child("first"), cfg.clone(), 7)
                    .await
                    .unwrap();
            for i in 0..5u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            let journal = journal.sync().await.unwrap();
            drop(journal);

            // Crash Scenario 1: after clear intent is synced and blobs are removed, but before
            // the new tail blob is created.
            let blob_part = blob_partition(&cfg);
            let mut checkpoint = Checkpoint::open(context.child("intent_meta"), &cfg.partition)
                .await
                .unwrap();
            checkpoint.set_clear_target(12);
            let checkpoint = checkpoint.sync().await.unwrap();
            drop(checkpoint);
            context.remove(&blob_part, None).await.unwrap();

            // Recovery should complete the interrupted init_at_size(12).
            let journal = Journal::<_, Digest>::init(
                context.child("crash").with_attribute("index", 1),
                cfg.clone(),
            )
            .await
            .expect("init failed after clear crash");
            let bounds = journal.bounds();
            assert_eq!(bounds.end, 12);
            assert_eq!(bounds.start, 12);
            drop(journal);

            // Restore metadata for next scenario (it might have been removed by init)
            let mut checkpoint = Checkpoint::open(context.child("restore_meta"), &cfg.partition)
                .await
                .unwrap();
            checkpoint.set_boundary_hint(7);
            checkpoint.set_clear_target(2);
            let checkpoint = checkpoint.sync().await.unwrap();
            drop(checkpoint);

            // Crash Scenario 2: after the new tail blob is created, but before final metadata
            // replaces the clear intent.
            let (blob, _) = context.open(&blob_part, &0u64.to_be_bytes()).await.unwrap();
            blob.sync().await.unwrap(); // Ensure it exists
            drop(blob);

            // Recovery should complete the interrupted init_at_size(2).
            let journal = Journal::<_, Digest>::init(
                context.child("crash").with_attribute("index", 2),
                cfg.clone(),
            )
            .await
            .expect("init failed after create crash");

            let bounds = journal.bounds();
            assert_eq!(bounds.start, 2);
            assert_eq!(bounds.end, 2);
            journal.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_fixed_journal_clear_to_size_crash_scenarios() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));

            // Setup: Init at 12 (Blob 2, offset 2)
            // Metadata = 12
            let journal =
                Journal::<_, Digest>::init_at_size(context.child("first"), cfg.clone(), 12)
                    .await
                    .unwrap();
            let journal = journal.sync().await.unwrap();
            drop(journal);

            // Crash Scenario: clear_to_size(2) after the intent is synced and blob 0 is created,
            // but before final metadata replaces the clear intent.

            let blob_part = blob_partition(&cfg);
            let mut checkpoint = Checkpoint::open(context.child("meta"), &cfg.partition)
                .await
                .unwrap();
            checkpoint.set_clear_target(2);
            let checkpoint = checkpoint.sync().await.unwrap();
            drop(checkpoint);

            context.remove(&blob_part, None).await.unwrap();

            let (blob, _) = context.open(&blob_part, &0u64.to_be_bytes()).await.unwrap();
            blob.sync().await.unwrap();

            let journal = Journal::<_, Digest>::init(context.child("crash_clear"), cfg.clone())
                .await
                .expect("init failed after clear_to_size crash");

            let bounds = journal.bounds();
            assert_eq!(bounds.start, 2);
            assert_eq!(bounds.end, 2);
            journal.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_fixed_journal_clear_to_size_crash_after_intent_before_blobs() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));
            let mut journal = Journal::<_, Digest>::init(context.child("first"), cfg.clone())
                .await
                .unwrap();
            for i in 0..12u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            journal = journal.sync().await.unwrap();

            drop(journal);
            let mut checkpoint = Checkpoint::open(context.child("meta"), &cfg.partition)
                .await
                .unwrap();
            checkpoint.set_clear_target(100);
            let checkpoint = checkpoint.sync().await.unwrap();
            drop(checkpoint);

            let mut journal = Journal::<_, Digest>::init(context.child("second"), cfg.clone())
                .await
                .expect("init failed after clear intent crash");
            assert_eq!(journal.bounds(), 100..100);
            let pos;
            (journal, pos) = journal.append(&test_digest(100)).await.unwrap();
            assert_eq!(pos, 100);
            journal.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_fixed_journal_clear_intent_skips_corrupt_stale_blobs() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));
            let blob_part = blob_partition(&cfg);
            let mut checkpoint = Checkpoint::open(context.child("meta"), &cfg.partition)
                .await
                .unwrap();
            checkpoint.set_clear_target(12);
            let checkpoint = checkpoint.sync().await.unwrap();
            drop(checkpoint);

            // This name would fail `Partition::open_many` if init tried to parse stale blobs before
            // honoring the clear intent.
            let (blob, _) = context.open(&blob_part, b"not-u64").await.unwrap();
            blob.write_at(0, vec![1, 2, 3], WriteOptions::SYNC)
                .await
                .unwrap();
            drop(blob);

            let journal = Journal::<_, Digest>::init(context.child("recover"), cfg.clone())
                .await
                .expect("clear intent should discard stale corrupt blobs before blob parsing");
            assert_eq!(journal.bounds(), 12..12);
            assert_eq!(journal.0.recovery_watermark(), 12);
            journal.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_fixed_journal_clear_to_size_crash_after_mid_blob_intent_with_old_blobs_present() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(10));
            let mut journal =
                Journal::<_, Digest>::init_at_size(context.child("first"), cfg.clone(), 10)
                    .await
                    .unwrap();

            for i in 0..6u64 {
                let pos;
                (journal, pos) = journal.append(&test_digest(i)).await.unwrap();
                assert_eq!(pos, 10 + i);
            }
            journal = journal.sync().await.unwrap();

            drop(journal);
            let mut checkpoint = Checkpoint::open(context.child("meta"), &cfg.partition)
                .await
                .unwrap();
            checkpoint.set_clear_target(15);
            let checkpoint = checkpoint.sync().await.unwrap();
            drop(checkpoint);

            let journal = Journal::<_, Digest>::init(context.child("second"), cfg.clone())
                .await
                .expect("init failed after mid-blob clear intent crash");
            assert_eq!(journal.bounds(), 15..15);
            drop(journal);

            let mut journal = Journal::<_, Digest>::init(context.child("third"), cfg.clone())
                .await
                .expect("init failed after completing mid-blob clear intent");
            assert_eq!(journal.bounds(), 15..15);
            assert!(matches!(journal.read(14).await, Err(Error::ItemPruned(14))));
            let pos;
            (journal, pos) = journal.append(&test_digest(100)).await.unwrap();
            assert_eq!(pos, 15);
            assert_eq!(journal.read(15).await.unwrap(), test_digest(100));
            journal.destroy().await.unwrap();
        });
    }

    #[rstest::rstest]
    #[case::lower(7)]
    #[case::same(50)]
    #[case::higher(60)]
    #[test_traced]
    fn test_fixed_journal_init_sync_after_interrupted_clear(#[case] start: u64) {
        // State sync is the first open after a clear crashed with its intent staged. The cases
        // start below, at, and above the staged target 50.
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(10));

            // Sync 30 items across three blobs.
            let mut journal = Journal::<_, Digest>::init(context.child("first"), cfg.clone())
                .await
                .unwrap();
            for i in 0..30u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            journal = journal.sync().await.unwrap();
            drop(journal);

            // Stage a clear to 50 and crash before any blob is removed. The blobs still hold all
            // 30 synced items.
            Journal::<_, Digest>::test_stage_clear(context.child("intent"), &cfg.partition, 50)
                .await
                .unwrap();

            // A staged clear spans only its target. Any other start replaces the intent without
            // recovering the stale blobs. A start at 50 completes it.
            let mut journal = authenticated::init_sync::<_, Journal<_, Digest>>(
                context.child("sync"),
                cfg.clone(),
                start..start + 20,
            )
            .await
            .expect("init_sync must complete or replace the staged clear");
            assert_eq!(journal.bounds(), start..start);

            // Appends begin at the sync start. Digests of 100 onward differ from every stale item,
            // so the reads below tell new items from stale ones.
            for i in 0..3u64 {
                let pos;
                (journal, pos) = journal.append(&test_digest(100 + i)).await.unwrap();
                assert_eq!(pos, start + i);
            }
            journal = journal.sync().await.unwrap();
            drop(journal);

            // Reopen with an ordinary init. A surviving intent would clear the new items, and stale
            // blobs would surface below the start or in place of the new items.
            let journal = Journal::<_, Digest>::init(context.child("reopen"), cfg.clone())
                .await
                .unwrap();
            assert_eq!(journal.bounds(), start..start + 3);

            // For start 7, the pruned position 6 held a stale item before the clear.
            assert!(matches!(
                journal.read(start - 1).await,
                Err(Error::ItemPruned(pos)) if pos == start - 1
            ));
            for i in 0..3u64 {
                assert_eq!(journal.read(start + i).await.unwrap(), test_digest(100 + i));
            }
            journal.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_fixed_journal_rejects_watermark_with_aligned_empty_tail() {
        // Watermark beyond the recovered size with an aligned pruning boundary.
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));

            let mut journal = Journal::<_, Digest>::init(context.child("first"), cfg.clone())
                .await
                .unwrap();
            for i in 0..10u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            let journal = journal.sync().await.unwrap();
            drop(journal);

            // Remove all blobs and create a single empty blob 1, leaving
            // recovery_watermark=10 in metadata.
            let blob_part = blob_partition(&cfg);
            context.remove(&blob_part, None).await.unwrap();
            let (blob, _) = context.open(&blob_part, &1u64.to_be_bytes()).await.unwrap();
            blob.sync().await.unwrap();

            drop(blob);
            let result = Journal::<_, Digest>::init(context.child("crash"), cfg.clone()).await;
            assert!(matches!(result, Err(Error::Corruption(_))));
        });
    }

    #[test_traced]
    fn test_fixed_journal_rejects_far_watermark_with_aligned_empty_tail() {
        // Same as above but the watermark is multiple blobs past the empty tail.
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));

            let mut journal = Journal::<_, Digest>::init(context.child("first"), cfg.clone())
                .await
                .unwrap();
            for i in 0..10u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            let journal = journal.sync().await.unwrap();
            drop(journal);

            // Remove all blobs and create a single empty blob 0, leaving
            // recovery_watermark=10 in metadata.
            let blob_part = blob_partition(&cfg);
            context.remove(&blob_part, None).await.unwrap();
            let (blob, _) = context.open(&blob_part, &0u64.to_be_bytes()).await.unwrap();
            blob.sync().await.unwrap();

            drop(blob);
            let result = Journal::<_, Digest>::init(context.child("crash"), cfg.clone()).await;
            assert!(matches!(result, Err(Error::Corruption(_))));
        });
    }

    #[test_traced]
    fn test_read_many_empty() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(10));
            let journal = Journal::<_, Digest>::init(context.child("j"), cfg)
                .await
                .unwrap();

            let (journal, reader) = journal.snapshot().await.unwrap();
            let items = reader.read_many(&[]).await.unwrap();
            assert!(items.is_empty());

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

    #[test_traced]
    fn test_read_many_single_blob() {
        // All positions within one blob.
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(10));
            let mut journal = Journal::init(context.child("j"), cfg).await.unwrap();

            for i in 0..5u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            assert_eq!(journal.size(), 5);

            let (journal, reader) = journal.snapshot().await.unwrap();
            let items = reader.read_many(&[0, 2, 4]).await.unwrap();
            assert_eq!(items, vec![test_digest(0), test_digest(2), test_digest(4)]);

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

    #[test_traced]
    #[should_panic(expected = "positions must be strictly increasing")]
    fn test_read_many_rejects_unsorted_positions() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(10));
            let mut journal = Journal::init(context.child("j"), cfg).await.unwrap();
            for i in 0..5u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }

            let (_journal, reader) = journal.snapshot().await.unwrap();
            let _ = reader.read_many(&[2, 1]).await;
        });
    }

    #[test_traced]
    #[should_panic(expected = "positions must be strictly increasing")]
    fn test_read_many_rejects_duplicate_positions() {
        // Duplicates are not strictly increasing either.
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(10));
            let mut journal = Journal::init(context.child("j"), cfg).await.unwrap();
            for i in 0..5u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }

            let (_journal, reader) = journal.snapshot().await.unwrap();
            let _ = reader.read_many(&[1, 1]).await;
        });
    }

    #[test_traced]
    fn test_read_many_across_blobs() {
        // Positions spanning multiple blobs (items_per_blob=3).
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(3));
            let mut journal = Journal::init(context.child("j"), cfg).await.unwrap();

            for i in 0..9u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            assert_eq!(journal.size(), 9);
            // Blobs: [0,1,2], [3,4,5], [6,7,8]

            let (journal, reader) = journal.snapshot().await.unwrap();
            let items = reader.read_many(&[1, 4, 7]).await.unwrap();
            assert_eq!(items, vec![test_digest(1), test_digest(4), test_digest(7)]);

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

    #[test_traced]
    fn test_read_many_after_prune() {
        // Read from positions that survive pruning.
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(3));
            let mut journal = Journal::init(context.child("j"), cfg).await.unwrap();

            for i in 0..9u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            assert_eq!(journal.size(), 9);
            journal = journal.sync().await.unwrap();

            // Prune first blob [0,1,2].
            (journal, _) = journal.prune(3).await.unwrap();
            assert_eq!(journal.bounds(), 3..9);

            let (journal, reader) = journal.snapshot().await.unwrap();
            let items = reader.read_many(&[3, 5, 8]).await.unwrap();
            assert_eq!(items, vec![test_digest(3), test_digest(5), test_digest(8)]);

            // Pruned position should error.
            let (journal, reader) = journal.snapshot().await.unwrap();
            let err = reader.read_many(&[1]).await.unwrap_err();
            assert!(matches!(err, Error::ItemPruned(1)));

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

    #[test_traced]
    fn test_read_many_out_of_range() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(10));
            let mut journal = Journal::init(context.child("j"), cfg).await.unwrap();

            for i in 0..3u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            assert_eq!(journal.size(), 3);

            let (journal, reader) = journal.snapshot().await.unwrap();
            let err = reader.read_many(&[0, 5]).await.unwrap_err();
            assert!(matches!(err, Error::ItemOutOfRange(5)));

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

    #[test_traced]
    fn test_read_many_matches_read() {
        // Verify batch read matches individual reads across blobs.
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(4));
            let mut journal = Journal::init(context.child("j"), cfg).await.unwrap();

            for i in 0..20u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            assert_eq!(journal.size(), 20);
            journal = journal.sync().await.unwrap();

            let positions: Vec<u64> = (0..20).collect();
            let reader;
            (journal, reader) = journal.snapshot().await.unwrap();
            let batch = reader.read_many(&positions).await.unwrap();

            for &pos in &positions {
                let single = reader.read(pos).await.unwrap();
                assert_eq!(batch[pos as usize], single);
            }
            drop(reader);

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

    #[test_traced]
    fn test_bounded_initialization_cached_reads() {
        deterministic::Runner::default().start(|context| async move {
            let cfg = test_cfg(NZU64!(4));
            let mut journal = Journal::<_, Digest>::init(context.child("seed"), cfg.clone())
                .await
                .unwrap();
            for i in 0..20 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            _ = journal.sync().await.unwrap();
            let journal = Journal::<_, Digest>::init_at_most(context.child("bounded"), cfg, 18)
                .await
                .unwrap();
            let (journal, reader) = journal.snapshot().await.unwrap();
            assert_eq!(reader.read(17).await.unwrap(), test_digest(17));
            let served = reader.try_read_many_sync(&[17, 18]);
            assert_eq!(served[0], Some(test_digest(17)));
            assert!(served[1].is_none());
            drop(reader);
            journal.destroy().await.unwrap();
        });
    }

    #[test_traced]
    fn test_try_read_many_sync_matches_read_many() {
        // Cached positions are served synchronously and match the async batched read.
        // Positions that fail validation are misses rather than errors.
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(4));
            let page_cache = cfg.page_cache.clone();
            let mut journal = Journal::init(context.child("j"), cfg.clone())
                .await
                .unwrap();

            for i in 0..20u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            journal = journal.sync().await.unwrap();

            let positions: Vec<u64> = (0..20).collect();
            let reader;
            (journal, reader) = journal.snapshot().await.unwrap();
            let expected = reader.read_many(&positions).await.unwrap();

            // Every synchronously served item must match the async read. Positions the
            // 3-page test cache cannot hold are misses, never wrong values.
            let served = reader.try_read_many_sync(&positions);
            assert_eq!(served.len(), positions.len());
            for (item, expected) in served.iter().zip(&expected) {
                if let Some(item) = item {
                    assert_eq!(item, expected);
                }
            }

            // Consume tail-page history, then fill the cache with unrelated single-page reads.
            // The synchronous-hit checks must hold without favorable prior admission state.
            page_cache.clear();
            reader
                .read_many(&(16..20).collect::<Vec<_>>())
                .await
                .unwrap();
            page_cache.clear();
            for pos in [0, 4, 8] {
                reader.read(pos).await.unwrap();
            }

            // Positions 18 and 19 span one cached page and the in-memory tail, so warming them
            // makes both available synchronously.
            let tail: Vec<u64> = (18..20).collect();
            reader.read_many(&tail).await.unwrap();
            let served = reader.try_read_many_sync(&tail);
            for (item, pos) in served.iter().zip(&tail) {
                assert_eq!(
                    item.as_ref().expect("warmed position is served"),
                    &expected[*pos as usize]
                );
            }

            // A long-evicted position is a miss.
            assert!(reader.try_read_many_sync(&[0])[0].is_none());

            // An out-of-range position is a miss, not an error, and does not poison the
            // valid position grouped before it.
            let served = reader.try_read_many_sync(&[19, 20]);
            assert!(served[0].is_some());
            assert!(served[1].is_none());
            drop(served);
            drop(reader);

            // After a truncate the journal's end is not blob-aligned, so an out-of-range
            // position can share a blob with a valid one. Validation trims the batch
            // instead of poisoning the shared group.
            journal = {
                _ = journal.sync().await.unwrap();
                Journal::<_, Digest>::init_at_most(context.child("cap"), cfg.clone(), 18).await
            }
            .unwrap();
            let reader;
            (journal, reader) = journal.snapshot().await.unwrap();
            reader.read_many(&[17]).await.unwrap();
            let served = reader.try_read_many_sync(&[17, 18]);
            assert!(served[0].is_some());
            assert!(served[1].is_none());
            drop(served);
            drop(reader);

            // A pruned position is trimmed from the prefix rather than reaching offset
            // derivation, and the valid single-page remainder is still served.
            (journal, _) = journal.prune(8).await.unwrap();
            let reader;
            (journal, reader) = journal.snapshot().await.unwrap();
            reader.read_many(&[8]).await.unwrap();
            let served = reader.try_read_many_sync(&[3, 8]);
            assert!(served[0].is_none());
            assert!(served[1].is_some());
            drop(served);
            drop(reader);

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

    #[test_traced]
    fn test_probe_then_read_many_matches_read_many() {
        // A probe completed by one batched read over its declined positions returns the same
        // items as read_many, cold and warm.
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(4));
            let mut journal = Journal::init(context.child("j"), cfg).await.unwrap();

            for i in 0..20u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            journal = journal.sync().await.unwrap();

            let positions: Vec<u64> = (0..20).collect();
            let reader;
            (journal, reader) = journal.snapshot().await.unwrap();
            let expected: Vec<_> = (0..20).map(test_digest).collect();
            for _ in 0..2 {
                let mut served = reader.try_read_many_sync(&positions);
                let misses: Vec<u64> = positions
                    .iter()
                    .zip(&served)
                    .filter_map(|(&pos, item)| item.is_none().then_some(pos))
                    .collect();
                let mut fetched = reader.read_many(&misses).await.unwrap().into_iter();
                for item in served.iter_mut().filter(|item| item.is_none()) {
                    *item = fetched.next();
                }
                let completed: Vec<_> = served.into_iter().map(Option::unwrap).collect();
                assert_eq!(completed, expected);
            }
            assert_eq!(reader.read_many(&positions).await.unwrap(), expected);
            drop(reader);

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

    #[test_traced]
    fn test_fixed_journal_metrics() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(2));
            let mut journal =
                Journal::<_, Digest>::init(context.child("fixed_metrics"), cfg.clone())
                    .await
                    .unwrap();

            let items: Vec<_> = (0..5).map(test_digest).collect();
            (journal, _) = journal.append_many(Many::Flat(&items)).await.unwrap();
            (journal, _) = journal.append(&test_digest(5)).await.unwrap();
            journal = journal.commit().await.unwrap();
            journal = journal.sync().await.unwrap();
            let handle;
            (journal, handle) = journal.start_sync().await.unwrap();
            handle.await.unwrap();
            let (journal, reader) = journal.snapshot().await.unwrap();
            reader.read(0).await.unwrap();
            let (journal, reader) = journal.snapshot().await.unwrap();
            reader.try_read_sync(0).unwrap();
            let (journal, reader) = journal.snapshot().await.unwrap();
            reader.read_many(&[1, 2, 4]).await.unwrap();
            let (journal, _) = journal.prune(2).await.unwrap();

            let buffer = context.encode();
            for expected in [
                "fixed_metrics_size 6",
                "fixed_metrics_pruning_boundary 2",
                "fixed_metrics_retained 4",
                "fixed_metrics_tail_items 2",
                "fixed_metrics_append_calls_total 1",
                "fixed_metrics_append_many_calls_total 1",
                "fixed_metrics_read_calls_total 1",
                "fixed_metrics_read_many_calls_total 1",
                "fixed_metrics_items_read_total 5",
                "fixed_metrics_start_sync_calls_total 1",
                "fixed_metrics_commit_calls_total 1",
                "fixed_metrics_sync_calls_total 1",
                "fixed_metrics_append_duration_count 1",
                "fixed_metrics_append_many_duration_count 1",
                "fixed_metrics_read_duration_count 0",
                "fixed_metrics_read_many_duration_count 1",
                "fixed_metrics_commit_duration_count 1",
                "fixed_metrics_sync_duration_count 1",
                "fixed_metrics_cache_hits_total",
                "fixed_metrics_cache_misses_total",
                "fixed_metrics_blobs_tracked",
            ] {
                assert!(buffer.contains(expected), "{expected}\n{buffer}");
            }

            journal.destroy().await.unwrap();
        });
    }
    /// A snapshot's bounds and contents are frozen across appends and rolls.
    #[test_traced]
    fn test_snapshot_frozen_across_roll() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));
            let mut journal = Journal::<_, Digest>::init(context.child("j"), cfg)
                .await
                .unwrap();
            for i in 0..7u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }

            let snapshot;
            (journal, snapshot) = journal.snapshot().await.unwrap();
            assert_eq!(snapshot.bounds(), 0..7);

            // Appending past the blob boundary rolls the snapshot's tail blob into
            // history; the snapshot keeps reading it through its own handle.
            for i in 7..23u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            assert_eq!(snapshot.bounds(), 0..7);
            for i in 0..7u64 {
                assert_eq!(snapshot.read(i).await.unwrap(), test_digest(i));
            }
            assert!(matches!(
                snapshot.read(7).await,
                Err(Error::ItemOutOfRange(7))
            ));

            let fresh;
            (journal, fresh) = journal.snapshot().await.unwrap();
            assert_eq!(fresh.bounds(), 0..23);
            assert_eq!(fresh.read(22).await.unwrap(), test_digest(22));

            drop(snapshot);
            drop(fresh);
            journal.destroy().await.unwrap();
        });
    }

    /// A snapshot taken before a prune keeps reading the pruned range; later snapshots observe
    /// the new boundary.
    #[test_traced]
    fn test_prune_under_snapshot() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));
            let mut journal = Journal::<_, Digest>::init(context.child("j"), cfg)
                .await
                .unwrap();
            for i in 0..17u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            journal = journal.sync().await.unwrap();

            let snapshot;
            (journal, snapshot) = journal.snapshot().await.unwrap();
            let pruned;
            (journal, pruned) = journal.prune(12).await.unwrap();
            assert!(pruned);

            // The straggler reads the pruned range through its own handles.
            assert_eq!(snapshot.bounds(), 0..17);
            for i in 0..17u64 {
                assert_eq!(snapshot.read(i).await.unwrap(), test_digest(i));
            }

            let fresh;
            (journal, fresh) = journal.snapshot().await.unwrap();
            assert_eq!(fresh.bounds(), 10..17);
            assert!(matches!(fresh.read(3).await, Err(Error::ItemPruned(3))));

            drop(snapshot);
            drop(fresh);
            journal.destroy().await.unwrap();
        });
    }

    /// Every snapshot shipped to a concurrent task is fully readable while the writer keeps
    /// appending and rolling.
    #[test_traced]
    fn test_snapshots_readable_during_concurrent_appends() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));
            let mut journal = Journal::<_, Digest>::init(context.child("j"), cfg)
                .await
                .unwrap();

            let (mut tx, mut rx) =
                futures::channel::mpsc::channel::<Reader<'static, Context, Digest>>(8);
            let validator = context.child("validator").spawn(|_| async move {
                let mut validated = 0usize;
                while let Some(snapshot) = rx.next().await {
                    let bounds = snapshot.bounds();
                    for i in bounds.clone() {
                        assert_eq!(snapshot.read(i).await.unwrap(), test_digest(i));
                    }
                    validated += (bounds.end - bounds.start) as usize;
                }
                validated
            });

            for i in 0..40u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
                if i % 7 == 0 {
                    let snapshot;
                    (journal, snapshot) = journal.snapshot().await.unwrap();
                    if tx.try_send(snapshot).is_err() {
                        break;
                    }
                }
            }
            drop(tx);
            assert!(validator.await.unwrap() > 0);

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

    /// A snapshot taken before rolls and a prune replays its full frozen range, streaming its
    /// then-tail blob through the snapshot's own handle.
    #[test_traced]
    fn test_replay_from_stale_snapshot() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = test_cfg(NZU64!(5));
            let mut journal = Journal::<_, Digest>::init(context.child("j"), cfg)
                .await
                .unwrap();
            for i in 0..7u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }

            // Positions 5..7 live in the snapshot's tail blob.
            let snapshot;
            (journal, snapshot) = journal.snapshot().await.unwrap();
            assert_eq!(snapshot.bounds(), 0..7);

            // Roll the snapshot's tail into history, then prune both of its blobs away.
            for i in 7..23u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            let pruned;
            (journal, pruned) = journal.prune(12).await.unwrap();
            assert!(pruned);

            {
                let stream = replay_items(
                    &snapshot,
                    0..snapshot.bounds().end,
                    NZUsize!(1024),
                    ReadOptions::default(),
                )
                .await
                .unwrap();
                pin_mut!(stream);
                let mut expected = 0u64;
                while let Some(result) = stream.next().await {
                    let (pos, item) = result.unwrap();
                    assert_eq!(pos, expected);
                    assert_eq!(item, test_digest(pos));
                    expected += 1;
                }
                assert_eq!(expected, 7);
            }

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

    #[test_traced]
    fn test_read_many_sparse_sections_and_hit_accounting() {
        // Verify the batched read path is byte-identical to per-item reads across multiple
        // blobs, with a mid-blob pruning boundary, a sparse subset of positions, and
        // exact hit/miss accounting over a mixed cached/uncached batch.
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let mut cfg = test_cfg(NZU64!(8));
            // Keep the whole batch resident so hit accounting is stable. Otherwise, the batch may
            // evict a page that a later per-item probe still expects to hit.
            cfg.page_cache = CacheRef::new(PAGE_SIZE, NZUsize!(16));
            let mut journal = Journal::init(context.child("j"), cfg).await.unwrap();

            for i in 0..50u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            journal = journal.sync().await.unwrap();
            // Prune mid-blob so first_in_blob differs from the blob start.
            (journal, _) = journal.prune(11).await.unwrap();

            let reader;
            (journal, reader) = journal.snapshot().await.unwrap();

            // Sparse subset spanning multiple blobs, including the pruning boundary.
            // `try_read_sync` probes do not populate the cache, so the cached subset is
            // whatever the append path left resident; derive the expected hit count from
            // probes so the batch read's hit/miss accounting is asserted exactly.
            let positions: Vec<u64> = vec![11, 12, 19, 20, 23, 31, 40, 47, 49];
            let expected_hits = positions
                .iter()
                .filter(|&&pos| reader.try_read_sync(pos).is_some())
                .count() as u64;
            let before = context.encode();
            let batch = reader.read_many(&positions).await.unwrap();
            let after = context.encode();
            assert_eq!(batch.len(), positions.len());
            assert_eq!(
                counter(&after, "cache_hits") - counter(&before, "cache_hits"),
                expected_hits,
                "batch read hit count should match the cached subset"
            );
            assert_eq!(
                counter(&after, "cache_misses") - counter(&before, "cache_misses"),
                positions.len() as u64 - expected_hits,
                "batch read miss count should cover the rest"
            );
            for (i, &pos) in positions.iter().enumerate() {
                let single = reader.read(pos).await.unwrap();
                assert_eq!(batch[i], single);
                assert_eq!(batch[i], test_digest(pos));
            }

            // Full contiguous range over retained items.
            let all: Vec<u64> = (11..50).collect();
            let batch = reader.read_many(&all).await.unwrap();
            for (i, &pos) in all.iter().enumerate() {
                assert_eq!(batch[i], reader.read(pos).await.unwrap());
            }
            drop(reader);

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

    #[test_traced]
    fn test_read_many_cold_blob_groups() {
        // Read a batch whose positions are all cache misses spread over four blobs, so the whole
        // batch is served by per-blob group reads into disjoint slices of the shared output
        // buffer. Each digest encodes its position, so a group read into the wrong slice fails
        // the value assertions.
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let mut cfg = test_cfg(NZU64!(8));
            cfg.page_cache = CacheRef::new(PAGE_SIZE, NZUsize!(32));
            let mut journal = Journal::<_, Digest>::init(context.child("first"), cfg.clone())
                .await
                .unwrap();
            for i in 0..40u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            let journal = journal.sync().await.unwrap();
            drop(journal);

            // Reopen with a fresh page cache so every full page is cold. The positions avoid
            // each blob's trailing bytes, which sealed blobs keep in memory and always serve
            // synchronously.
            cfg.page_cache = CacheRef::new(PAGE_SIZE, NZUsize!(32));
            let mut journal = Journal::<_, Digest>::init(context.child("second"), cfg)
                .await
                .unwrap();
            let reader;
            (journal, reader) = journal.snapshot().await.unwrap();
            let positions = [1, 5, 10, 14, 17, 22, 25, 30];
            for pos in positions {
                assert!(reader.try_read_sync(pos).is_none(), "position {pos}");
            }

            // Every item requires a blob read, split into four groups of two.
            let before = context.encode();
            let batch = reader.read_many(&positions).await.unwrap();
            let after = context.encode();
            assert_eq!(
                counter(&after, "second_cache_misses") - counter(&before, "second_cache_misses"),
                positions.len() as u64
            );
            assert_eq!(
                counter(&after, "second_cache_hits"),
                counter(&before, "second_cache_hits")
            );
            for (i, &pos) in positions.iter().enumerate() {
                assert_eq!(batch[i], test_digest(pos), "position {pos}");
            }

            // The first pass cached every page it faulted, so a second pass is all hits and must
            // return the same items.
            let before = context.encode();
            let batch = reader.read_many(&positions).await.unwrap();
            let after = context.encode();
            assert_eq!(
                counter(&after, "second_cache_hits") - counter(&before, "second_cache_hits"),
                positions.len() as u64
            );
            for (i, &pos) in positions.iter().enumerate() {
                assert_eq!(batch[i], test_digest(pos), "position {pos}");
            }
            drop(reader);

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

    #[test_traced]
    fn test_fixed_journal_read_miss_timed() {
        // Reads served from storage record a read_duration sample; cache hits do not.
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let mut journal =
                Journal::<_, Digest>::init(context.child("miss"), test_cfg(NZU64!(2)))
                    .await
                    .unwrap();
            for i in 0..20 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            journal = journal.sync().await.unwrap();

            // The page cache cannot hold every page, so some position must be cold.
            let reader;
            (journal, reader) = journal.snapshot().await.unwrap();
            let pos = (0..20)
                .find(|&pos| reader.try_read_sync(pos).is_none())
                .expect("some position should be cold");
            assert_eq!(reader.read(pos).await.unwrap(), test_digest(pos));
            drop(reader);

            let buffer = context.encode();
            assert!(buffer.contains("miss_read_duration_count 1"), "{buffer}");

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

    /// A failed watermark advance that nobody observed surfaces at destroy before any data blob
    /// is removed, so the journal reopens with every durable item.
    #[test_traced]
    fn test_fixed_destroy_observes_failed_watermark_sync_before_removal() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let pending = PendingSyncs::default();
            let cfg = test_cfg(NZU64!(100));
            let data_partition = blob_partition(&cfg);
            let mut journal = Box::new(
                Inner::<_, u64>::init(
                    DelayedSyncContext {
                        inner: context.child("journal"),
                        pending: pending.clone(),
                    },
                    cfg.clone(),
                )
                .await
                .unwrap(),
            );
            (journal, _) = journal.append_many(Many::Flat(&[1, 2, 3])).await.unwrap();
            journal = drive_pending_syncs(&pending, journal.sync()).await.unwrap();

            // Make item 4 durable without advancing the watermark, then start the advance to 4,
            // drop its handle, and fail its checkpoint sync, the only sync that advance starts.
            (journal, _) = journal.append(&4).await.unwrap();
            let (journal, handle) = journal.start_sync().await.unwrap();
            release_pending_syncs(&pending);
            handle.await.unwrap();
            let (journal, handle) = journal.start_sync().await.unwrap();
            drop(handle);
            assert_eq!(pending.lock().len(), 1);
            fail_pending_syncs(&pending);

            assert!(journal.destroy().await.is_err());
            assert_eq!(
                scan_partition(&context, &data_partition).await,
                vec![0u64.to_be_bytes().to_vec()]
            );
            let journal = Inner::<_, u64>::init(context.child("second"), cfg)
                .await
                .unwrap();
            assert_eq!(journal.size(), 4);
            journal.destroy().await.unwrap();
        });
    }

    /// A bounded open whose cap lies below the watermark, both inside a torn page, repairs the
    /// page at the end of its durable prefix. The acknowledged items survive, the persisted
    /// watermark is untouched, and a later open with a larger cap or none recovers all of them.
    #[test_traced]
    fn test_fixed_bounded_recovery_below_watermark_keeps_acknowledged_items() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            const PAGE: u64 = 100;
            let cfg = Config {
                partition: "torn-rewrite-below-watermark".into(),
                items_per_blob: NZU64!(64),
                page_cache: CacheRef::new(NZU16!(PAGE as u16), NZUsize!(3)),
                write_buffer: NZUsize!(2048),
                replay_buffer: NZUsize!(2048),
            };
            let mut journal = Journal::<_, Digest>::init(context.child("first"), cfg.clone())
                .await
                .unwrap();
            for i in 0..33u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            journal = journal.sync().await.unwrap();
            for i in 33..38u64 {
                (journal, _) = journal.append(&test_digest(i)).await.unwrap();
            }
            journal.commit().await.unwrap();
            tear_page_rewrite(
                &context,
                &blob_partition(&cfg),
                &0u64.to_be_bytes(),
                10,
                60,
                PAGE,
            )
            .await;

            // Cap 32 and watermark 33 both end inside torn page 10.
            let checkpoint = Checkpoint::open(context.child("meta"), &cfg.partition)
                .await
                .unwrap();
            let recovery = Recovery::<_, Digest>::open(
                context.child("capped"),
                cfg.clone(),
                checkpoint,
                Some(32),
            )
            .await
            .unwrap();
            assert_eq!(recovery.bounds, 0..32);
            assert_eq!(recovery.item(31).await.unwrap(), test_digest(31));
            drop(recovery);
            assert_eq!(
                Journal::<_, Digest>::persisted_watermark(
                    context.child("watermark"),
                    &cfg.partition
                )
                .await
                .unwrap(),
                Some(33)
            );

            // A larger cap and no cap both recover the 33 acknowledged items.
            let checkpoint = Checkpoint::open(context.child("wider_meta"), &cfg.partition)
                .await
                .unwrap();
            let recovery = Recovery::<_, Digest>::open(
                context.child("wider"),
                cfg.clone(),
                checkpoint,
                Some(36),
            )
            .await
            .unwrap();
            assert_eq!(recovery.bounds, 0..33);
            assert_eq!(recovery.item(32).await.unwrap(), test_digest(32));
            drop(recovery);
            let journal = Journal::<_, Digest>::init(context.child("unbounded"), cfg)
                .await
                .unwrap();
            assert_eq!(journal.bounds(), 0..33);
            for i in 0..33u64 {
                assert_eq!(journal.read(i).await.unwrap(), test_digest(i));
            }
            journal.destroy().await.unwrap();
        });
    }
}