Skip to main content

commonware_storage/journal/segmented/
oversized.rs

1//! Segmented journal for oversized values.
2//!
3//! This module combines [super::fixed::Journal] with [super::glob::Glob] to handle
4//! entries that reference variable-length "oversized" values. It provides coordinated
5//! operations and built-in crash recovery.
6//!
7//! # Architecture
8//!
9//! ```text
10//! +-------------------+     +-------------------+
11//! | Fixed Journal     |     | Glob (Values)     |
12//! | (Index Entries)   |     |                   |
13//! +-------------------+     +-------------------+
14//! | entry_0           | --> | value_0           |
15//! | entry_1           | --> | value_1           |
16//! | ...               |     | ...               |
17//! +-------------------+     +-------------------+
18//! ```
19//!
20//! Each index entry contains `(value_offset, value_size)` pointing to its value in glob.
21//!
22//! # Crash Recovery
23//!
24//! On unclean shutdown, the index journal and glob may have different lengths:
25//! - Index entry pointing to non-existent glob data (dangerous)
26//! - Glob value without index entry (orphan - acceptable but cleaned up)
27//! - Glob sections without corresponding index sections (orphan sections - removed)
28//!
29//! During initialization, crash recovery is performed:
30//! 1. Each section's last valid entry is found by scanning backwards: an entry is valid
31//!    only if its glob reference is in bounds (`value_offset + value_size <= glob_size`)
32//!    and its value's checksum verifies
33//! 2. Entries beyond the last valid one are skipped and the index journal is rewound
34//! 3. Orphan value sections (sections in glob but not in index) are removed
35//!
36//! This allows async writes (glob first, then index) while ensuring consistency
37//! after recovery: a trailing run of entries that became durable ahead of their value
38//! bytes is rewound at the next init, whether the glob is short (range check) or covers
39//! the ranges with garbage (checksum check). Entries below the last valid one are kept
40//! without reading their values (monotonically increasing offsets make them range-valid),
41//! so their checksums are verified lazily at `get_value()`, which can fail for a kept
42//! entry if the underlying storage is corrupted. Rewinds (including the truncations
43//! recovery itself performs) make both journals' truncations durable before returning,
44//! so neither a dropped index entry nor the stale bytes it referenced can survive a
45//! crash once later appends may reuse the freed offsets.
46//!
47//! When a checkpoint is provided ([Oversized::init_with_checkpoint]), recovery restores the
48//! state instead of inferring one: each section below the checkpoint is adopted at its
49//! validated terminal boundary (without reading values), the checkpointed section is
50//! durably truncated to the committed size, and everything after it is removed. A
51//! missing or damaged durable boundary fails init rather than being repaired. Other
52//! committed damage the checkpoint covers surfaces lazily as read errors.
53//!
54//! Tracked recovery persists a per-section committed item count. Entries below the marker are
55//! adopted once their index/value boundary is proven, entries above it are value-verified in
56//! order, and the first invalid value truncates the section's remainder. Markers trail proven
57//! syncs, publishing once a section is durable and idle (or on an empty flush).
58
59use super::{
60    fixed::{
61        Config as FixedConfig, Journal as FixedJournal, RecoveryPreflight, Replay as FixedReplay,
62    },
63    glob::{Config as GlobConfig, Glob},
64};
65use crate::{
66    Context, SyncCompletion,
67    journal::{Error, durability::Barrier},
68    metadata::{Config as MetadataConfig, Metadata},
69};
70use commonware_codec::{Codec, CodecFixed, CodecShared};
71use commonware_runtime::{Error as RError, Handle, ReadOptions};
72use commonware_utils::sequence::U64 as SectionKey;
73use futures::{FutureExt as _, future::try_join};
74use std::{
75    collections::{BTreeMap, BTreeSet, HashSet},
76    num::NonZeroUsize,
77};
78use tracing::{debug, warn};
79
80/// Trait for index entries that reference oversized values in glob storage.
81///
82/// Implementations must provide access to the value location for crash recovery validation,
83/// and a way to set the location when appending.
84pub trait Record: CodecFixed<Cfg = ()> + Clone {
85    /// Returns `(value_offset, value_size)` for crash recovery validation.
86    fn value_location(&self) -> (u64, u32);
87
88    /// Returns a new entry with the value location set.
89    ///
90    /// Called during `append` after the value is written to glob storage.
91    fn with_location(self, offset: u64, size: u32) -> Self;
92}
93
94/// Configuration for oversized journal.
95#[derive(Clone)]
96pub struct Config<C> {
97    /// Partition for the fixed index journal.
98    pub index_partition: String,
99
100    /// Partition for the glob value storage.
101    pub value_partition: String,
102
103    /// Page cache for index journal caching.
104    pub index_page_cache: commonware_runtime::buffer::paged::CacheRef,
105
106    /// Write buffer size for the index journal.
107    pub index_write_buffer: NonZeroUsize,
108
109    /// Write buffer size for the value journal.
110    pub value_write_buffer: NonZeroUsize,
111
112    /// Buffer size for sequential index recovery.
113    pub replay_buffer: NonZeroUsize,
114
115    /// Optional compression level for values (using zstd).
116    pub compression: Option<u8>,
117
118    /// Codec configuration for values.
119    pub codec_config: C,
120}
121
122/// Recovery contract applied while opening the index and value journals.
123///
124/// Exactly one mode establishes their shared boundary: `Restore` uses an explicit checkpoint,
125/// `Floors` preserves per-section validated prefixes while repairing any suffix, and `Infer`
126/// derives the boundary entirely from journal contents.
127enum Recovery<'a> {
128    Restore { section: u64, index_size: u64 },
129    Floors(&'a BTreeMap<u64, u64>),
130    Infer,
131}
132
133/// Durable recovery state for a journal that validates every uncommitted value during replay.
134struct Tracking<E: Context> {
135    /// Durable committed item count for each retained section.
136    metadata: Metadata<E, SectionKey, u64>,
137
138    /// Completion of the marker generation currently being persisted.
139    marker_sync_pending: Option<SyncCompletion>,
140
141    /// Joint index/value durability proofs not yet fully published as markers.
142    barriers: BTreeMap<u64, Barrier>,
143}
144
145impl<E: Context> Tracking<E> {
146    /// Stage a changed marker while preserving absence as the canonical zero boundary.
147    fn stage_marker(&mut self, section: u64, floor: u64) -> bool {
148        let key = SectionKey::new(section);
149        match self.metadata.get(&key) {
150            None if floor == 0 => false,
151            Some(stored) if *stored == floor => false,
152            _ => {
153                self.metadata.put(key, floor);
154                true
155            }
156        }
157    }
158
159    /// Return the section's barrier, seeding a replacement at its staged floor.
160    ///
161    /// A staged floor never exceeds durably synced data, so it is the newest boundary a
162    /// replacement barrier may claim without observing a completed sync.
163    fn barrier(&mut self, section: u64) -> &mut Barrier {
164        let floor = self
165            .metadata
166            .get(&SectionKey::new(section))
167            .copied()
168            .unwrap_or(0);
169        self.barriers
170            .entry(section)
171            .or_insert_with(|| Barrier::new(floor))
172    }
173
174    /// Observe an in-flight marker without blocking and discard proofs it published.
175    ///
176    /// Returns whether a marker generation is still in flight.
177    fn observe_marker_sync(&mut self) -> Result<bool, Error> {
178        let Some(completion) = self.marker_sync_pending.as_mut() else {
179            return Ok(false);
180        };
181        let Some(result) = completion.now_or_never() else {
182            return Ok(true);
183        };
184        result.map_err(|err| Error::Metadata(crate::metadata::Error::Runtime(err)))?;
185        self.marker_sync_pending = None;
186
187        // Retire only barriers whose proof is fully published. A barrier still awaiting a
188        // sync outcome protects a boundary beyond its marker and must survive to observe it.
189        let metadata = &self.metadata;
190        self.barriers.retain(|section, barrier| {
191            let published = metadata
192                .get(&SectionKey::new(*section))
193                .copied()
194                .unwrap_or(0);
195            barrier.boundary() > published || !barrier.settled()
196        });
197        Ok(false)
198    }
199}
200
201/// State for the one marker-aware replay performed while opening a tracked journal.
202struct Validation {
203    /// Cursor state for the section currently being replayed.
204    current_section: Option<u64>,
205    floor: u64,
206    truncated: bool,
207
208    /// First invalid position in each section, applied after replay releases the index journal.
209    rewinds: Vec<(u64, u64)>,
210
211    /// Whether replay yielded an error that makes the journal unavailable.
212    failed: bool,
213}
214
215impl Validation {
216    const fn new() -> Self {
217        Self {
218            current_section: None,
219            floor: 0,
220            truncated: false,
221            rewinds: Vec::new(),
222            failed: false,
223        }
224    }
225}
226
227/// Segmented journal for entries with oversized values.
228///
229/// Combines a fixed-size index journal with glob storage for variable-length values.
230/// Provides coordinated operations and crash recovery.
231///
232/// Mutating functions consume the journal and return it only on success: an error (or a dropped
233/// future) destroys the handle. [Oversized::replay] consumes the journal into an owned [Replay]
234/// reader, which returns it via [Replay::finish] once exhausted. Mutations on pruned sections
235/// fail with [Error::AlreadyPrunedToSection]. Check [Oversized::pruned] first to keep the
236/// handle.
237pub struct Oversized<E: Context, I: Record, V: Codec> {
238    index: FixedJournal<E, I>,
239    values: Glob<E, V>,
240    tracking: Option<Tracking<E>>,
241}
242
243impl<E: Context, I: Record + Send + Sync, V: CodecShared> std::fmt::Debug for Oversized<E, I, V> {
244    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
245        f.debug_struct("Oversized")
246            .field("oldest_section", &self.oldest_section())
247            .field("newest_section", &self.newest_section())
248            .finish_non_exhaustive()
249    }
250}
251
252impl<E: Context, I: Record + Send + Sync, V: CodecShared> Oversized<E, I, V> {
253    /// Initialize with inferred crash recovery.
254    ///
255    /// Recovery infers the durable state: it finds each section's last valid entry (in
256    /// bounds of the glob, checksum-verified) and rewinds the index journal to exclude
257    /// the entries beyond it.
258    pub async fn init(context: E, cfg: Config<V::Cfg>) -> Result<Self, Error> {
259        let replay_buffer = cfg.replay_buffer;
260        let journal = Self::init_inner(context, cfg, Recovery::Infer).await?;
261        journal.recover_inferred(replay_buffer).await
262    }
263
264    /// Initialize with crash recovery restoring a durable checkpoint, as
265    /// `(section, index size)`.
266    ///
267    /// Recovery keeps exactly the checkpointed state: each section below `section` is
268    /// adopted at its validated terminal boundary, `section` is truncated to `index
269    /// size`, and everything after it is removed. A missing or damaged boundary the
270    /// checkpoint covers fails init with [Error::Corruption], while interior damage
271    /// below a boundary surfaces lazily as read errors. Callers must only provide a
272    /// checkpoint that was durably synced before it was published (see
273    /// [crate::freezer::Freezer]).
274    pub async fn init_with_checkpoint(
275        context: E,
276        cfg: Config<V::Cfg>,
277        checkpoint: (u64, u64),
278    ) -> Result<Self, Error> {
279        let (section, index_size) = checkpoint;
280        Self::init_inner(
281            context,
282            cfg,
283            Recovery::Restore {
284                section,
285                index_size,
286            },
287        )
288        .await
289    }
290
291    /// Initialize tracked recovery and return its required full replay.
292    ///
293    /// The caller must drain the replay and call [Replay::finish_tracked]. Entries below each
294    /// durable marker are retained after their cross-journal boundary is proven. Entries above it
295    /// are value-validated in order and the first invalid entry truncates its section.
296    pub async fn init_with_metadata(
297        context: &E,
298        cfg: Config<V::Cfg>,
299        metadata_partition: String,
300        read_options: ReadOptions,
301    ) -> Result<Replay<E, I, V>, Error> {
302        let replay_buffer = cfg.replay_buffer;
303
304        // Open the commit markers before the data they constrain.
305        let metadata = Metadata::init(
306            context.child("metadata"),
307            MetadataConfig {
308                partition: metadata_partition,
309                codec_config: (),
310            },
311        )
312        .await?;
313        let floors = metadata
314            .keys()
315            .map(|key| {
316                (
317                    u64::from(key),
318                    *metadata.get(key).expect("metadata key must have a value"),
319                )
320            })
321            .collect::<BTreeMap<_, _>>();
322
323        // Every advertised prefix is proven before ordinary suffix repair may mutate either
324        // journal. An empty sidecar retains the legacy inferred-recovery behavior.
325        let recovery = if floors.is_empty() {
326            Recovery::Infer
327        } else {
328            Recovery::Floors(&floors)
329        };
330        let mut journal = Self::init_inner(context.child("oversized"), cfg, recovery).await?;
331        journal.tracking = Some(Tracking {
332            metadata,
333            marker_sync_pending: None,
334            barriers: BTreeMap::new(),
335        });
336        let mut replay = journal.replay(0, 0, replay_buffer, read_options).await?;
337        replay.validation = Some(Validation::new());
338        Ok(replay)
339    }
340
341    /// Open the index and value journals, reconciling them per the selected [Recovery] mode.
342    async fn init_inner(
343        context: E,
344        cfg: Config<V::Cfg>,
345        recovery: Recovery<'_>,
346    ) -> Result<Self, Error> {
347        let index_cfg = FixedConfig {
348            partition: cfg.index_partition,
349            page_cache: cfg.index_page_cache,
350            write_buffer: cfg.index_write_buffer,
351        };
352        let index_context = context.child("index");
353        let value_cfg = GlobConfig {
354            partition: cfg.value_partition,
355            compression: cfg.compression,
356            codec_config: cfg.codec_config,
357            write_buffer: cfg.value_write_buffer,
358        };
359        let value_context = context.child("values");
360
361        let (index, values) = match recovery {
362            Recovery::Infer => {
363                let index = FixedJournal::init(index_context, index_cfg).await?;
364                (index, Glob::init(value_context, value_cfg).await?)
365            }
366            Recovery::Floors(minimum_items) => {
367                let preflight =
368                    FixedJournal::preflight_floors(index_context, index_cfg, minimum_items).await?;
369                let values = Glob::init(value_context, value_cfg).await?;
370                Self::validate_value_floors(&values, &preflight)?;
371                (preflight.finish().await?, values)
372            }
373            Recovery::Restore {
374                section,
375                index_size,
376            } => {
377                let preflight =
378                    FixedJournal::preflight_restore(index_context, index_cfg, section, index_size)
379                        .await?;
380                let values = Glob::init(value_context, value_cfg).await?;
381                let value_size = Self::validate_restore_values(&values, &preflight, section)?;
382                let index = preflight.finish().await?;
383
384                // The index truncation is already durable. Release its unreferenced values only
385                // after that proof, preserving the index-first crash-recovery order.
386                let values = values.rewind(section, value_size).await?;
387                (index, values.sync(section).await?)
388            }
389        };
390        Ok(Self {
391            index,
392            values,
393            tracking: None,
394        })
395    }
396
397    /// Drain the fixed journal's ordered recovery pass, then reconcile the value tail of each
398    /// section. The replay buffer controls every forward index read.
399    async fn recover_inferred(self, buffer: NonZeroUsize) -> Result<Self, Error> {
400        let mut replay = self.replay(0, 0, buffer, ReadOptions::default()).await?;
401        while let Some(result) = replay.next().await {
402            result?;
403        }
404        replay.finish()?.repair().await
405    }
406
407    /// Return the value boundary owned by an optional terminal index entry.
408    fn boundary_value_end(section: u64, entry: &Option<I>) -> Result<u64, Error> {
409        let Some(entry) = entry else {
410            return Ok(0);
411        };
412        let (offset, size) = entry.value_location();
413        offset.checked_add(u64::from(size)).ok_or_else(|| {
414            Error::Corruption(format!(
415                "section {section} has an overflowing terminal value range"
416            ))
417        })
418    }
419
420    /// Perform crash recovery by validating index entries against glob contents.
421    ///
422    /// Only checks entries from the end of each section until one is valid. Since entries
423    /// are appended sequentially and value offsets are monotonically increasing within a
424    /// section, all earlier entries must be range-valid (their value checksums are
425    /// verified lazily at read).
426    async fn repair(mut self) -> Result<Self, Error> {
427        let chunk_size = FixedJournal::<E, I>::CHUNK_SIZE as u64;
428        let sections: Vec<u64> = self.index.sections().collect();
429
430        let mut rewound_index = Vec::new();
431        let mut rewound_values = Vec::new();
432        for section in sections {
433            let index_size = self.index.size(section)?;
434            let glob_size = match self.values.size(section) {
435                Ok(size) => size,
436                Err(Error::AlreadyPrunedToSection(oldest)) => {
437                    // This shouldn't happen in normal operation: prune() prunes the index
438                    // first, then the glob. A crash between these would leave the glob
439                    // NOT pruned (opposite of this case). We handle this defensively in
440                    // case of external manipulation or future changes.
441                    warn!(
442                        section,
443                        oldest, "index has section that glob already pruned"
444                    );
445                    0
446                }
447                Err(e) => return Err(e),
448            };
449
450            // Truncate any trailing partial entry
451            let entry_count = index_size / chunk_size;
452            let aligned_size = entry_count * chunk_size;
453            if aligned_size < index_size {
454                warn!(
455                    section,
456                    index_size, aligned_size, "trailing bytes detected: truncating"
457                );
458                self.index = self.index.rewind_section(section, aligned_size).await?;
459                rewound_index.push(section);
460            }
461
462            // Values are reachable only through index entries.
463            if entry_count == 0 {
464                if glob_size > 0 {
465                    debug!(section, glob_size, "truncating orphaned value bytes");
466                    self.values = self.values.rewind_section(section, 0).await?;
467                    rewound_values.push(section);
468                }
469                continue;
470            }
471
472            // Find last valid entry and target glob size
473            let (valid_count, glob_target) = self
474                .find_last_valid_entry(section, entry_count, glob_size)
475                .await?;
476
477            // Rewind index if any entries are invalid
478            if valid_count < entry_count {
479                let valid_size = valid_count * chunk_size;
480                debug!(section, entry_count, valid_count, "rewinding index");
481                self.index = self.index.rewind_section(section, valid_size).await?;
482                rewound_index.push(section);
483            }
484
485            // Truncate glob trailing garbage (can occur when value was written but
486            // index entry wasn't, or when index was truncated but glob wasn't)
487            if glob_size > glob_target {
488                debug!(
489                    section,
490                    glob_size, glob_target, "truncating glob trailing garbage"
491                );
492                self.values = self.values.rewind_section(section, glob_target).await?;
493                rewound_values.push(section);
494            }
495        }
496
497        // Make the truncations durable before appends can reuse the freed value ranges. A
498        // dropped index entry that stayed durable would be adopted by a later recovery
499        // referencing whatever bytes a subsequent append placed at its offsets, and stale
500        // glob bytes that stayed durable would satisfy a later entry's range with another
501        // record's frame.
502        self.values = self.values.sync(&rewound_values).await?;
503        self.index = self.index.sync(&rewound_index).await?;
504
505        // Clean up orphan value sections that don't exist in index
506        self.cleanup_orphan_value_sections().await
507    }
508
509    /// Verify every floor's terminal value extent before repair can mutate either journal.
510    fn validate_value_floors(
511        values: &Glob<E, V>,
512        preflight: &RecoveryPreflight<E, I>,
513    ) -> Result<(), Error> {
514        for (&section, entry) in preflight.boundaries() {
515            let required = Self::boundary_value_end(section, entry)?;
516            if required == 0 {
517                continue;
518            }
519            let retained = values.size(section)?;
520            if retained < required {
521                return Err(Error::Corruption(format!(
522                    "section {section} retains {retained} value bytes, below the validation \
523                     floor of {required}"
524                )));
525            }
526        }
527        Ok(())
528    }
529
530    /// Verify checkpoint-covered value extents against preflighted index boundaries, returning
531    /// the checkpoint section's terminal value end.
532    fn validate_restore_values(
533        values: &Glob<E, V>,
534        preflight: &RecoveryPreflight<E, I>,
535        section: u64,
536    ) -> Result<u64, Error> {
537        // Every earlier section is immutable under the checkpoint, so its terminal index entry
538        // must end exactly at the retained value length.
539        for (&candidate, entry) in preflight.boundaries().range(..section) {
540            let required = Self::boundary_value_end(candidate, entry)?;
541            let retained = values.size(candidate)?;
542            if retained != required {
543                return Err(Error::Corruption(format!(
544                    "section {candidate} index ends at value byte {required}, but its glob size is {retained}"
545                )));
546            }
547        }
548
549        // A value-only section below the checkpoint proves that covered index data was lost.
550        if let Some(orphan) = values.sections().find(|candidate| {
551            *candidate < section && !preflight.boundaries().contains_key(candidate)
552        }) {
553            return Err(Error::Corruption(format!(
554                "section {orphan} has values but no index"
555            )));
556        }
557
558        // The current section may retain a suffix, but it must cover its committed terminal value.
559        let required = Self::boundary_value_end(section, &preflight.boundaries()[&section])?;
560        if required > 0 {
561            let retained = values.size(section)?;
562            if retained < required {
563                return Err(Error::Corruption(format!(
564                    "section {section} retains {retained} of {required} committed value bytes"
565                )));
566            }
567        }
568
569        Ok(required)
570    }
571
572    /// Remove any value sections that don't have corresponding index sections.
573    ///
574    /// This can happen if a crash occurs after writing to values but before
575    /// writing to index for a new section. Since sections don't have to be
576    /// contiguous, we compare the actual sets of sections rather than just
577    /// comparing the newest section numbers.
578    async fn cleanup_orphan_value_sections(mut self) -> Result<Self, Error> {
579        // Collect index sections into a set for O(1) lookup
580        let index_sections: HashSet<u64> = self.index.sections().collect();
581
582        // Find value sections that don't exist in index
583        let orphan_sections: Vec<u64> = self
584            .values
585            .sections()
586            .filter(|s| !index_sections.contains(s))
587            .collect();
588
589        // Remove each orphan section
590        for section in orphan_sections {
591            warn!(section, "removing orphan value section");
592            (self.values, _) = self.values.remove_section(section).await?;
593        }
594
595        Ok(self)
596    }
597
598    /// Truncate value suffixes that became unreachable while fixed replay repaired index pages.
599    async fn align_values_to_index(mut self) -> Result<Self, Error> {
600        let sections = self.index.sections().collect::<Vec<_>>();
601        let mut rewound = Vec::new();
602        for section in sections {
603            let target = Self::boundary_value_end(section, &self.index.last(section).await?)?;
604            let retained = self.values.size(section)?;
605            if retained < target {
606                return Err(Error::Corruption(format!(
607                    "section {section} retains {retained} of {target} indexed value bytes"
608                )));
609            }
610            if retained > target {
611                self.values = self.values.rewind_section(section, target).await?;
612                rewound.push(section);
613            }
614        }
615        self.values = self.values.sync(&rewound).await?;
616        self.cleanup_orphan_value_sections().await
617    }
618
619    /// Find the number of valid entries and the corresponding glob target size.
620    ///
621    /// Scans backwards from the last entry until a valid one is found: an entry is valid
622    /// only if its byte range fits within the glob and its value's checksum verifies.
623    /// Returns `(valid_count, glob_target)` where `glob_target` is the end offset
624    /// of the last valid entry's value.
625    async fn find_last_valid_entry(
626        &self,
627        section: u64,
628        entry_count: u64,
629        glob_size: u64,
630    ) -> Result<(u64, u64), Error> {
631        for pos in (0..entry_count).rev() {
632            match self.index.get(section, pos).await {
633                Ok(entry) => {
634                    let (offset, size) = entry.value_location();
635                    let entry_end = offset.saturating_add(u64::from(size));
636                    if entry_end <= glob_size && self.values.verify(section, offset, size).await? {
637                        return Ok((pos + 1, entry_end));
638                    }
639                    if pos == entry_count - 1 {
640                        warn!(
641                            section,
642                            pos, glob_size, entry_end, "invalid entry: glob truncated or corrupt"
643                        );
644                    }
645                }
646                Err(Error::ItemOutOfRange(_) | Error::Runtime(RError::InvalidChecksum)) => {
647                    if pos == entry_count - 1 {
648                        warn!(section, pos, "corrupted last entry, scanning backwards");
649                    }
650                }
651                Err(err) => return Err(err),
652            }
653        }
654        Ok((0, 0))
655    }
656
657    /// Reconcile durable markers with the retained state after tracked startup recovery.
658    async fn reconcile_markers(mut self) -> Result<Self, Error> {
659        let mut tracking = self
660            .tracking
661            .take()
662            .expect("tracked replay preserves its recovery state");
663
664        let mut dirty = false;
665        for section in self.index.sections() {
666            let items = self.index.section_len(section)?;
667            dirty |= tracking.stage_marker(section, items);
668        }
669        if dirty {
670            let marker;
671            (tracking.metadata, marker) = tracking.metadata.start_sync().await?;
672            tracking.marker_sync_pending = Some(marker.boxed().shared());
673        }
674        self.tracking = Some(tracking);
675        Ok(self)
676    }
677
678    /// Lower tracked floors before an operation can free any bytes they authorize.
679    async fn prepare_rewind(
680        &mut self,
681        section: u64,
682        index_size: u64,
683        remove_later: bool,
684    ) -> Result<(), Error> {
685        let Some(mut tracking) = self.tracking.take() else {
686            return Ok(());
687        };
688        let items = index_size / FixedJournal::<E, I>::CHUNK_SIZE as u64;
689        let mut dirty = false;
690
691        if remove_later {
692            tracking.metadata.retain(|key, _| {
693                let keep = u64::from(key) <= section;
694                dirty |= !keep;
695                keep
696            });
697        }
698        let key = SectionKey::new(section);
699        if tracking
700            .metadata
701            .get(&key)
702            .is_some_and(|floor| *floor > items)
703        {
704            tracking.metadata.put(key, items);
705            dirty = true;
706        }
707        if dirty {
708            tracking.metadata = tracking.metadata.sync().await?;
709            tracking.marker_sync_pending = None;
710        }
711
712        if remove_later {
713            tracking
714                .barriers
715                .retain(|candidate, _| *candidate <= section);
716        }
717        if let Some(barrier) = tracking.barriers.get_mut(&section) {
718            barrier.truncate(items);
719        }
720        self.tracking = Some(tracking);
721        Ok(())
722    }
723
724    /// Append entry + value.
725    ///
726    /// Writes value to glob first, then writes index entry with the value location.
727    ///
728    /// Returns `(self, position, offset, size)` where:
729    /// - `position`: Position in the index journal
730    /// - `offset`: Byte offset in glob
731    /// - `size`: Size of value in glob (including checksum)
732    pub async fn append(
733        mut self,
734        section: u64,
735        entry: I,
736        value: &V,
737    ) -> Result<(Self, u64, u64, u32), Error> {
738        // Write value first (glob). This will typically write to an in-memory
739        // buffer and return quickly (only blocks when the buffer is full).
740        let (offset, size);
741        (self.values, offset, size) = self.values.append(section, value).await?;
742
743        // Update entry with actual location and write to index
744        let entry_with_location = entry.with_location(offset, size);
745        let position;
746        (self.index, position) = self.index.append(section, &entry_with_location).await?;
747
748        // Track this section so a later sync can prove and publish its new length. A fresh
749        // barrier claims only the staged floor, never the unproven pre-append prefix.
750        if let Some(tracking) = &mut self.tracking {
751            tracking.barrier(section);
752        }
753
754        Ok((self, position, offset, size))
755    }
756
757    /// Get entry at position (index entry only, not value).
758    pub async fn get(&self, section: u64, position: u64) -> Result<I, Error> {
759        self.index.get(section, position).await
760    }
761
762    /// Get the last entry for a section, if any.
763    ///
764    /// Returns `Ok(None)` if the section is empty.
765    ///
766    /// # Errors
767    ///
768    /// - [Error::AlreadyPrunedToSection] if the section has been pruned.
769    /// - [Error::SectionOutOfRange] if the section doesn't exist.
770    pub async fn last(&self, section: u64) -> Result<Option<I>, Error> {
771        self.index.last(section).await
772    }
773
774    /// Get value using offset/size from entry.
775    ///
776    /// The offset should be the byte offset from `append()` or from the entry's `value_location()`.
777    pub async fn get_value(&self, section: u64, offset: u64, size: u32) -> Result<V, Error> {
778        self.values.get(section, offset, size).await
779    }
780
781    /// Consumes the journal and returns an owned [Replay] reader over index entries
782    /// starting from `start_position` in `start_section`.
783    ///
784    /// Setup flushes the index journal's buffered pages so the reader observes every
785    /// accepted write. Every backing index-journal read performed by the returned
786    /// replay uses `read_options`, including reads after advancing to another
787    /// section.
788    pub async fn replay(
789        self,
790        start_section: u64,
791        start_position: u64,
792        buffer: NonZeroUsize,
793        read_options: ReadOptions,
794    ) -> Result<Replay<E, I, V>, Error> {
795        let Self {
796            index,
797            values,
798            tracking,
799        } = self;
800        let index = index
801            .replay(start_section, start_position, buffer, read_options)
802            .await?;
803        Ok(Replay {
804            index,
805            values,
806            tracking,
807            validation: None,
808        })
809    }
810
811    /// Start a joint data sync and publish only previously completed tracked boundaries.
812    ///
813    /// The returned handle covers only the requested data syncs. Marker generations trail
814    /// data durability by design, and a marker failure surfaces as [Error::Metadata] when a
815    /// later request observes it.
816    pub(crate) async fn start_sync_tracked(
817        mut self,
818        sections: &BTreeSet<u64>,
819        active: &BTreeSet<u64>,
820    ) -> Result<(Self, Handle<()>), Error> {
821        let mut tracking = self
822            .tracking
823            .take()
824            .expect("tracked sync preserves its recovery state");
825        let marker_pending = tracking.observe_marker_sync()?;
826        let lengths = sections
827            .iter()
828            .map(|&section| Ok((section, self.index.section_len(section)?)))
829            .collect::<Result<Vec<_>, Error>>()?;
830        let ((index, index_handle), (values, values_handle)) = try_join(
831            self.index.start_sync(sections),
832            self.values.start_sync(sections),
833        )
834        .await?;
835        self.index = index;
836        self.values = values;
837        let completion: SyncCompletion =
838            async move { try_join(index_handle, values_handle).await.map(|_| ()) }
839                .boxed()
840                .shared();
841
842        // Bind every selected target to the new joint completion. The recorded length becomes
843        // publishable only once this completion is observed to have succeeded.
844        for (section, length) in lengths {
845            tracking.barrier(section).record(length, completion.clone());
846        }
847
848        // Do not mutate Metadata while its prior generation is in flight. Completed barriers stay
849        // as debt and are published when their section is no longer active, or on an empty flush.
850        if !marker_pending {
851            let publish = tracking
852                .barriers
853                .iter_mut()
854                .filter(|(section, _)| !active.contains(section))
855                .map(|(&section, barrier)| (section, barrier.boundary()))
856                .collect::<Vec<_>>();
857            let mut metadata_dirty = false;
858            for (section, floor) in publish {
859                metadata_dirty |= tracking.stage_marker(section, floor);
860            }
861            if metadata_dirty {
862                let handle;
863                (tracking.metadata, handle) = tracking.metadata.start_sync().await?;
864                tracking.marker_sync_pending = Some(handle.boxed().shared());
865            }
866        }
867
868        self.tracking = Some(tracking);
869        Ok((self, Handle::from_future(completion)))
870    }
871
872    /// Block until the selected data syncs are durable, surfacing any marker failure the
873    /// completed generation already exposed.
874    pub(crate) async fn sync_tracked(
875        self,
876        sections: &BTreeSet<u64>,
877        active: &BTreeSet<u64>,
878    ) -> Result<Self, Error> {
879        let (mut journal, handle) = self.start_sync_tracked(sections, active).await?;
880        handle.await?;
881        journal
882            .tracking
883            .as_mut()
884            .expect("tracking mode is preserved")
885            .observe_marker_sync()?;
886        Ok(journal)
887    }
888
889    /// Sync both journals for the given `sections`.
890    pub async fn sync(mut self, sections: impl crate::Sections) -> Result<Self, Error> {
891        if self.tracking.is_some() {
892            let sections = sections.sections().collect::<BTreeSet<_>>();
893            return self.sync_tracked(&sections, &sections).await;
894        }
895
896        let sections = sections.sections().collect::<Vec<_>>();
897        (self.index, self.values) =
898            try_join(self.index.sync(&sections), self.values.sync(&sections)).await?;
899        Ok(self)
900    }
901
902    /// Start syncing both journals for the given `sections`.
903    ///
904    /// The returned handle completes once both journals' syncs complete, failing with the first
905    /// error encountered. An error reported by the returned [Handle] is fatal to the journal:
906    /// the caller must stop using the returned journal.
907    pub async fn start_sync(
908        mut self,
909        sections: impl crate::Sections,
910    ) -> Result<(Self, Handle<()>), Error> {
911        if self.tracking.is_some() {
912            let sections = sections.sections().collect::<BTreeSet<_>>();
913            return self.start_sync_tracked(&sections, &sections).await;
914        }
915
916        let sections = sections.sections().collect::<Vec<_>>();
917        let ((index, index_handle), (values, values_handle)) = try_join(
918            self.index.start_sync(&sections),
919            self.values.start_sync(&sections),
920        )
921        .await?;
922        self.index = index;
923        self.values = values;
924        Ok((
925            self,
926            Handle::from_future(
927                async move { try_join(index_handle, values_handle).await.map(|_| ()) },
928            ),
929        ))
930    }
931
932    /// Sync all sections.
933    pub async fn sync_all(mut self) -> Result<Self, Error> {
934        if self.tracking.is_some() {
935            let sections = self.index.sections().collect::<BTreeSet<_>>();
936            return self.sync(sections).await;
937        }
938        (self.index, self.values) = try_join(self.index.sync_all(), self.values.sync_all()).await?;
939        Ok(self)
940    }
941
942    /// Prune both journals. Returns true if any sections were pruned.
943    ///
944    /// Prunes index first, then glob. This order ensures crash safety:
945    /// - If crash after index prune but before glob: orphan data in glob (acceptable)
946    /// - If crash before index prune: no change, retry works
947    pub async fn prune(mut self, min: u64) -> Result<(Self, bool), Error> {
948        // Remove and durably sync tracked floors before their section names can be deleted and
949        // later reused.
950        if let Some(mut tracking) = self.tracking.take() {
951            let mut removed = false;
952            tracking.metadata.retain(|key, _| {
953                let keep = u64::from(key) >= min;
954                removed |= !keep;
955                keep
956            });
957            if removed {
958                tracking.metadata = tracking.metadata.sync().await?;
959                tracking.marker_sync_pending = None;
960            }
961            tracking.barriers = tracking.barriers.split_off(&min);
962            self.tracking = Some(tracking);
963        }
964
965        let index_pruned;
966        (self.index, index_pruned) = self.index.prune(min).await?;
967        let value_pruned;
968        (self.values, value_pruned) = self.values.prune(min).await?;
969        Ok((self, index_pruned || value_pruned))
970    }
971
972    /// Derive the value boundary owned by `section`'s last entry after an index rewind.
973    ///
974    /// A rewind to zero may leave no section behind, which owns no value bytes.
975    async fn rewound_value_end(&self, section: u64, index_size: u64) -> Result<u64, Error> {
976        match self.index.last(section).await {
977            Ok(Some(entry)) => {
978                let (offset, size) = entry.value_location();
979                offset
980                    .checked_add(u64::from(size))
981                    .ok_or(Error::OffsetOverflow)
982            }
983            Ok(None) => Ok(0),
984            Err(Error::SectionOutOfRange(_)) if index_size == 0 => Ok(0),
985            Err(e) => Err(e),
986        }
987    }
988
989    /// Rewind both journals to a specific section and index size.
990    ///
991    /// This rewinds the section to the given index size and removes all sections
992    /// after the given section. The value size is derived from the last entry.
993    ///
994    /// Both of `section`'s truncations are durable before this returns: a crash recovers
995    /// `section` to either its pre-rewind or its post-rewind state. Each journal removes
996    /// its later sections (newest first) before truncating `section`, and those removals
997    /// carry the storage layer's removal durability.
998    pub async fn rewind(mut self, section: u64, index_size: u64) -> Result<Self, Error> {
999        self.prepare_rewind(section, index_size, true).await?;
1000
1001        // Rewind index first (this also removes sections after `section`)
1002        self.index = self.index.rewind(section, index_size).await?;
1003
1004        // Derive value size from last entry (section may not exist if empty)
1005        let value_size = self.rewound_value_end(section, index_size).await?;
1006
1007        // Make the index truncation durable before the values are rewound: rewinding the
1008        // values frees their ranges for reuse by later appends, and a dropped index entry
1009        // that stayed durable would be adopted referencing whatever bytes a later append
1010        // placed at its offsets.
1011        self.index = self.index.sync(section).await?;
1012
1013        // Rewind values (this also removes sections after `section`)
1014        self.values = self.values.rewind(section, value_size).await?;
1015        self.values = self.values.sync(section).await?;
1016        Ok(self)
1017    }
1018
1019    /// Rewind only the given section to a specific index size.
1020    ///
1021    /// Unlike `rewind`, this does not affect other sections.
1022    /// The value size is derived from the last entry after rewinding the index.
1023    ///
1024    /// Both truncations are made durable before returning (see [Self::rewind]).
1025    pub async fn rewind_section(mut self, section: u64, index_size: u64) -> Result<Self, Error> {
1026        self.prepare_rewind(section, index_size, false).await?;
1027
1028        // Rewind index first
1029        self.index = self.index.rewind_section(section, index_size).await?;
1030
1031        // Derive value size from last entry (section may not exist if empty)
1032        let value_size = self.rewound_value_end(section, index_size).await?;
1033
1034        // Make the index truncation durable before the values are rewound (see Self::rewind).
1035        self.index = self.index.sync(section).await?;
1036
1037        // Rewind values
1038        self.values = self.values.rewind_section(section, value_size).await?;
1039        self.values = self.values.sync(section).await?;
1040        Ok(self)
1041    }
1042
1043    /// Get index size for checkpoint.
1044    ///
1045    /// The value size can be derived from the last entry's location when needed.
1046    pub fn size(&self, section: u64) -> Result<u64, Error> {
1047        self.index.size(section)
1048    }
1049
1050    /// Get the value size for a section, derived from the last entry's location.
1051    pub async fn value_size(&self, section: u64) -> Result<u64, Error> {
1052        match self.index.last(section).await {
1053            Ok(Some(entry)) => {
1054                let (offset, size) = entry.value_location();
1055                offset
1056                    .checked_add(u64::from(size))
1057                    .ok_or(Error::OffsetOverflow)
1058            }
1059            Ok(None) | Err(Error::SectionOutOfRange(_)) => Ok(0),
1060            Err(e) => Err(e),
1061        }
1062    }
1063
1064    /// Returns true when `section` is below the prune floor.
1065    ///
1066    /// The floor only tracks prunes from the current execution and resets at init, so a
1067    /// section pruned in a previous execution reports false.
1068    pub fn pruned(&self, section: u64) -> bool {
1069        self.index.pruned(section)
1070    }
1071
1072    /// Returns the oldest section number, if any exist.
1073    pub fn oldest_section(&self) -> Option<u64> {
1074        self.index.oldest_section()
1075    }
1076
1077    /// Returns the newest section number, if any exist.
1078    pub fn newest_section(&self) -> Option<u64> {
1079        self.index.newest_section()
1080    }
1081
1082    /// Destroy all underlying storage.
1083    pub async fn destroy(mut self) -> Result<(), Error> {
1084        // Remove tracked floors first so an interrupted destroy can only force conservative
1085        // recovery of any pair data left behind.
1086        if let Some(tracking) = self.tracking.take() {
1087            tracking.metadata.destroy().await?;
1088        }
1089        try_join(self.index.destroy(), self.values.destroy())
1090            .await
1091            .map(|_| ())
1092    }
1093}
1094
1095/// Owned replay reader over an [Oversized]'s index entries.
1096///
1097/// Yields `(section, position, entry)` in order. Dropping the reader before it is exhausted
1098/// destroys the journal: recovery is re-initialization. Call [Replay::finish] on an exhausted
1099/// reader to get the journal back.
1100pub struct Replay<E: Context, I: Record, V: Codec> {
1101    index: FixedReplay<E, I>,
1102    values: Glob<E, V>,
1103    tracking: Option<Tracking<E>>,
1104    validation: Option<Validation>,
1105}
1106
1107impl<E: Context, I: Record + Send + Sync, V: CodecShared> Replay<E, I, V> {
1108    /// Returns the next `(section, position, entry)`, or `None` once every section is
1109    /// exhausted.
1110    ///
1111    /// An index error ends the section that produced it, and iteration continues with
1112    /// the next section. A value-verification error is returned without ending its
1113    /// section and dooms the tracked replay: finishing it fails with
1114    /// [Error::ReplayFailed]. Errors while mutating storage to repair a section, and
1115    /// [Error::ReplayInterrupted], end the replay.
1116    pub async fn next(&mut self) -> Option<Result<(u64, u64, I), Error>> {
1117        loop {
1118            let result = self.index.next().await?;
1119            let (section, position, entry) = match result {
1120                Ok(entry) => entry,
1121                Err(err) => return Some(Err(err)),
1122            };
1123            let (tracking, validation) = (&self.tracking, &mut self.validation);
1124            let Some(validation) = validation else {
1125                return Some(Ok((section, position, entry)));
1126            };
1127
1128            // Each section validates forward from its durable floor. Once one value fails, later
1129            // entries in that section are outside the retained prefix and are not yielded.
1130            if validation.current_section != Some(section) {
1131                validation.current_section = Some(section);
1132                validation.floor = tracking
1133                    .as_ref()
1134                    .expect("tracked replay preserves its recovery state")
1135                    .metadata
1136                    .get(&SectionKey::new(section))
1137                    .copied()
1138                    .unwrap_or(0);
1139                validation.truncated = false;
1140            }
1141            if validation.truncated {
1142                continue;
1143            }
1144            if position < validation.floor {
1145                return Some(Ok((section, position, entry)));
1146            }
1147
1148            let (offset, size) = entry.value_location();
1149            match self.values.verify(section, offset, size).await {
1150                Ok(true) => return Some(Ok((section, position, entry))),
1151                Ok(false) => {
1152                    validation.rewinds.push((section, position));
1153                    validation.truncated = true;
1154                }
1155                Err(err) => {
1156                    validation.failed = true;
1157                    return Some(Err(err));
1158                }
1159            }
1160        }
1161    }
1162
1163    /// Returns the journal.
1164    ///
1165    /// Fails when the reader was not fully drained or yielded an error: the journal is
1166    /// destroyed and recovery is re-initialization.
1167    pub fn finish(self) -> Result<Oversized<E, I, V>, Error> {
1168        if self.validation.is_some() {
1169            return Err(Error::ReplayFailed);
1170        }
1171        Ok(Oversized {
1172            index: self.index.finish()?,
1173            values: self.values,
1174            tracking: self.tracking,
1175        })
1176    }
1177
1178    /// Finish marker-aware startup recovery and return the tracked journal.
1179    pub async fn finish_tracked(self) -> Result<Oversized<E, I, V>, Error> {
1180        let Some(validation) = self.validation else {
1181            return Err(Error::ReplayFailed);
1182        };
1183        if validation.failed {
1184            return Err(Error::ReplayFailed);
1185        }
1186        let mut journal = Oversized {
1187            index: self.index.finish()?,
1188            values: self.values,
1189            tracking: self.tracking,
1190        };
1191
1192        // Apply each section's first invalid position only after replay releases the index
1193        // journal, then publish the exact retained lengths as the next marker generation.
1194        let chunk_size = FixedJournal::<E, I>::CHUNK_SIZE as u64;
1195        for (section, items) in validation.rewinds {
1196            let index_size = items.checked_mul(chunk_size).ok_or(Error::OffsetOverflow)?;
1197            journal = journal.rewind_section(section, index_size).await?;
1198        }
1199        journal
1200            .align_values_to_index()
1201            .await?
1202            .reconcile_markers()
1203            .await
1204    }
1205}
1206
1207#[cfg(test)]
1208mod tests {
1209    use super::*;
1210    use commonware_codec::{FixedSize, Read, ReadExt, Write};
1211    use commonware_cryptography::Crc32;
1212    use commonware_macros::test_traced;
1213    use commonware_runtime::{
1214        Blob as _, Buf, BufMut, BufferPooler, Runner, Storage as _, Supervisor as _, WriteOptions,
1215        buffer::paged::{CacheRef, corrupt_page},
1216        deterministic,
1217        mocks::{DelayedSyncContext, PendingSyncs, SyncFaultContext, drive_pending_syncs},
1218    };
1219    use commonware_utils::{NZU16, NZUsize};
1220
1221    /// Convert offset + size to byte end position (for truncation tests).
1222    fn byte_end(offset: u64, size: u32) -> u64 {
1223        offset + u64::from(size)
1224    }
1225
1226    /// Test index entry that stores a u64 id and references a value.
1227    #[derive(Debug, Clone, PartialEq)]
1228    struct TestEntry {
1229        id: u64,
1230        value_offset: u64,
1231        value_size: u32,
1232    }
1233
1234    impl TestEntry {
1235        fn new(id: u64, value_offset: u64, value_size: u32) -> Self {
1236            Self {
1237                id,
1238                value_offset,
1239                value_size,
1240            }
1241        }
1242    }
1243
1244    impl Write for TestEntry {
1245        fn write(&self, buf: &mut impl BufMut) {
1246            self.id.write(buf);
1247            self.value_offset.write(buf);
1248            self.value_size.write(buf);
1249        }
1250    }
1251
1252    impl Read for TestEntry {
1253        type Cfg = ();
1254
1255        fn read_cfg(buf: &mut impl Buf, _: &Self::Cfg) -> Result<Self, commonware_codec::Error> {
1256            let id = u64::read(buf)?;
1257            let value_offset = u64::read(buf)?;
1258            let value_size = u32::read(buf)?;
1259            Ok(Self {
1260                id,
1261                value_offset,
1262                value_size,
1263            })
1264        }
1265    }
1266
1267    impl FixedSize for TestEntry {
1268        const SIZE: usize = u64::SIZE + u64::SIZE + u32::SIZE;
1269    }
1270
1271    impl Record for TestEntry {
1272        fn value_location(&self) -> (u64, u32) {
1273            (self.value_offset, self.value_size)
1274        }
1275
1276        fn with_location(mut self, offset: u64, size: u32) -> Self {
1277            self.value_offset = offset;
1278            self.value_size = size;
1279            self
1280        }
1281    }
1282
1283    fn test_cfg(pooler: &impl BufferPooler) -> Config<()> {
1284        Config {
1285            index_partition: "test-index".into(),
1286            value_partition: "test-values".into(),
1287            index_page_cache: CacheRef::from_pooler(pooler, NZU16!(64), NZUsize!(8)),
1288            index_write_buffer: NZUsize!(1024),
1289            value_write_buffer: NZUsize!(1024),
1290            replay_buffer: NZUsize!(4096),
1291            compression: None,
1292            codec_config: (),
1293        }
1294    }
1295
1296    /// Test configuration sized so each index page holds exactly one entry.
1297    fn entry_cfg(pooler: &impl BufferPooler) -> Config<()> {
1298        let mut cfg = test_cfg(pooler);
1299        cfg.index_page_cache =
1300            CacheRef::from_pooler(pooler, NZU16!(TestEntry::SIZE as u16), NZUsize!(8));
1301        cfg
1302    }
1303
1304    /// Simple test value type with unit config.
1305    type TestValue = [u8; 16];
1306
1307    #[test_traced]
1308    fn test_oversized_append_and_get() {
1309        let executor = deterministic::Runner::default();
1310        executor.start(|context| async move {
1311            let cfg = test_cfg(&context);
1312            let mut oversized: Oversized<_, TestEntry, TestValue> =
1313                Oversized::init(context, cfg).await.expect("Failed to init");
1314
1315            // Append entry with value
1316            let value: TestValue = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16];
1317            let entry = TestEntry::new(42, 0, 0);
1318            let (position, offset, size);
1319            (oversized, position, offset, size) = oversized
1320                .append(1, entry, &value)
1321                .await
1322                .expect("Failed to append");
1323
1324            assert_eq!(position, 0);
1325
1326            // Get entry
1327            let retrieved_entry = oversized.get(1, position).await.expect("Failed to get");
1328            assert_eq!(retrieved_entry.id, 42);
1329
1330            // Get value
1331            let retrieved_value = oversized
1332                .get_value(1, offset, size)
1333                .await
1334                .expect("Failed to get value");
1335            assert_eq!(retrieved_value, value);
1336
1337            oversized.destroy().await.expect("Failed to destroy");
1338        });
1339    }
1340
1341    #[test_traced]
1342    fn test_oversized_crash_recovery() {
1343        let executor = deterministic::Runner::default();
1344        executor.start(|context| async move {
1345            let cfg = test_cfg(&context);
1346
1347            // Create and populate oversized journal
1348            let mut oversized: Oversized<_, TestEntry, TestValue> =
1349                Oversized::init(context.child("first"), cfg.clone())
1350                    .await
1351                    .expect("Failed to init");
1352
1353            // Append multiple entries
1354            let mut locations = Vec::new();
1355            for i in 0..5u8 {
1356                let value: TestValue = [i; 16];
1357                let entry = TestEntry::new(i as u64, 0, 0);
1358                let (position, offset, size);
1359                (oversized, position, offset, size) = oversized
1360                    .append(1, entry, &value)
1361                    .await
1362                    .expect("Failed to append");
1363                locations.push((position, offset, size));
1364            }
1365            oversized = oversized.sync(1).await.expect("Failed to sync");
1366            drop(oversized);
1367
1368            // Simulate crash: truncate glob to lose last 2 values
1369            let (blob, _) = context
1370                .open(&cfg.value_partition, &1u64.to_be_bytes())
1371                .await
1372                .expect("Failed to open blob");
1373
1374            // Calculate size to keep first 3 entries
1375            let keep_size = byte_end(locations[2].1, locations[2].2);
1376            blob.resize(keep_size).await.expect("Failed to truncate");
1377            blob.sync().await.expect("Failed to sync");
1378            drop(blob);
1379
1380            // Reinitialize - should recover and rewind index
1381            let oversized: Oversized<_, TestEntry, TestValue> =
1382                Oversized::init(context.child("second"), cfg.clone())
1383                    .await
1384                    .expect("Failed to reinit");
1385
1386            // First 3 entries should still be valid
1387            for i in 0..3u8 {
1388                let (position, offset, size) = locations[i as usize];
1389                let entry = oversized.get(1, position).await.expect("Failed to get");
1390                assert_eq!(entry.id, i as u64);
1391
1392                let value = oversized
1393                    .get_value(1, offset, size)
1394                    .await
1395                    .expect("Failed to get value");
1396                assert_eq!(value, [i; 16]);
1397            }
1398
1399            // Entry at position 3 should fail (index was rewound)
1400            let result = oversized.get(1, 3).await;
1401            assert!(result.is_err());
1402
1403            oversized.destroy().await.expect("Failed to destroy");
1404        });
1405    }
1406
1407    #[test_traced]
1408    fn test_oversized_persistence() {
1409        let executor = deterministic::Runner::default();
1410        executor.start(|context| async move {
1411            let cfg = test_cfg(&context);
1412
1413            // Create and populate
1414            let mut oversized: Oversized<_, TestEntry, TestValue> =
1415                Oversized::init(context.child("first"), cfg.clone())
1416                    .await
1417                    .expect("Failed to init");
1418
1419            let value: TestValue = [42; 16];
1420            let entry = TestEntry::new(123, 0, 0);
1421            let (position, offset, size);
1422            (oversized, position, offset, size) = oversized
1423                .append(1, entry, &value)
1424                .await
1425                .expect("Failed to append");
1426            oversized = oversized.sync(1).await.expect("Failed to sync");
1427            drop(oversized);
1428
1429            // Reopen and verify
1430            let oversized: Oversized<_, TestEntry, TestValue> =
1431                Oversized::init(context.child("second"), cfg)
1432                    .await
1433                    .expect("Failed to reinit");
1434
1435            let retrieved_entry = oversized.get(1, position).await.expect("Failed to get");
1436            assert_eq!(retrieved_entry.id, 123);
1437
1438            let retrieved_value = oversized
1439                .get_value(1, offset, size)
1440                .await
1441                .expect("Failed to get value");
1442            assert_eq!(retrieved_value, value);
1443
1444            oversized.destroy().await.expect("Failed to destroy");
1445        });
1446    }
1447
1448    #[test_traced]
1449    fn test_oversized_sync() {
1450        let executor = deterministic::Runner::default();
1451        executor.start(|context| async move {
1452            let cfg = test_cfg(&context);
1453
1454            let mut oversized: Oversized<_, TestEntry, TestValue> =
1455                Oversized::init(context.child("first"), cfg.clone())
1456                    .await
1457                    .expect("Failed to init");
1458
1459            // One sub-page entry/value per section stays buffered until synced.
1460            let mut located = Vec::new();
1461            for section in 1u64..=3 {
1462                let value: TestValue = [section as u8; 16];
1463                let entry = TestEntry::new(section, 0, 0);
1464                let (position, offset, size);
1465                (oversized, position, offset, size) = oversized
1466                    .append(section, entry, &value)
1467                    .await
1468                    .expect("Failed to append");
1469                located.push((section, position, offset, size, value));
1470            }
1471
1472            // Sync sections 1 and 3 (both index and values); a nonexistent section (99) is
1473            // skipped, not an error.
1474            oversized = oversized
1475                .sync(&[1, 3, 99])
1476                .await
1477                .expect("Failed to sync sections");
1478            drop(oversized);
1479
1480            // Only the synced sections survive the unclean drop, with both index and value durable.
1481            let oversized: Oversized<_, TestEntry, TestValue> =
1482                Oversized::init(context.child("second"), cfg)
1483                    .await
1484                    .expect("Failed to reinit");
1485            for &(section, position, offset, size, value) in &located {
1486                let result = oversized.get(section, position).await;
1487                if section == 2 {
1488                    assert!(result.is_err(), "unsynced section 2 must not be durable");
1489                    continue;
1490                }
1491                assert_eq!(result.expect("synced entry durable").id, section);
1492                let retrieved = oversized
1493                    .get_value(section, offset, size)
1494                    .await
1495                    .expect("synced value durable");
1496                assert_eq!(retrieved, value);
1497            }
1498
1499            oversized.destroy().await.expect("Failed to destroy");
1500        });
1501    }
1502
1503    /// Assert that every entry recovery adopted in section 1 reads back the value that was
1504    /// appended with it.
1505    async fn assert_adopted_entries_consistent(
1506        oversized: &Oversized<deterministic::Context, TestEntry, TestValue>,
1507    ) {
1508        let chunk = FixedJournal::<deterministic::Context, TestEntry>::CHUNK_SIZE as u64;
1509        for position in 0..oversized.size(1).expect("size") / chunk {
1510            let entry = oversized.get(1, position).await.expect("Failed to get");
1511            let (offset, size) = entry.value_location();
1512            let value = oversized
1513                .get_value(1, offset, size)
1514                .await
1515                .expect("adopted entry must reference durable bytes");
1516            assert_eq!(
1517                value, [entry.id as u8; 16],
1518                "entry {} must read back the value appended with it",
1519                entry.id
1520            );
1521        }
1522    }
1523
1524    #[test_traced]
1525    fn test_oversized_rewind_truncation_durable_before_offset_reuse() {
1526        let executor = deterministic::Runner::default();
1527        let (_, checkpoint) = executor.start_and_recover(|context| async move {
1528            // One fully durable entry/value pair.
1529            let mut oversized: Oversized<_, TestEntry, TestValue> =
1530                Oversized::init(context.child("first"), test_cfg(&context))
1531                    .await
1532                    .expect("Failed to init");
1533            (oversized, _, _, _) = oversized
1534                .append(1, TestEntry::new(1, 0, 0), &[1; 16])
1535                .await
1536                .expect("Failed to append");
1537            oversized = oversized.sync(1).await.expect("Failed to sync");
1538
1539            // Rewind entry 1 away and append entry 2 at entry 1's glob offset, then crash
1540            // between the value sync and the index sync: entry 2's bytes (same size, valid
1541            // checksum) become durable at the exact range entry 1 referenced. Only the
1542            // durable truncation in `rewind` prevents recovery from resurrecting entry 1
1543            // pointing at entry 2's value. The range and checksum checks cannot reject it.
1544            oversized = oversized.rewind(1, 0).await.expect("Failed to rewind");
1545            (oversized, _, _, _) = oversized
1546                .append(1, TestEntry::new(2, 0, 0), &[2; 16])
1547                .await
1548                .expect("Failed to append");
1549            oversized.values = oversized
1550                .values
1551                .sync(1)
1552                .await
1553                .expect("Failed to sync values");
1554        });
1555
1556        deterministic::Runner::from(checkpoint).start(|context| async move {
1557            let oversized: Oversized<_, TestEntry, TestValue> =
1558                Oversized::init(context.child("second"), test_cfg(&context))
1559                    .await
1560                    .expect("Failed to reinit");
1561            assert_eq!(
1562                oversized.size(1).expect("size"),
1563                0,
1564                "rewound entry must not be revived over reused value bytes"
1565            );
1566            oversized.destroy().await.expect("Failed to destroy");
1567        });
1568    }
1569
1570    #[test_traced]
1571    fn test_oversized_recovery_never_adopts_entries_for_lost_values() {
1572        // Crash 1: entry 1 becomes durable but its value does not (an index write surviving
1573        // a crash its value bytes did not).
1574        let executor = deterministic::Runner::default();
1575        let (_, checkpoint) = executor.start_and_recover(|context| async move {
1576            let mut oversized: Oversized<_, TestEntry, TestValue> =
1577                Oversized::init(context.child("first"), test_cfg(&context))
1578                    .await
1579                    .expect("Failed to init");
1580            (oversized, _, _, _) = oversized
1581                .append(1, TestEntry::new(1, 0, 0), &[1; 16])
1582                .await
1583                .expect("Failed to append");
1584            oversized.index = oversized.index.sync(1).await.expect("Failed to sync index");
1585        });
1586
1587        // Boot 2: recovery rewinds entry 1 (its range is out of bounds) and must make that
1588        // truncation durable. A new append then reuses entry 1's offset. Crash 2 lands
1589        // after the value sync and before the index sync.
1590        let (_, checkpoint) =
1591            deterministic::Runner::from(checkpoint).start_and_recover(|context| async move {
1592                let mut oversized: Oversized<_, TestEntry, TestValue> =
1593                    Oversized::init(context.child("second"), test_cfg(&context))
1594                        .await
1595                        .expect("Failed to reinit");
1596                assert_eq!(
1597                    oversized.size(1).expect("size"),
1598                    0,
1599                    "entry without durable value bytes must be rewound"
1600                );
1601                (oversized, _, _, _) = oversized
1602                    .append(1, TestEntry::new(2, 0, 0), &[2; 16])
1603                    .await
1604                    .expect("Failed to append");
1605                oversized.values = oversized
1606                    .values
1607                    .sync(1)
1608                    .await
1609                    .expect("Failed to sync values");
1610            });
1611
1612        // Boot 3: without a durable truncation in recovery, the index would still hold
1613        // entry 1, now range-valid over entry 2's bytes.
1614        deterministic::Runner::from(checkpoint).start(|context| async move {
1615            let oversized: Oversized<_, TestEntry, TestValue> =
1616                Oversized::init(context.child("third"), test_cfg(&context))
1617                    .await
1618                    .expect("Failed to reinit");
1619            assert_adopted_entries_consistent(&oversized).await;
1620            oversized.destroy().await.expect("Failed to destroy");
1621        });
1622    }
1623
1624    #[test_traced]
1625    fn test_oversized_rewind_fails_when_truncation_cannot_be_made_durable() {
1626        let executor = deterministic::Runner::default();
1627        let (_, checkpoint) = executor.start_and_recover(|context| async move {
1628            // One fully durable entry/value pair.
1629            let mut oversized: Oversized<_, TestEntry, TestValue> =
1630                Oversized::init(context.child("first"), test_cfg(&context))
1631                    .await
1632                    .expect("Failed to init");
1633            (oversized, _, _, _) = oversized
1634                .append(1, TestEntry::new(1, 0, 0), &[1; 16])
1635                .await
1636                .expect("Failed to append");
1637            oversized = oversized.sync(1).await.expect("Failed to sync");
1638            drop(oversized);
1639
1640            // Boot 2: the glob cannot be synced, so `rewind` must fail rather than return
1641            // with a values truncation that is not durable (later appends could otherwise
1642            // reuse entry 1's still-durable value range).
1643            let faulty_values = SyncFaultContext {
1644                inner: context.child("second"),
1645                fail_partition: "test-values".into(),
1646            };
1647            let oversized: Oversized<_, TestEntry, TestValue> =
1648                Oversized::init(faulty_values, test_cfg(&context))
1649                    .await
1650                    .expect("Failed to reinit");
1651            assert!(
1652                oversized.rewind(1, 0).await.is_err(),
1653                "rewind must fail when its truncation cannot be made durable"
1654            );
1655        });
1656
1657        // The index truncation was made durable before the failure, so the dropped entry
1658        // must not be adopted at recovery.
1659        deterministic::Runner::from(checkpoint).start(|context| async move {
1660            let oversized: Oversized<_, TestEntry, TestValue> =
1661                Oversized::init(context.child("third"), test_cfg(&context))
1662                    .await
1663                    .expect("Failed to reinit");
1664            assert_eq!(oversized.size(1).expect("size"), 0);
1665            oversized.destroy().await.expect("Failed to destroy");
1666        });
1667    }
1668
1669    #[test_traced]
1670    fn test_oversized_recovery_glob_truncation_durable_before_offset_reuse() {
1671        // Crash 1: the index truncation is durable but the glob still holds entry 2's
1672        // frame (the state a crash inside `rewind` leaves behind).
1673        let executor = deterministic::Runner::default();
1674        let (_, checkpoint) = executor.start_and_recover(|context| async move {
1675            let mut oversized: Oversized<_, TestEntry, TestValue> =
1676                Oversized::init(context.child("first"), test_cfg(&context))
1677                    .await
1678                    .expect("Failed to init");
1679            (oversized, _, _, _) = oversized
1680                .append(1, TestEntry::new(1, 0, 0), &[1; 16])
1681                .await
1682                .expect("Failed to append");
1683            (oversized, _, _, _) = oversized
1684                .append(1, TestEntry::new(2, 0, 0), &[2; 16])
1685                .await
1686                .expect("Failed to append");
1687            oversized = oversized.sync(1).await.expect("Failed to sync");
1688
1689            let chunk = FixedJournal::<deterministic::Context, TestEntry>::CHUNK_SIZE as u64;
1690            oversized.index = oversized
1691                .index
1692                .rewind(1, chunk)
1693                .await
1694                .expect("Failed to rewind index");
1695            oversized.index = oversized.index.sync(1).await.expect("Failed to sync index");
1696        });
1697
1698        // Boot 2: recovery truncates the glob to entry 1's end and must make that
1699        // truncation durable. Entry 3 (same size) then reuses entry 2's freed range.
1700        // Crash 2 lands after the index sync and before the values sync.
1701        let (_, checkpoint) =
1702            deterministic::Runner::from(checkpoint).start_and_recover(|context| async move {
1703                let mut oversized: Oversized<_, TestEntry, TestValue> =
1704                    Oversized::init(context.child("second"), test_cfg(&context))
1705                        .await
1706                        .expect("Failed to reinit");
1707                (oversized, _, _, _) = oversized
1708                    .append(1, TestEntry::new(3, 0, 0), &[3; 16])
1709                    .await
1710                    .expect("Failed to append");
1711                oversized.index = oversized.index.sync(1).await.expect("Failed to sync index");
1712            });
1713
1714        // Boot 3: without a durable glob truncation in recovery, entry 3 would be
1715        // adopted referencing entry 2's still-durable frame.
1716        deterministic::Runner::from(checkpoint).start(|context| async move {
1717            let oversized: Oversized<_, TestEntry, TestValue> =
1718                Oversized::init(context.child("third"), test_cfg(&context))
1719                    .await
1720                    .expect("Failed to reinit");
1721            assert_adopted_entries_consistent(&oversized).await;
1722            oversized.destroy().await.expect("Failed to destroy");
1723        });
1724    }
1725
1726    #[test_traced]
1727    fn test_oversized_rewind_crash_between_truncations_recovers_post_rewind() {
1728        let executor = deterministic::Runner::default();
1729        let (_, checkpoint) = executor.start_and_recover(|context| async move {
1730            // Two fully durable entry/value pairs.
1731            let mut oversized: Oversized<_, TestEntry, TestValue> =
1732                Oversized::init(context.child("first"), test_cfg(&context))
1733                    .await
1734                    .expect("Failed to init");
1735            (oversized, _, _, _) = oversized
1736                .append(1, TestEntry::new(1, 0, 0), &[1; 16])
1737                .await
1738                .expect("Failed to append");
1739            (oversized, _, _, _) = oversized
1740                .append(1, TestEntry::new(2, 0, 0), &[2; 16])
1741                .await
1742                .expect("Failed to append");
1743            oversized = oversized.sync(1).await.expect("Failed to sync");
1744
1745            // Replay `rewind(1, chunk)`'s steps up to the worst crash point: the index
1746            // truncation is durable but the freed value bytes are not yet rewound.
1747            let chunk = FixedJournal::<deterministic::Context, TestEntry>::CHUNK_SIZE as u64;
1748            oversized.index = oversized
1749                .index
1750                .rewind(1, chunk)
1751                .await
1752                .expect("Failed to rewind index");
1753            oversized.index = oversized.index.sync(1).await.expect("Failed to sync index");
1754        });
1755
1756        // Recovery must truncate the orphaned value bytes and land on the post-rewind
1757        // state.
1758        deterministic::Runner::from(checkpoint).start(|context| async move {
1759            let oversized: Oversized<_, TestEntry, TestValue> =
1760                Oversized::init(context.child("second"), test_cfg(&context))
1761                    .await
1762                    .expect("Failed to reinit");
1763            let chunk = FixedJournal::<deterministic::Context, TestEntry>::CHUNK_SIZE as u64;
1764            assert_eq!(oversized.size(1).expect("size"), chunk);
1765            let entry = oversized.get(1, 0).await.expect("Failed to get");
1766            let (offset, size) = entry.value_location();
1767            assert_eq!(
1768                oversized.values.size(1).expect("glob size"),
1769                byte_end(offset, size),
1770                "orphaned value bytes must be truncated"
1771            );
1772            assert_adopted_entries_consistent(&oversized).await;
1773            oversized.destroy().await.expect("Failed to destroy");
1774        });
1775    }
1776
1777    #[test_traced]
1778    fn test_oversized_start_sync_completion_means_recoverable() {
1779        let executor = deterministic::Runner::default();
1780        let (_, checkpoint) = executor.start_and_recover(|context| async move {
1781            let mut oversized: Oversized<_, TestEntry, TestValue> =
1782                Oversized::init(context.child("first"), test_cfg(&context))
1783                    .await
1784                    .expect("Failed to init");
1785            (oversized, _, _, _) = oversized
1786                .append(1, TestEntry::new(1, 0, 0), &[1; 16])
1787                .await
1788                .expect("Failed to append");
1789            let (_oversized, handle) = oversized.start_sync(1).await.expect("Failed to start sync");
1790            handle.await.expect("sync must complete");
1791            // Crash: everything covered by the completed handle must survive.
1792        });
1793
1794        deterministic::Runner::from(checkpoint).start(|context| async move {
1795            let oversized: Oversized<_, TestEntry, TestValue> =
1796                Oversized::init(context.child("second"), test_cfg(&context))
1797                    .await
1798                    .expect("Failed to reinit");
1799            let chunk = FixedJournal::<deterministic::Context, TestEntry>::CHUNK_SIZE as u64;
1800            assert_eq!(oversized.size(1).expect("size"), chunk);
1801            assert_adopted_entries_consistent(&oversized).await;
1802            oversized.destroy().await.expect("Failed to destroy");
1803        });
1804    }
1805
1806    #[test_traced]
1807    fn test_oversized_sync_values_failure_recovers_clean() {
1808        let executor = deterministic::Runner::default();
1809        let (_, checkpoint) = executor.start_and_recover(|context| async move {
1810            let faulty_values = SyncFaultContext {
1811                inner: context.child("first"),
1812                fail_partition: "test-values".into(),
1813            };
1814            let mut oversized: Oversized<_, TestEntry, TestValue> =
1815                Oversized::init(faulty_values, test_cfg(&context))
1816                    .await
1817                    .expect("Failed to init");
1818            (oversized, _, _, _) = oversized
1819                .append(1, TestEntry::new(1, 0, 0), &[1; 16])
1820                .await
1821                .expect("Failed to append");
1822
1823            // The value sync fails, so the caller is never acknowledged. The index sync
1824            // may still land, but recovery must not adopt an entry whose value bytes never
1825            // became durable.
1826            assert!(oversized.sync(1).await.is_err(), "value sync must fail");
1827        });
1828
1829        deterministic::Runner::from(checkpoint).start(|context| async move {
1830            let oversized: Oversized<_, TestEntry, TestValue> =
1831                Oversized::init(context.child("second"), test_cfg(&context))
1832                    .await
1833                    .expect("Failed to reinit");
1834            assert_eq!(
1835                oversized.size(1).expect("size"),
1836                0,
1837                "entry without durable value bytes must be rewound"
1838            );
1839            oversized.destroy().await.expect("Failed to destroy");
1840        });
1841    }
1842
1843    #[test_traced]
1844    fn test_oversized_start_sync_values_failure_recovers_clean() {
1845        let executor = deterministic::Runner::default();
1846        let (_, checkpoint) = executor.start_and_recover(|context| async move {
1847            let faulty_values = SyncFaultContext {
1848                inner: context.child("first"),
1849                fail_partition: "test-values".into(),
1850            };
1851            let mut oversized: Oversized<_, TestEntry, TestValue> =
1852                Oversized::init(faulty_values, test_cfg(&context))
1853                    .await
1854                    .expect("Failed to init");
1855            (oversized, _, _, _) = oversized
1856                .append(1, TestEntry::new(1, 0, 0), &[1; 16])
1857                .await
1858                .expect("Failed to append");
1859
1860            // The value sync fails, so the handle must surface the failure and the caller
1861            // is never acknowledged.
1862            let (_oversized, handle) = oversized.start_sync(1).await.expect("Failed to start sync");
1863            assert!(handle.await.is_err(), "value sync must fail");
1864        });
1865
1866        deterministic::Runner::from(checkpoint).start(|context| async move {
1867            let oversized: Oversized<_, TestEntry, TestValue> =
1868                Oversized::init(context.child("second"), test_cfg(&context))
1869                    .await
1870                    .expect("Failed to reinit");
1871            assert_eq!(
1872                oversized.size(1).expect("size"),
1873                0,
1874                "entry without durable value bytes must be rewound"
1875            );
1876            oversized.destroy().await.expect("Failed to destroy");
1877        });
1878    }
1879
1880    #[test_traced]
1881    fn test_oversized_start_sync_dropped_handle_driven_by_next_sync() {
1882        let executor = deterministic::Runner::default();
1883        let (_, checkpoint) = executor.start_and_recover(|context| async move {
1884            let mut oversized: Oversized<_, TestEntry, TestValue> =
1885                Oversized::init(context.child("first"), test_cfg(&context))
1886                    .await
1887                    .expect("Failed to init");
1888            (oversized, _, _, _) = oversized
1889                .append(1, TestEntry::new(1, 0, 0), &[1; 16])
1890                .await
1891                .expect("Failed to append");
1892
1893            // Drop the handle without observing it: the next sync must wait for the
1894            // started syncs and complete the work.
1895            let (oversized, handle) = oversized.start_sync(1).await.expect("Failed to start sync");
1896            drop(handle);
1897            oversized.sync(1).await.expect("Failed to sync");
1898        });
1899
1900        deterministic::Runner::from(checkpoint).start(|context| async move {
1901            let oversized: Oversized<_, TestEntry, TestValue> =
1902                Oversized::init(context.child("second"), test_cfg(&context))
1903                    .await
1904                    .expect("Failed to reinit");
1905            let chunk = FixedJournal::<deterministic::Context, TestEntry>::CHUNK_SIZE as u64;
1906            assert_eq!(oversized.size(1).expect("size"), chunk);
1907            assert_adopted_entries_consistent(&oversized).await;
1908            oversized.destroy().await.expect("Failed to destroy");
1909        });
1910    }
1911
1912    #[test_traced]
1913    fn test_oversized_recovery_rejects_entry_with_torn_value_bytes() {
1914        let executor = deterministic::Runner::default();
1915        let (_, checkpoint) = executor.start_and_recover(|context| async move {
1916            // One fully durable entry/value pair.
1917            let mut oversized: Oversized<_, TestEntry, TestValue> =
1918                Oversized::init(context.child("first"), test_cfg(&context))
1919                    .await
1920                    .expect("Failed to init");
1921            (oversized, _, _, _) = oversized
1922                .append(1, TestEntry::new(1, 0, 0), &[1; 16])
1923                .await
1924                .expect("Failed to append");
1925            oversized = oversized.sync(1).await.expect("Failed to sync");
1926
1927            // Append entry 2, make its index entry durable, then persist the glob's LENGTH
1928            // over entry 2's range without its bytes (writeback-mode metadata journaling):
1929            // overwrite the frame with same-length garbage and sync the values journal.
1930            let (offset, size);
1931            (oversized, _, offset, size) = oversized
1932                .append(1, TestEntry::new(2, 0, 0), &[2; 16])
1933                .await
1934                .expect("Failed to append");
1935            oversized.index = oversized.index.sync(1).await.expect("Failed to sync index");
1936            oversized
1937                .values
1938                .inject(1, offset, vec![0xFF; size as usize])
1939                .await
1940                .expect("Failed to overwrite value bytes");
1941            oversized.values = oversized
1942                .values
1943                .sync(1)
1944                .await
1945                .expect("Failed to sync values");
1946        });
1947
1948        // Entry 2's range fits within the glob, so only the checksum check can reject it.
1949        deterministic::Runner::from(checkpoint).start(|context| async move {
1950            let oversized: Oversized<_, TestEntry, TestValue> =
1951                Oversized::init(context.child("second"), test_cfg(&context))
1952                    .await
1953                    .expect("Failed to reinit");
1954            let chunk = FixedJournal::<deterministic::Context, TestEntry>::CHUNK_SIZE as u64;
1955            assert_eq!(
1956                oversized.size(1).expect("size"),
1957                chunk,
1958                "entry with torn value bytes must be rewound"
1959            );
1960            assert_adopted_entries_consistent(&oversized).await;
1961            oversized.destroy().await.expect("Failed to destroy");
1962        });
1963    }
1964
1965    #[test_traced]
1966    fn test_recovery_scans_past_torn_interior_index_page() {
1967        let executor = deterministic::Runner::default();
1968        executor.start(|context| async move {
1969            // Use page size = entry size so each entry is on its own page.
1970            let cfg = entry_cfg(&context);
1971
1972            // Create five durable entry/value pairs.
1973            let mut oversized: Oversized<_, TestEntry, TestValue> =
1974                Oversized::init(context.child("first"), cfg.clone())
1975                    .await
1976                    .expect("Failed to init");
1977            for i in 1..=5u8 {
1978                (oversized, _, _, _) = oversized
1979                    .append(1, TestEntry::new(i as u64, 0, 0), &[i; 16])
1980                    .await
1981                    .expect("Failed to append");
1982            }
1983            oversized = oversized.sync(1).await.expect("Failed to sync");
1984            drop(oversized);
1985
1986            // Corrupt the CRC record of the THIRD entry's index page: the backward open
1987            // scan stops at the (valid) last page, so the torn page survives in bounds.
1988            let physical_page = TestEntry::SIZE as u64 + 12;
1989            let (index_blob, size) = context
1990                .open(&cfg.index_partition, &1u64.to_be_bytes())
1991                .await
1992                .expect("Failed to open index blob");
1993            assert_eq!(size, 5 * physical_page);
1994            index_blob
1995                .write_at(
1996                    2 * physical_page + TestEntry::SIZE as u64,
1997                    vec![0xFF; 12],
1998                    WriteOptions::SYNC,
1999                )
2000                .await
2001                .expect("Failed to corrupt index page");
2002            drop(index_blob);
2003
2004            // Corrupt the fourth and fifth values so the backward scan must walk past
2005            // their entries and read the torn page.
2006            let (values_blob, _) = context
2007                .open(&cfg.value_partition, &1u64.to_be_bytes())
2008                .await
2009                .expect("Failed to open values blob");
2010            values_blob
2011                .write_at(60, vec![0xFF; 20], WriteOptions::SYNC)
2012                .await
2013                .expect("Failed to corrupt value");
2014            values_blob
2015                .write_at(80, vec![0xFF; 20], WriteOptions::SYNC)
2016                .await
2017                .expect("Failed to corrupt value");
2018            drop(values_blob);
2019
2020            // Recovery scans past the invalid tail values and the torn page (surfaced
2021            // as a checksum failure, not a generic read error) to the last valid pair.
2022            let oversized: Oversized<_, TestEntry, TestValue> =
2023                Oversized::init(context.child("second"), cfg)
2024                    .await
2025                    .expect("Failed to reinit");
2026            let chunk = FixedJournal::<deterministic::Context, TestEntry>::CHUNK_SIZE as u64;
2027            assert_eq!(oversized.size(1).expect("size"), 2 * chunk);
2028            assert_adopted_entries_consistent(&oversized).await;
2029            oversized.destroy().await.expect("Failed to destroy");
2030        });
2031    }
2032
2033    #[test_traced]
2034    fn test_oversized_restore_discards_beyond_checkpoint() {
2035        let executor = deterministic::Runner::default();
2036        let (_, checkpoint) = executor.start_and_recover(|context| async move {
2037            // One committed entry, then torn state beyond the checkpoint: entries in
2038            // section 1 and 2 whose index becomes durable ahead of their values.
2039            let mut oversized: Oversized<_, TestEntry, TestValue> =
2040                Oversized::init(context.child("first"), test_cfg(&context))
2041                    .await
2042                    .expect("Failed to init");
2043            (oversized, _, _, _) = oversized
2044                .append(1, TestEntry::new(1, 0, 0), &[1; 16])
2045                .await
2046                .expect("Failed to append");
2047            oversized = oversized.sync(1).await.expect("Failed to sync");
2048            (oversized, _, _, _) = oversized
2049                .append(1, TestEntry::new(2, 0, 0), &[2; 16])
2050                .await
2051                .expect("Failed to append");
2052            (oversized, _, _, _) = oversized
2053                .append(2, TestEntry::new(3, 0, 0), &[3; 16])
2054                .await
2055                .expect("Failed to append");
2056            oversized.index = oversized
2057                .index
2058                .sync(&[1, 2])
2059                .await
2060                .expect("Failed to sync index");
2061        });
2062
2063        // Restore truncates to the checkpoint without validating the discarded state.
2064        deterministic::Runner::from(checkpoint).start(|context| async move {
2065            let chunk = FixedJournal::<deterministic::Context, TestEntry>::CHUNK_SIZE as u64;
2066            let oversized: Oversized<_, TestEntry, TestValue> = Oversized::init_with_checkpoint(
2067                context.child("second"),
2068                test_cfg(&context),
2069                (1, chunk),
2070            )
2071            .await
2072            .expect("Failed to reinit");
2073            assert_eq!(oversized.size(1).expect("size"), chunk);
2074            assert_eq!(oversized.newest_section(), Some(1));
2075            assert_adopted_entries_consistent(&oversized).await;
2076            oversized.destroy().await.expect("Failed to destroy");
2077        });
2078    }
2079
2080    #[test_traced]
2081    fn test_oversized_restore_does_not_repair_discarded_sections() {
2082        let executor = deterministic::Runner::default();
2083        executor.start(|context| async move {
2084            let cfg = entry_cfg(&context);
2085            let mut oversized: Oversized<_, TestEntry, TestValue> =
2086                Oversized::init(context.child("seed"), cfg.clone())
2087                    .await
2088                    .expect("failed to init");
2089            (oversized, _, _, _) = oversized
2090                .append(1, TestEntry::new(0, 0, 0), &[0; 16])
2091                .await
2092                .expect("failed to append checkpoint entry");
2093            for id in 1..=3 {
2094                (oversized, _, _, _) = oversized
2095                    .append(2, TestEntry::new(id, 0, 0), &[id as u8; 16])
2096                    .await
2097                    .expect("failed to append discardable entry");
2098            }
2099            oversized = oversized.sync_all().await.expect("failed to sync");
2100            drop(oversized);
2101
2102            // The valid final page hides this interior hole from Writer::new. Restore owns no
2103            // bytes in section 2 and must remove it without first repairing and syncing it.
2104            corrupt_page(
2105                &context,
2106                &cfg.index_partition,
2107                &2u64.to_be_bytes(),
2108                1,
2109                TestEntry::SIZE as u64,
2110            )
2111            .await;
2112
2113            let pending = PendingSyncs::default();
2114            pending.arm();
2115            let delayed = DelayedSyncContext {
2116                inner: context,
2117                pending: pending.clone(),
2118            };
2119            let chunk = TestEntry::SIZE as u64;
2120            let oversized: Oversized<_, TestEntry, TestValue> = drive_pending_syncs(
2121                &pending,
2122                Oversized::init_with_checkpoint(delayed.child("restore"), cfg, (1, chunk)),
2123            )
2124            .await
2125            .expect("checkpoint restore failed");
2126
2127            // Restoring an already exact checkpoint syncs its index and values once each. Any
2128            // additional durability work came from repairing data that restore discards.
2129            assert_eq!(pending.calls(), 2);
2130            oversized.destroy().await.expect("failed to destroy");
2131        });
2132    }
2133
2134    #[test_traced]
2135    fn test_oversized_restore_incomplete_section_errors() {
2136        let executor = deterministic::Runner::default();
2137        executor.start(|context| async move {
2138            // Two committed sections
2139            let mut oversized: Oversized<_, TestEntry, TestValue> =
2140                Oversized::init(context.child("first"), test_cfg(&context))
2141                    .await
2142                    .expect("Failed to init");
2143            (oversized, _, _, _) = oversized
2144                .append(1, TestEntry::new(1, 0, 0), &[1; 16])
2145                .await
2146                .expect("Failed to append");
2147            (oversized, _, _, _) = oversized
2148                .append(2, TestEntry::new(2, 0, 0), &[2; 16])
2149                .await
2150                .expect("Failed to append");
2151            oversized = oversized.sync_all().await.expect("Failed to sync");
2152            drop(oversized);
2153
2154            // Truncate section 1's values, simulating lost durable state below the
2155            // checkpoint
2156            let (blob, len) = context
2157                .open("test-values", &1u64.to_be_bytes())
2158                .await
2159                .expect("Failed to open values blob");
2160            blob.resize(len - 1).await.expect("Failed to resize");
2161            blob.sync().await.expect("Failed to sync");
2162            drop(blob);
2163
2164            // The checkpoint covers the damaged section, so init must fail. Nothing is
2165            // repaired, so the failure persists across restarts.
2166            let chunk = FixedJournal::<deterministic::Context, TestEntry>::CHUNK_SIZE as u64;
2167            for instance in ["second", "third"] {
2168                let result: Result<Oversized<_, TestEntry, TestValue>, Error> =
2169                    Oversized::init_with_checkpoint(
2170                        context.child(instance),
2171                        test_cfg(&context),
2172                        (2, chunk),
2173                    )
2174                    .await;
2175                assert!(matches!(result, Err(Error::Corruption(_))));
2176            }
2177        });
2178    }
2179
2180    #[test_traced]
2181    fn test_oversized_restore_adopts_interior_index_corruption() {
2182        let executor = deterministic::Runner::default();
2183        executor.start(|context| async move {
2184            for (child, seed_child, restore_child, checkpoint) in [
2185                (
2186                    "checkpoint_section",
2187                    "seed_checkpoint_section",
2188                    "restore_checkpoint_section",
2189                    (1, 3 * TestEntry::SIZE as u64),
2190                ),
2191                (
2192                    "earlier_section",
2193                    "seed_earlier_section",
2194                    "restore_earlier_section",
2195                    (2, TestEntry::SIZE as u64),
2196                ),
2197            ] {
2198                let mut cfg = entry_cfg(&context);
2199                cfg.index_partition = format!("test-index-{child}");
2200                cfg.value_partition = format!("test-values-{child}");
2201
2202                // Persist three entries in section 1 and a later checkpoint candidate. One entry
2203                // per integrity page makes the damaged page strictly interior.
2204                let mut oversized: Oversized<_, TestEntry, TestValue> =
2205                    Oversized::init(context.child(seed_child), cfg.clone())
2206                        .await
2207                        .expect("failed to init");
2208                for id in 0..3 {
2209                    (oversized, _, _, _) = oversized
2210                        .append(1, TestEntry::new(id, 0, 0), &[id as u8; 16])
2211                        .await
2212                        .expect("failed to append");
2213                }
2214                (oversized, _, _, _) = oversized
2215                    .append(2, TestEntry::new(3, 0, 0), &[3; 16])
2216                    .await
2217                    .expect("failed to append later section");
2218                oversized = oversized.sync_all().await.expect("failed to sync");
2219                drop(oversized);
2220
2221                // Checkpoint recovery trusts the completed durability boundary and reads only its
2222                // terminal entry. Arbitrary post-commit bit rot remains a lazy read error.
2223                corrupt_page(
2224                    &context,
2225                    &cfg.index_partition,
2226                    &1u64.to_be_bytes(),
2227                    1,
2228                    TestEntry::SIZE as u64,
2229                )
2230                .await;
2231                let (blob, expected_size) = context
2232                    .open(&cfg.index_partition, &1u64.to_be_bytes())
2233                    .await
2234                    .expect("failed to open index");
2235                let expected = blob
2236                    .read_at(0, expected_size as usize, ReadOptions::default())
2237                    .await
2238                    .expect("failed to snapshot damaged index")
2239                    .coalesce();
2240                drop(blob);
2241
2242                let oversized: Oversized<_, TestEntry, TestValue> =
2243                    Oversized::init_with_checkpoint(
2244                        context.child(restore_child),
2245                        cfg.clone(),
2246                        checkpoint,
2247                    )
2248                    .await
2249                    .expect("checkpoint restore should adopt the covered prefix");
2250                assert!(matches!(
2251                    oversized.get(1, 1).await,
2252                    Err(Error::Runtime(RError::InvalidChecksum))
2253                ));
2254                drop(oversized);
2255
2256                let (blob, actual_size) = context
2257                    .open(&cfg.index_partition, &1u64.to_be_bytes())
2258                    .await
2259                    .expect("failed to reopen index");
2260                assert_eq!(actual_size, expected_size);
2261                let actual = blob
2262                    .read_at(0, actual_size as usize, ReadOptions::default())
2263                    .await
2264                    .expect("failed to read damaged index")
2265                    .coalesce();
2266                assert_eq!(actual.as_ref(), expected.as_ref());
2267            }
2268        });
2269    }
2270
2271    #[test_traced]
2272    fn test_oversized_restore_adopts_rotted_committed_value() {
2273        let executor = deterministic::Runner::default();
2274        executor.start(|context| async move {
2275            // Two committed sections
2276            let mut oversized: Oversized<_, TestEntry, TestValue> =
2277                Oversized::init(context.child("first"), test_cfg(&context))
2278                    .await
2279                    .expect("Failed to init");
2280            let (offset, size);
2281            (oversized, _, offset, size) = oversized
2282                .append(1, TestEntry::new(1, 0, 0), &[1; 16])
2283                .await
2284                .expect("Failed to append");
2285            (oversized, _, _, _) = oversized
2286                .append(2, TestEntry::new(2, 0, 0), &[2; 16])
2287                .await
2288                .expect("Failed to append");
2289            oversized = oversized.sync_all().await.expect("Failed to sync");
2290
2291            // Corrupt entry 1's committed value in place (sizes unchanged)
2292            oversized
2293                .values
2294                .inject(1, offset, vec![0xFF; size as usize])
2295                .await
2296                .expect("Failed to corrupt value");
2297            oversized.values = oversized.values.sync(1).await.expect("Failed to sync");
2298            drop(oversized);
2299
2300            // Restore adopts the section without probing its values: the corruption
2301            // surfaces at read on exactly the affected entry.
2302            let chunk = FixedJournal::<deterministic::Context, TestEntry>::CHUNK_SIZE as u64;
2303            let oversized: Oversized<_, TestEntry, TestValue> = Oversized::init_with_checkpoint(
2304                context.child("second"),
2305                test_cfg(&context),
2306                (2, chunk),
2307            )
2308            .await
2309            .expect("Failed to reinit");
2310            assert!(matches!(
2311                oversized.get_value(1, offset, size).await,
2312                Err(Error::ChecksumMismatch(_, _))
2313            ));
2314            let entry = oversized.get(2, 0).await.expect("Failed to get");
2315            let (offset, size) = entry.value_location();
2316            assert_eq!(
2317                oversized
2318                    .get_value(2, offset, size)
2319                    .await
2320                    .expect("Failed to get value"),
2321                [2; 16]
2322            );
2323            oversized.destroy().await.expect("Failed to destroy");
2324        });
2325    }
2326
2327    #[test_traced]
2328    fn test_oversized_replay_empty_finishes_immediately() {
2329        let executor = deterministic::Runner::default();
2330        executor.start(|context| async move {
2331            let cfg = test_cfg(&context);
2332            let oversized: Oversized<_, TestEntry, TestValue> =
2333                Oversized::init(context, cfg).await.expect("Failed to init");
2334
2335            // An empty journal's reader is exhausted from the start
2336            let replay = oversized
2337                .replay(0, 0, NZUsize!(1024), ReadOptions::default())
2338                .await
2339                .expect("Failed to replay");
2340            let oversized = replay.finish().expect("failed to finish replay");
2341            oversized.destroy().await.expect("Failed to destroy");
2342        });
2343    }
2344
2345    #[test_traced]
2346    fn test_oversized_replay_propagates_read_options() {
2347        let executor = deterministic::Runner::default();
2348        executor.start(|context| async move {
2349            let (context, recordings) = commonware_runtime::mocks::RecordingContext::new(context);
2350            let cfg = test_cfg(&context);
2351            let page_cache = cfg.index_page_cache.clone();
2352            let mut oversized: Oversized<_, TestEntry, TestValue> =
2353                Oversized::init(context, cfg).await.expect("Failed to init");
2354            (oversized, _, _, _) = oversized
2355                .append(1, TestEntry::new(1, 0, 0), &[1; 16])
2356                .await
2357                .expect("Failed to append");
2358            oversized = oversized.sync(1).await.expect("Failed to sync");
2359
2360            // Evict the index page so replay must exercise the backing journal read.
2361            page_cache.clear();
2362            let mut replay = oversized
2363                .replay(1, 0, NZUsize!(1024), ReadOptions::DONT_CACHE)
2364                .await
2365                .expect("Failed to replay");
2366            recordings.clear();
2367
2368            // The adapter must preserve the caller's policy on the lazy refill.
2369            let (section, position, entry) = replay
2370                .next()
2371                .await
2372                .expect("missing replay item")
2373                .expect("Failed to read replay item");
2374            assert_eq!((section, position, entry.id), (1, 0, 1));
2375
2376            let reads = recordings.snapshot().reads;
2377            assert!(!reads.is_empty());
2378            assert!(
2379                reads
2380                    .iter()
2381                    .all(|options| *options == ReadOptions::DONT_CACHE)
2382            );
2383
2384            assert!(replay.next().await.is_none());
2385            replay
2386                .finish()
2387                .expect("failed to finish replay")
2388                .destroy()
2389                .await
2390                .expect("Failed to destroy");
2391        });
2392    }
2393
2394    #[test_traced]
2395    fn test_oversized_replay_finish_before_drain_fails() {
2396        let executor = deterministic::Runner::default();
2397        executor.start(|context| async move {
2398            let cfg = test_cfg(&context);
2399            let mut oversized: Oversized<_, TestEntry, TestValue> =
2400                Oversized::init(context, cfg).await.expect("Failed to init");
2401            (oversized, _, _, _) = oversized
2402                .append(1, TestEntry::new(1, 0, 0), &[1; 16])
2403                .await
2404                .expect("Failed to append");
2405            oversized = oversized.sync(1).await.expect("Failed to sync");
2406
2407            let replay = oversized
2408                .replay(0, 0, NZUsize!(1024), ReadOptions::default())
2409                .await
2410                .expect("Failed to replay");
2411            assert!(matches!(replay.finish(), Err(Error::ReplayFailed)));
2412        });
2413    }
2414
2415    #[test_traced]
2416    fn test_oversized_prune() {
2417        let executor = deterministic::Runner::default();
2418        executor.start(|context| async move {
2419            let cfg = test_cfg(&context);
2420            let mut oversized: Oversized<_, TestEntry, TestValue> =
2421                Oversized::init(context, cfg).await.expect("Failed to init");
2422
2423            // Append to multiple sections
2424            for section in 1u64..=5 {
2425                let value: TestValue = [section as u8; 16];
2426                let entry = TestEntry::new(section, 0, 0);
2427                (oversized, _, _, _) = oversized
2428                    .append(section, entry, &value)
2429                    .await
2430                    .expect("Failed to append");
2431                oversized = oversized.sync(section).await.expect("Failed to sync");
2432            }
2433
2434            // Prune sections < 3
2435            (oversized, _) = oversized.prune(3).await.expect("Failed to prune");
2436
2437            // The public accessor mirrors the guard
2438            assert!(oversized.pruned(1));
2439            assert!(oversized.pruned(2));
2440            assert!(!oversized.pruned(3));
2441
2442            // Sections 1, 2 should be gone
2443            assert!(oversized.get(1, 0).await.is_err());
2444            assert!(oversized.get(2, 0).await.is_err());
2445
2446            // Sections 3, 4, 5 should exist
2447            assert!(oversized.get(3, 0).await.is_ok());
2448            assert!(oversized.get(4, 0).await.is_ok());
2449            assert!(oversized.get(5, 0).await.is_ok());
2450
2451            oversized.destroy().await.expect("Failed to destroy");
2452        });
2453    }
2454
2455    #[test_traced]
2456    fn test_recovery_empty_section() {
2457        let executor = deterministic::Runner::default();
2458        executor.start(|context| async move {
2459            let cfg = test_cfg(&context);
2460
2461            // Create oversized journal
2462            let mut oversized: Oversized<_, TestEntry, TestValue> =
2463                Oversized::init(context.child("first"), cfg.clone())
2464                    .await
2465                    .expect("Failed to init");
2466
2467            // Append to section 2 only (section 1 remains empty after being opened)
2468            let value: TestValue = [42; 16];
2469            let entry = TestEntry::new(1, 0, 0);
2470            (oversized, _, _, _) = oversized
2471                .append(2, entry, &value)
2472                .await
2473                .expect("Failed to append");
2474            oversized = oversized.sync(2).await.expect("Failed to sync");
2475            drop(oversized);
2476
2477            // Reinitialize - recovery should handle the empty/non-existent section 1
2478            let oversized: Oversized<_, TestEntry, TestValue> =
2479                Oversized::init(context.child("second"), cfg)
2480                    .await
2481                    .expect("Failed to reinit");
2482
2483            // Section 2 entry should be valid
2484            let entry = oversized.get(2, 0).await.expect("Failed to get");
2485            assert_eq!(entry.id, 1);
2486
2487            oversized.destroy().await.expect("Failed to destroy");
2488        });
2489    }
2490
2491    #[test_traced]
2492    fn test_recovery_all_entries_invalid() {
2493        let executor = deterministic::Runner::default();
2494        executor.start(|context| async move {
2495            let cfg = test_cfg(&context);
2496
2497            // Create and populate
2498            let mut oversized: Oversized<_, TestEntry, TestValue> =
2499                Oversized::init(context.child("first"), cfg.clone())
2500                    .await
2501                    .expect("Failed to init");
2502
2503            // Append 5 entries
2504            for i in 0..5u8 {
2505                let value: TestValue = [i; 16];
2506                let entry = TestEntry::new(i as u64, 0, 0);
2507                (oversized, _, _, _) = oversized
2508                    .append(1, entry, &value)
2509                    .await
2510                    .expect("Failed to append");
2511            }
2512            oversized = oversized.sync(1).await.expect("Failed to sync");
2513            drop(oversized);
2514
2515            // Truncate glob to 0 bytes - ALL entries become invalid
2516            let (blob, _) = context
2517                .open(&cfg.value_partition, &1u64.to_be_bytes())
2518                .await
2519                .expect("Failed to open blob");
2520            blob.resize(0).await.expect("Failed to truncate");
2521            blob.sync().await.expect("Failed to sync");
2522            drop(blob);
2523
2524            // Reinitialize - should recover and rewind index to 0
2525            let mut oversized: Oversized<_, TestEntry, TestValue> =
2526                Oversized::init(context.child("second"), cfg)
2527                    .await
2528                    .expect("Failed to reinit");
2529
2530            // No entries should be accessible
2531            let result = oversized.get(1, 0).await;
2532            assert!(result.is_err());
2533
2534            // Should be able to append after recovery
2535            let value: TestValue = [99; 16];
2536            let entry = TestEntry::new(100, 0, 0);
2537            let (pos, offset, size);
2538            (oversized, pos, offset, size) = oversized
2539                .append(1, entry, &value)
2540                .await
2541                .expect("Failed to append after recovery");
2542            assert_eq!(pos, 0);
2543
2544            let retrieved = oversized.get(1, 0).await.expect("Failed to get");
2545            assert_eq!(retrieved.id, 100);
2546            let retrieved_value = oversized
2547                .get_value(1, offset, size)
2548                .await
2549                .expect("Failed to get value");
2550            assert_eq!(retrieved_value, value);
2551
2552            oversized.destroy().await.expect("Failed to destroy");
2553        });
2554    }
2555
2556    #[test_traced]
2557    fn test_recovery_multiple_sections_mixed_validity() {
2558        let executor = deterministic::Runner::default();
2559        executor.start(|context| async move {
2560            let cfg = test_cfg(&context);
2561
2562            // Create and populate multiple sections
2563            let mut oversized: Oversized<_, TestEntry, TestValue> =
2564                Oversized::init(context.child("first"), cfg.clone())
2565                    .await
2566                    .expect("Failed to init");
2567
2568            // Section 1: 3 entries
2569            let mut section1_locations = Vec::new();
2570            for i in 0..3u8 {
2571                let value: TestValue = [i; 16];
2572                let entry = TestEntry::new(i as u64, 0, 0);
2573                let (position, offset, size);
2574                (oversized, position, offset, size) = oversized
2575                    .append(1, entry, &value)
2576                    .await
2577                    .expect("Failed to append");
2578                section1_locations.push((position, offset, size));
2579            }
2580            oversized = oversized.sync(1).await.expect("Failed to sync");
2581
2582            // Section 2: 5 entries
2583            let mut section2_locations = Vec::new();
2584            for i in 0..5u8 {
2585                let value: TestValue = [10 + i; 16];
2586                let entry = TestEntry::new(10 + i as u64, 0, 0);
2587                let (position, offset, size);
2588                (oversized, position, offset, size) = oversized
2589                    .append(2, entry, &value)
2590                    .await
2591                    .expect("Failed to append");
2592                section2_locations.push((position, offset, size));
2593            }
2594            oversized = oversized.sync(2).await.expect("Failed to sync");
2595
2596            // Section 3: 2 entries
2597            for i in 0..2u8 {
2598                let value: TestValue = [20 + i; 16];
2599                let entry = TestEntry::new(20 + i as u64, 0, 0);
2600                (oversized, _, _, _) = oversized
2601                    .append(3, entry, &value)
2602                    .await
2603                    .expect("Failed to append");
2604            }
2605            oversized = oversized.sync(3).await.expect("Failed to sync");
2606            drop(oversized);
2607
2608            // Truncate section 1 glob to keep only first entry
2609            let (blob, _) = context
2610                .open(&cfg.value_partition, &1u64.to_be_bytes())
2611                .await
2612                .expect("Failed to open blob");
2613            let keep_size = byte_end(section1_locations[0].1, section1_locations[0].2);
2614            blob.resize(keep_size).await.expect("Failed to truncate");
2615            blob.sync().await.expect("Failed to sync");
2616            drop(blob);
2617
2618            // Truncate section 2 glob to keep first 3 entries
2619            let (blob, _) = context
2620                .open(&cfg.value_partition, &2u64.to_be_bytes())
2621                .await
2622                .expect("Failed to open blob");
2623            let keep_size = byte_end(section2_locations[2].1, section2_locations[2].2);
2624            blob.resize(keep_size).await.expect("Failed to truncate");
2625            blob.sync().await.expect("Failed to sync");
2626            drop(blob);
2627
2628            // Section 3 remains intact
2629
2630            // Reinitialize
2631            let oversized: Oversized<_, TestEntry, TestValue> =
2632                Oversized::init(context.child("second"), cfg)
2633                    .await
2634                    .expect("Failed to reinit");
2635
2636            // Section 1: only position 0 valid
2637            assert!(oversized.get(1, 0).await.is_ok());
2638            assert!(oversized.get(1, 1).await.is_err());
2639            assert!(oversized.get(1, 2).await.is_err());
2640
2641            // Section 2: positions 0,1,2 valid
2642            assert!(oversized.get(2, 0).await.is_ok());
2643            assert!(oversized.get(2, 1).await.is_ok());
2644            assert!(oversized.get(2, 2).await.is_ok());
2645            assert!(oversized.get(2, 3).await.is_err());
2646            assert!(oversized.get(2, 4).await.is_err());
2647
2648            // Section 3: both positions valid
2649            assert!(oversized.get(3, 0).await.is_ok());
2650            assert!(oversized.get(3, 1).await.is_ok());
2651
2652            oversized.destroy().await.expect("Failed to destroy");
2653        });
2654    }
2655
2656    #[test_traced]
2657    fn test_recovery_corrupted_last_index_entry() {
2658        let executor = deterministic::Runner::default();
2659        executor.start(|context| async move {
2660            // Use page size = entry size so each entry is on its own page.
2661            // This allows corrupting just the last entry's page without affecting others.
2662            // Physical page size = TestEntry::SIZE (20) + 12 (CRC record) = 32 bytes.
2663            let cfg = entry_cfg(&context);
2664
2665            // Create and populate
2666            let mut oversized: Oversized<_, TestEntry, TestValue> =
2667                Oversized::init(context.child("first"), cfg.clone())
2668                    .await
2669                    .expect("Failed to init");
2670
2671            // Append 5 entries (each on its own page)
2672            for i in 0..5u8 {
2673                let value: TestValue = [i; 16];
2674                let entry = TestEntry::new(i as u64, 0, 0);
2675                (oversized, _, _, _) = oversized
2676                    .append(1, entry, &value)
2677                    .await
2678                    .expect("Failed to append");
2679            }
2680            oversized = oversized.sync(1).await.expect("Failed to sync");
2681            drop(oversized);
2682
2683            // Corrupt the last page's CRC to trigger page-level integrity failure
2684            let (blob, size) = context
2685                .open(&cfg.index_partition, &1u64.to_be_bytes())
2686                .await
2687                .expect("Failed to open blob");
2688
2689            // Physical page size = 20 + 12 = 32 bytes
2690            // 5 entries = 5 pages = 160 bytes total
2691            // Last page CRC starts at offset 160 - 12 = 148
2692            assert_eq!(size, 160);
2693            let last_page_crc_offset = size - 12;
2694            blob.write_at(last_page_crc_offset, vec![0xFF; 12], WriteOptions::SYNC)
2695                .await
2696                .expect("Failed to corrupt");
2697            drop(blob);
2698
2699            // Reinitialize - should detect page corruption and truncate
2700            let mut oversized: Oversized<_, TestEntry, TestValue> =
2701                Oversized::init(context.child("second"), cfg)
2702                    .await
2703                    .expect("Failed to reinit");
2704
2705            // First 4 entries should be valid (on pages 0-3)
2706            for i in 0..4u8 {
2707                let entry = oversized.get(1, i as u64).await.expect("Failed to get");
2708                assert_eq!(entry.id, i as u64);
2709            }
2710
2711            // Entry 4 should be gone (its page was corrupted)
2712            assert!(oversized.get(1, 4).await.is_err());
2713
2714            // Should be able to append after recovery
2715            let value: TestValue = [99; 16];
2716            let entry = TestEntry::new(100, 0, 0);
2717            let (pos, offset, size);
2718            (oversized, pos, offset, size) = oversized
2719                .append(1, entry, &value)
2720                .await
2721                .expect("Failed to append after recovery");
2722            assert_eq!(pos, 4);
2723
2724            let retrieved = oversized.get(1, 4).await.expect("Failed to get");
2725            assert_eq!(retrieved.id, 100);
2726            let retrieved_value = oversized
2727                .get_value(1, offset, size)
2728                .await
2729                .expect("Failed to get value");
2730            assert_eq!(retrieved_value, value);
2731
2732            oversized.destroy().await.expect("Failed to destroy");
2733        });
2734    }
2735
2736    #[test_traced]
2737    fn test_recovery_all_entries_valid() {
2738        let executor = deterministic::Runner::default();
2739        executor.start(|context| async move {
2740            let cfg = test_cfg(&context);
2741
2742            // Create and populate
2743            let mut oversized: Oversized<_, TestEntry, TestValue> =
2744                Oversized::init(context.child("first"), cfg.clone())
2745                    .await
2746                    .expect("Failed to init");
2747
2748            // Append entries to multiple sections
2749            for section in 1u64..=3 {
2750                for i in 0..10u8 {
2751                    let value: TestValue = [(section as u8) * 10 + i; 16];
2752                    let entry = TestEntry::new(section * 100 + i as u64, 0, 0);
2753                    (oversized, _, _, _) = oversized
2754                        .append(section, entry, &value)
2755                        .await
2756                        .expect("Failed to append");
2757                }
2758                oversized = oversized.sync(section).await.expect("Failed to sync");
2759            }
2760            drop(oversized);
2761
2762            // Reinitialize with no corruption - should be fast
2763            let oversized: Oversized<_, TestEntry, TestValue> =
2764                Oversized::init(context.child("second"), cfg)
2765                    .await
2766                    .expect("Failed to reinit");
2767
2768            // All entries should be valid
2769            for section in 1u64..=3 {
2770                for i in 0..10u8 {
2771                    let entry = oversized
2772                        .get(section, i as u64)
2773                        .await
2774                        .expect("Failed to get");
2775                    assert_eq!(entry.id, section * 100 + i as u64);
2776                }
2777            }
2778
2779            oversized.destroy().await.expect("Failed to destroy");
2780        });
2781    }
2782
2783    #[test_traced]
2784    fn test_recovery_single_entry_invalid() {
2785        let executor = deterministic::Runner::default();
2786        executor.start(|context| async move {
2787            let cfg = test_cfg(&context);
2788
2789            // Create and populate with single entry
2790            let mut oversized: Oversized<_, TestEntry, TestValue> =
2791                Oversized::init(context.child("first"), cfg.clone())
2792                    .await
2793                    .expect("Failed to init");
2794
2795            let value: TestValue = [42; 16];
2796            let entry = TestEntry::new(1, 0, 0);
2797            (oversized, _, _, _) = oversized
2798                .append(1, entry, &value)
2799                .await
2800                .expect("Failed to append");
2801            oversized = oversized.sync(1).await.expect("Failed to sync");
2802            drop(oversized);
2803
2804            // Truncate glob to 0 - single entry becomes invalid
2805            let (blob, _) = context
2806                .open(&cfg.value_partition, &1u64.to_be_bytes())
2807                .await
2808                .expect("Failed to open blob");
2809            blob.resize(0).await.expect("Failed to truncate");
2810            blob.sync().await.expect("Failed to sync");
2811            drop(blob);
2812
2813            // Reinitialize
2814            let oversized: Oversized<_, TestEntry, TestValue> =
2815                Oversized::init(context.child("second"), cfg)
2816                    .await
2817                    .expect("Failed to reinit");
2818
2819            // Entry should be gone
2820            assert!(oversized.get(1, 0).await.is_err());
2821
2822            oversized.destroy().await.expect("Failed to destroy");
2823        });
2824    }
2825
2826    #[test_traced]
2827    fn test_recovery_last_entry_off_by_one() {
2828        let executor = deterministic::Runner::default();
2829        executor.start(|context| async move {
2830            let cfg = test_cfg(&context);
2831
2832            // Create and populate
2833            let mut oversized: Oversized<_, TestEntry, TestValue> =
2834                Oversized::init(context.child("first"), cfg.clone())
2835                    .await
2836                    .expect("Failed to init");
2837
2838            let mut locations = Vec::new();
2839            for i in 0..3u8 {
2840                let value: TestValue = [i; 16];
2841                let entry = TestEntry::new(i as u64, 0, 0);
2842                let (position, offset, size);
2843                (oversized, position, offset, size) = oversized
2844                    .append(1, entry, &value)
2845                    .await
2846                    .expect("Failed to append");
2847                locations.push((position, offset, size));
2848            }
2849            oversized = oversized.sync(1).await.expect("Failed to sync");
2850            drop(oversized);
2851
2852            // Truncate glob to be off by 1 byte from last entry
2853            let (blob, _) = context
2854                .open(&cfg.value_partition, &1u64.to_be_bytes())
2855                .await
2856                .expect("Failed to open blob");
2857
2858            // Last entry needs: offset + size bytes
2859            // Truncate to offset + size - 1 (missing 1 byte)
2860            let last = &locations[2];
2861            let truncate_to = byte_end(last.1, last.2) - 1;
2862            blob.resize(truncate_to).await.expect("Failed to truncate");
2863            blob.sync().await.expect("Failed to sync");
2864            drop(blob);
2865
2866            // Reinitialize
2867            let mut oversized: Oversized<_, TestEntry, TestValue> =
2868                Oversized::init(context.child("second"), cfg)
2869                    .await
2870                    .expect("Failed to reinit");
2871
2872            // First 2 entries should be valid
2873            assert!(oversized.get(1, 0).await.is_ok());
2874            assert!(oversized.get(1, 1).await.is_ok());
2875
2876            // Entry 2 should be gone (truncated)
2877            assert!(oversized.get(1, 2).await.is_err());
2878
2879            // Should be able to append after recovery
2880            let value: TestValue = [99; 16];
2881            let entry = TestEntry::new(100, 0, 0);
2882            let (pos, offset, size);
2883            (oversized, pos, offset, size) = oversized
2884                .append(1, entry, &value)
2885                .await
2886                .expect("Failed to append after recovery");
2887            assert_eq!(pos, 2);
2888
2889            let retrieved = oversized.get(1, 2).await.expect("Failed to get");
2890            assert_eq!(retrieved.id, 100);
2891            let retrieved_value = oversized
2892                .get_value(1, offset, size)
2893                .await
2894                .expect("Failed to get value");
2895            assert_eq!(retrieved_value, value);
2896
2897            oversized.destroy().await.expect("Failed to destroy");
2898        });
2899    }
2900
2901    #[test_traced]
2902    fn test_recovery_glob_missing_entirely() {
2903        let executor = deterministic::Runner::default();
2904        executor.start(|context| async move {
2905            let cfg = test_cfg(&context);
2906
2907            // Create and populate
2908            let mut oversized: Oversized<_, TestEntry, TestValue> =
2909                Oversized::init(context.child("first"), cfg.clone())
2910                    .await
2911                    .expect("Failed to init");
2912
2913            for i in 0..3u8 {
2914                let value: TestValue = [i; 16];
2915                let entry = TestEntry::new(i as u64, 0, 0);
2916                (oversized, _, _, _) = oversized
2917                    .append(1, entry, &value)
2918                    .await
2919                    .expect("Failed to append");
2920            }
2921            oversized = oversized.sync(1).await.expect("Failed to sync");
2922            drop(oversized);
2923
2924            // Delete the glob file entirely
2925            context
2926                .remove(&cfg.value_partition, Some(&1u64.to_be_bytes()))
2927                .await
2928                .expect("Failed to remove");
2929
2930            // Reinitialize - glob size will be 0, all entries invalid
2931            let oversized: Oversized<_, TestEntry, TestValue> =
2932                Oversized::init(context.child("second"), cfg)
2933                    .await
2934                    .expect("Failed to reinit");
2935
2936            // All entries should be gone
2937            assert!(oversized.get(1, 0).await.is_err());
2938            assert!(oversized.get(1, 1).await.is_err());
2939            assert!(oversized.get(1, 2).await.is_err());
2940
2941            oversized.destroy().await.expect("Failed to destroy");
2942        });
2943    }
2944
2945    #[test_traced]
2946    fn test_recovery_can_append_after_recovery() {
2947        let executor = deterministic::Runner::default();
2948        executor.start(|context| async move {
2949            let cfg = test_cfg(&context);
2950
2951            // Create and populate
2952            let mut oversized: Oversized<_, TestEntry, TestValue> =
2953                Oversized::init(context.child("first"), cfg.clone())
2954                    .await
2955                    .expect("Failed to init");
2956
2957            let mut locations = Vec::new();
2958            for i in 0..5u8 {
2959                let value: TestValue = [i; 16];
2960                let entry = TestEntry::new(i as u64, 0, 0);
2961                let (position, offset, size);
2962                (oversized, position, offset, size) = oversized
2963                    .append(1, entry, &value)
2964                    .await
2965                    .expect("Failed to append");
2966                locations.push((position, offset, size));
2967            }
2968            oversized = oversized.sync(1).await.expect("Failed to sync");
2969            drop(oversized);
2970
2971            // Truncate glob to keep only first 2 entries
2972            let (blob, _) = context
2973                .open(&cfg.value_partition, &1u64.to_be_bytes())
2974                .await
2975                .expect("Failed to open blob");
2976            let keep_size = byte_end(locations[1].1, locations[1].2);
2977            blob.resize(keep_size).await.expect("Failed to truncate");
2978            blob.sync().await.expect("Failed to sync");
2979            drop(blob);
2980
2981            // Reinitialize
2982            let mut oversized: Oversized<_, TestEntry, TestValue> =
2983                Oversized::init(context.child("second"), cfg.clone())
2984                    .await
2985                    .expect("Failed to reinit");
2986
2987            // Verify first 2 entries exist
2988            assert!(oversized.get(1, 0).await.is_ok());
2989            assert!(oversized.get(1, 1).await.is_ok());
2990            assert!(oversized.get(1, 2).await.is_err());
2991
2992            // Append new entries after recovery
2993            for i in 10..15u8 {
2994                let value: TestValue = [i; 16];
2995                let entry = TestEntry::new(i as u64, 0, 0);
2996                (oversized, _, _, _) = oversized
2997                    .append(1, entry, &value)
2998                    .await
2999                    .expect("Failed to append after recovery");
3000            }
3001            oversized = oversized.sync(1).await.expect("Failed to sync");
3002
3003            // Verify new entries at positions 2, 3, 4, 5, 6
3004            for i in 0..5u8 {
3005                let entry = oversized
3006                    .get(1, 2 + i as u64)
3007                    .await
3008                    .expect("Failed to get new entry");
3009                assert_eq!(entry.id, (10 + i) as u64);
3010            }
3011
3012            oversized.destroy().await.expect("Failed to destroy");
3013        });
3014    }
3015
3016    #[test_traced]
3017    fn test_recovery_glob_pruned_but_index_not() {
3018        let executor = deterministic::Runner::default();
3019        executor.start(|context| async move {
3020            let cfg = test_cfg(&context);
3021
3022            // Create and populate multiple sections
3023            let mut oversized: Oversized<_, TestEntry, TestValue> =
3024                Oversized::init(context.child("first"), cfg.clone())
3025                    .await
3026                    .expect("Failed to init");
3027
3028            for section in 1u64..=3 {
3029                let value: TestValue = [section as u8; 16];
3030                let entry = TestEntry::new(section, 0, 0);
3031                (oversized, _, _, _) = oversized
3032                    .append(section, entry, &value)
3033                    .await
3034                    .expect("Failed to append");
3035                oversized = oversized.sync(section).await.expect("Failed to sync");
3036            }
3037            drop(oversized);
3038
3039            // Simulate crash during prune: prune ONLY the glob, not the index
3040            // This creates the "glob pruned but index not" scenario
3041            use crate::journal::segmented::glob::{Config as GlobConfig, Glob};
3042            let glob_cfg = GlobConfig {
3043                partition: cfg.value_partition.clone(),
3044                compression: cfg.compression,
3045                codec_config: (),
3046                write_buffer: cfg.value_write_buffer,
3047            };
3048            let mut glob: Glob<_, TestValue> = Glob::init(context.child("glob"), glob_cfg)
3049                .await
3050                .expect("Failed to init glob");
3051            (glob, _) = glob.prune(2).await.expect("Failed to prune glob");
3052            glob = glob.sync_all().await.expect("Failed to sync glob");
3053            drop(glob);
3054
3055            // Reinitialize - should recover gracefully with warning
3056            // Index section 1 will be rewound to 0 entries
3057            let oversized: Oversized<_, TestEntry, TestValue> =
3058                Oversized::init(context.child("second"), cfg.clone())
3059                    .await
3060                    .expect("Failed to reinit");
3061
3062            // Section 1 entries should be gone (index rewound due to glob pruned)
3063            assert!(oversized.get(1, 0).await.is_err());
3064
3065            // Sections 2 and 3 should still be valid
3066            assert!(oversized.get(2, 0).await.is_ok());
3067            assert!(oversized.get(3, 0).await.is_ok());
3068
3069            oversized.destroy().await.expect("Failed to destroy");
3070        });
3071    }
3072
3073    #[test_traced]
3074    fn test_recovery_index_partition_deleted() {
3075        let executor = deterministic::Runner::default();
3076        executor.start(|context| async move {
3077            let cfg = test_cfg(&context);
3078
3079            // Create and populate multiple sections
3080            let mut oversized: Oversized<_, TestEntry, TestValue> =
3081                Oversized::init(context.child("first"), cfg.clone())
3082                    .await
3083                    .expect("Failed to init");
3084
3085            for section in 1u64..=3 {
3086                let value: TestValue = [section as u8; 16];
3087                let entry = TestEntry::new(section, 0, 0);
3088                (oversized, _, _, _) = oversized
3089                    .append(section, entry, &value)
3090                    .await
3091                    .expect("Failed to append");
3092                oversized = oversized.sync(section).await.expect("Failed to sync");
3093            }
3094            drop(oversized);
3095
3096            // Delete index blob for section 2 (simulate corruption/loss)
3097            context
3098                .remove(&cfg.index_partition, Some(&2u64.to_be_bytes()))
3099                .await
3100                .expect("Failed to remove index");
3101
3102            // Reinitialize - should handle gracefully
3103            // Section 2 is gone from index, orphan data in glob is acceptable
3104            let oversized: Oversized<_, TestEntry, TestValue> =
3105                Oversized::init(context.child("second"), cfg.clone())
3106                    .await
3107                    .expect("Failed to reinit");
3108
3109            // Section 1 and 3 should still be valid
3110            assert!(oversized.get(1, 0).await.is_ok());
3111            assert!(oversized.get(3, 0).await.is_ok());
3112
3113            // Section 2 should be gone (index file deleted)
3114            assert!(oversized.get(2, 0).await.is_err());
3115
3116            oversized.destroy().await.expect("Failed to destroy");
3117        });
3118    }
3119
3120    #[test_traced]
3121    fn test_recovery_index_synced_but_glob_not() {
3122        let executor = deterministic::Runner::default();
3123        executor.start(|context| async move {
3124            let cfg = test_cfg(&context);
3125
3126            // Create and populate
3127            let mut oversized: Oversized<_, TestEntry, TestValue> =
3128                Oversized::init(context.child("first"), cfg.clone())
3129                    .await
3130                    .expect("Failed to init");
3131
3132            // Append entries and sync
3133            let mut locations = Vec::new();
3134            for i in 0..3u8 {
3135                let value: TestValue = [i; 16];
3136                let entry = TestEntry::new(i as u64, 0, 0);
3137                let (position, offset, size);
3138                (oversized, position, offset, size) = oversized
3139                    .append(1, entry, &value)
3140                    .await
3141                    .expect("Failed to append");
3142                locations.push((position, offset, size));
3143            }
3144            oversized = oversized.sync(1).await.expect("Failed to sync");
3145
3146            // Add more entries WITHOUT syncing (simulates unsynced writes)
3147            for i in 10..15u8 {
3148                let value: TestValue = [i; 16];
3149                let entry = TestEntry::new(i as u64, 0, 0);
3150                (oversized, _, _, _) = oversized
3151                    .append(1, entry, &value)
3152                    .await
3153                    .expect("Failed to append");
3154            }
3155            // Note: NOT calling sync() here
3156            drop(oversized);
3157
3158            // Simulate crash where index was synced but glob wasn't:
3159            // Truncate glob back to the synced size (3 entries)
3160            let (blob, _) = context
3161                .open(&cfg.value_partition, &1u64.to_be_bytes())
3162                .await
3163                .expect("Failed to open blob");
3164            let synced_size = byte_end(locations[2].1, locations[2].2);
3165            blob.resize(synced_size).await.expect("Failed to truncate");
3166            blob.sync().await.expect("Failed to sync");
3167            drop(blob);
3168
3169            // Reinitialize - should rewind index to match glob
3170            let oversized: Oversized<_, TestEntry, TestValue> =
3171                Oversized::init(context.child("second"), cfg)
3172                    .await
3173                    .expect("Failed to reinit");
3174
3175            // First 3 entries should be valid
3176            for i in 0..3u8 {
3177                let entry = oversized.get(1, i as u64).await.expect("Failed to get");
3178                assert_eq!(entry.id, i as u64);
3179            }
3180
3181            // Entries 3-7 should be gone (unsynced, index rewound)
3182            assert!(oversized.get(1, 3).await.is_err());
3183
3184            oversized.destroy().await.expect("Failed to destroy");
3185        });
3186    }
3187
3188    #[test_traced]
3189    fn test_recovery_glob_synced_but_index_not() {
3190        let executor = deterministic::Runner::default();
3191        executor.start(|context| async move {
3192            // Use page size = entry size so each entry is exactly one page.
3193            // This allows truncating by entry count to equal truncating by full pages,
3194            // maintaining page-level integrity.
3195            let cfg = entry_cfg(&context);
3196
3197            // Create and populate
3198            let mut oversized: Oversized<_, TestEntry, TestValue> =
3199                Oversized::init(context.child("first"), cfg.clone())
3200                    .await
3201                    .expect("Failed to init");
3202
3203            // Append entries and sync
3204            let mut locations = Vec::new();
3205            for i in 0..3u8 {
3206                let value: TestValue = [i; 16];
3207                let entry = TestEntry::new(i as u64, 0, 0);
3208                let (position, offset, size);
3209                (oversized, position, offset, size) = oversized
3210                    .append(1, entry, &value)
3211                    .await
3212                    .expect("Failed to append");
3213                locations.push((position, offset, size));
3214            }
3215            oversized = oversized.sync(1).await.expect("Failed to sync");
3216            drop(oversized);
3217
3218            // Simulate crash: truncate INDEX but leave GLOB intact
3219            // This creates orphan data in glob (glob ahead of index)
3220            let (blob, _size) = context
3221                .open(&cfg.index_partition, &1u64.to_be_bytes())
3222                .await
3223                .expect("Failed to open blob");
3224
3225            // Keep only first 2 index entries (2 full pages)
3226            // Physical page size = logical (20) + CRC record (12) = 32 bytes
3227            let physical_page_size = (TestEntry::SIZE + 12) as u64;
3228            blob.resize(2 * physical_page_size)
3229                .await
3230                .expect("Failed to truncate");
3231            blob.sync().await.expect("Failed to sync");
3232            drop(blob);
3233
3234            // Reinitialize - glob has orphan data from entry 3
3235            let mut oversized: Oversized<_, TestEntry, TestValue> =
3236                Oversized::init(context.child("second"), cfg.clone())
3237                    .await
3238                    .expect("Failed to reinit");
3239
3240            // First 2 entries should be valid
3241            for i in 0..2u8 {
3242                let (position, offset, size) = locations[i as usize];
3243                let entry = oversized.get(1, position).await.expect("Failed to get");
3244                assert_eq!(entry.id, i as u64);
3245
3246                let value = oversized
3247                    .get_value(1, offset, size)
3248                    .await
3249                    .expect("Failed to get value");
3250                assert_eq!(value, [i; 16]);
3251            }
3252
3253            // Entry at position 2 should fail (index was truncated)
3254            assert!(oversized.get(1, 2).await.is_err());
3255
3256            // Append new entries - should work despite orphan data in glob
3257            let mut new_locations = Vec::new();
3258            for i in 10..13u8 {
3259                let value: TestValue = [i; 16];
3260                let entry = TestEntry::new(i as u64, 0, 0);
3261                let (position, offset, size);
3262                (oversized, position, offset, size) = oversized
3263                    .append(1, entry, &value)
3264                    .await
3265                    .expect("Failed to append after recovery");
3266
3267                // New entries start at position 2 (after the 2 valid entries)
3268                assert_eq!(position, (i - 10 + 2) as u64);
3269                new_locations.push((position, offset, size, i));
3270
3271                // Verify we can read the new entry
3272                let retrieved = oversized.get(1, position).await.expect("Failed to get");
3273                assert_eq!(retrieved.id, i as u64);
3274
3275                let retrieved_value = oversized
3276                    .get_value(1, offset, size)
3277                    .await
3278                    .expect("Failed to get value");
3279                assert_eq!(retrieved_value, value);
3280            }
3281
3282            // Sync and restart again to verify persistence with orphan data
3283            oversized = oversized.sync(1).await.expect("Failed to sync");
3284            drop(oversized);
3285
3286            // Reinitialize after adding data on top of orphan glob data
3287            let oversized: Oversized<_, TestEntry, TestValue> =
3288                Oversized::init(context.child("third"), cfg)
3289                    .await
3290                    .expect("Failed to reinit after append");
3291
3292            // Read all valid entries in the index
3293            // First 2 entries from original data
3294            for i in 0..2u8 {
3295                let (position, offset, size) = locations[i as usize];
3296                let entry = oversized.get(1, position).await.expect("Failed to get");
3297                assert_eq!(entry.id, i as u64);
3298
3299                let value = oversized
3300                    .get_value(1, offset, size)
3301                    .await
3302                    .expect("Failed to get value");
3303                assert_eq!(value, [i; 16]);
3304            }
3305
3306            // New entries added after recovery
3307            for (position, offset, size, expected_id) in &new_locations {
3308                let entry = oversized
3309                    .get(1, *position)
3310                    .await
3311                    .expect("Failed to get new entry after restart");
3312                assert_eq!(entry.id, *expected_id as u64);
3313
3314                let value = oversized
3315                    .get_value(1, *offset, *size)
3316                    .await
3317                    .expect("Failed to get new value after restart");
3318                assert_eq!(value, [*expected_id; 16]);
3319            }
3320
3321            // Verify total entry count: 2 original + 3 new = 5
3322            assert!(oversized.get(1, 4).await.is_ok());
3323            assert!(oversized.get(1, 5).await.is_err());
3324
3325            oversized.destroy().await.expect("Failed to destroy");
3326        });
3327    }
3328
3329    #[test_traced]
3330    fn test_recovery_partial_index_entry() {
3331        let executor = deterministic::Runner::default();
3332        executor.start(|context| async move {
3333            let cfg = test_cfg(&context);
3334
3335            // Create and populate
3336            let mut oversized: Oversized<_, TestEntry, TestValue> =
3337                Oversized::init(context.child("first"), cfg.clone())
3338                    .await
3339                    .expect("Failed to init");
3340
3341            // Append 3 entries
3342            for i in 0..3u8 {
3343                let value: TestValue = [i; 16];
3344                let entry = TestEntry::new(i as u64, 0, 0);
3345                (oversized, _, _, _) = oversized
3346                    .append(1, entry, &value)
3347                    .await
3348                    .expect("Failed to append");
3349            }
3350            oversized = oversized.sync(1).await.expect("Failed to sync");
3351            drop(oversized);
3352
3353            // Simulate crash during write: truncate index to partial entry
3354            // Each entry is TestEntry::SIZE (20) + 4 (CRC32) = 24 bytes
3355            // Truncate to 3 full entries + 10 bytes of partial entry
3356            let (blob, _) = context
3357                .open(&cfg.index_partition, &1u64.to_be_bytes())
3358                .await
3359                .expect("Failed to open blob");
3360            let partial_size = 3 * 24 + 10; // 3 full entries + partial
3361            blob.resize(partial_size).await.expect("Failed to resize");
3362            blob.sync().await.expect("Failed to sync");
3363            drop(blob);
3364
3365            // Reinitialize - should handle partial entry gracefully
3366            let mut oversized: Oversized<_, TestEntry, TestValue> =
3367                Oversized::init(context.child("second"), cfg.clone())
3368                    .await
3369                    .expect("Failed to reinit");
3370
3371            // First 3 entries should still be valid
3372            for i in 0..3u8 {
3373                let entry = oversized.get(1, i as u64).await.expect("Failed to get");
3374                assert_eq!(entry.id, i as u64);
3375            }
3376
3377            // Entry 3 should not exist (partial entry was removed)
3378            assert!(oversized.get(1, 3).await.is_err());
3379
3380            // Append new entry after recovery
3381            let value: TestValue = [42; 16];
3382            let entry = TestEntry::new(100, 0, 0);
3383            let (pos, offset, size);
3384            (oversized, pos, offset, size) = oversized
3385                .append(1, entry, &value)
3386                .await
3387                .expect("Failed to append after recovery");
3388            assert_eq!(pos, 3);
3389
3390            // Verify we can read the new entry
3391            let retrieved = oversized.get(1, 3).await.expect("Failed to get new entry");
3392            assert_eq!(retrieved.id, 100);
3393            let retrieved_value = oversized
3394                .get_value(1, offset, size)
3395                .await
3396                .expect("Failed to get new value");
3397            assert_eq!(retrieved_value, value);
3398
3399            oversized.destroy().await.expect("Failed to destroy");
3400        });
3401    }
3402
3403    #[test_traced]
3404    fn test_recovery_only_partial_entry() {
3405        let executor = deterministic::Runner::default();
3406        executor.start(|context| async move {
3407            let cfg = test_cfg(&context);
3408
3409            // Create and populate with single entry
3410            let mut oversized: Oversized<_, TestEntry, TestValue> =
3411                Oversized::init(context.child("first"), cfg.clone())
3412                    .await
3413                    .expect("Failed to init");
3414
3415            let value: TestValue = [42; 16];
3416            let entry = TestEntry::new(1, 0, 0);
3417            (oversized, _, _, _) = oversized
3418                .append(1, entry, &value)
3419                .await
3420                .expect("Failed to append");
3421            oversized = oversized.sync(1).await.expect("Failed to sync");
3422            drop(oversized);
3423
3424            // Truncate index to only partial data (less than one full entry)
3425            let (blob, _) = context
3426                .open(&cfg.index_partition, &1u64.to_be_bytes())
3427                .await
3428                .expect("Failed to open blob");
3429            blob.resize(10).await.expect("Failed to resize"); // Less than chunk size
3430            blob.sync().await.expect("Failed to sync");
3431            drop(blob);
3432
3433            // Reinitialize - should handle gracefully (rewind to 0)
3434            let mut oversized: Oversized<_, TestEntry, TestValue> =
3435                Oversized::init(context.child("second"), cfg.clone())
3436                    .await
3437                    .expect("Failed to reinit");
3438
3439            // No entries should exist
3440            assert!(oversized.get(1, 0).await.is_err());
3441
3442            // Should be able to append after recovery
3443            let value: TestValue = [99; 16];
3444            let entry = TestEntry::new(100, 0, 0);
3445            let (pos, offset, size);
3446            (oversized, pos, offset, size) = oversized
3447                .append(1, entry, &value)
3448                .await
3449                .expect("Failed to append after recovery");
3450            assert_eq!(pos, 0);
3451
3452            let retrieved = oversized.get(1, 0).await.expect("Failed to get");
3453            assert_eq!(retrieved.id, 100);
3454            let retrieved_value = oversized
3455                .get_value(1, offset, size)
3456                .await
3457                .expect("Failed to get value");
3458            assert_eq!(retrieved_value, value);
3459
3460            oversized.destroy().await.expect("Failed to destroy");
3461        });
3462    }
3463
3464    #[test_traced]
3465    fn test_recovery_crash_during_rewind_index_ahead() {
3466        // Simulates crash where index was rewound but glob wasn't
3467        let executor = deterministic::Runner::default();
3468        executor.start(|context| async move {
3469            // Use page size = entry size so each entry is exactly one page.
3470            // This allows truncating by entry count to equal truncating by full pages,
3471            // maintaining page-level integrity.
3472            let cfg = entry_cfg(&context);
3473
3474            // Create and populate
3475            let mut oversized: Oversized<_, TestEntry, TestValue> =
3476                Oversized::init(context.child("first"), cfg.clone())
3477                    .await
3478                    .expect("Failed to init");
3479
3480            let mut locations = Vec::new();
3481            for i in 0..5u8 {
3482                let value: TestValue = [i; 16];
3483                let entry = TestEntry::new(i as u64, 0, 0);
3484                let (position, offset, size);
3485                (oversized, position, offset, size) = oversized
3486                    .append(1, entry, &value)
3487                    .await
3488                    .expect("Failed to append");
3489                locations.push((position, offset, size));
3490            }
3491            oversized = oversized.sync(1).await.expect("Failed to sync");
3492            drop(oversized);
3493
3494            // Simulate crash during rewind: truncate index to 2 entries but leave glob intact
3495            // This simulates: rewind(index) succeeded, crash before rewind(glob)
3496            let (blob, _) = context
3497                .open(&cfg.index_partition, &1u64.to_be_bytes())
3498                .await
3499                .expect("Failed to open blob");
3500            // Physical page size = logical (20) + CRC record (12) = 32 bytes
3501            let physical_page_size = (TestEntry::SIZE + 12) as u64;
3502            blob.resize(2 * physical_page_size)
3503                .await
3504                .expect("Failed to truncate");
3505            blob.sync().await.expect("Failed to sync");
3506            drop(blob);
3507
3508            // Reinitialize - recovery should succeed (glob has orphan data)
3509            let mut oversized: Oversized<_, TestEntry, TestValue> =
3510                Oversized::init(context.child("second"), cfg.clone())
3511                    .await
3512                    .expect("Failed to reinit");
3513
3514            // First 2 entries should be valid
3515            for i in 0..2u8 {
3516                let entry = oversized.get(1, i as u64).await.expect("Failed to get");
3517                assert_eq!(entry.id, i as u64);
3518            }
3519
3520            // Entries 2-4 should be gone (index was truncated)
3521            assert!(oversized.get(1, 2).await.is_err());
3522
3523            // Should be able to append new entries
3524            let pos;
3525            (oversized, pos, _, _) = oversized
3526                .append(1, TestEntry::new(100, 0, 0), &[100u8; 16])
3527                .await
3528                .expect("Failed to append");
3529            assert_eq!(pos, 2);
3530
3531            oversized.destroy().await.expect("Failed to destroy");
3532        });
3533    }
3534
3535    #[test_traced]
3536    fn test_recovery_crash_during_rewind_glob_ahead() {
3537        // Simulates crash where glob was rewound but index wasn't
3538        let executor = deterministic::Runner::default();
3539        executor.start(|context| async move {
3540            let cfg = test_cfg(&context);
3541
3542            // Create and populate
3543            let mut oversized: Oversized<_, TestEntry, TestValue> =
3544                Oversized::init(context.child("first"), cfg.clone())
3545                    .await
3546                    .expect("Failed to init");
3547
3548            let mut locations = Vec::new();
3549            for i in 0..5u8 {
3550                let value: TestValue = [i; 16];
3551                let entry = TestEntry::new(i as u64, 0, 0);
3552                let (position, offset, size);
3553                (oversized, position, offset, size) = oversized
3554                    .append(1, entry, &value)
3555                    .await
3556                    .expect("Failed to append");
3557                locations.push((position, offset, size));
3558            }
3559            oversized = oversized.sync(1).await.expect("Failed to sync");
3560            drop(oversized);
3561
3562            // Simulate crash during rewind: truncate glob to 2 entries but leave index intact
3563            // This simulates: rewind(glob) succeeded, crash before rewind(index)
3564            let (blob, _) = context
3565                .open(&cfg.value_partition, &1u64.to_be_bytes())
3566                .await
3567                .expect("Failed to open blob");
3568            let keep_size = byte_end(locations[1].1, locations[1].2);
3569            blob.resize(keep_size).await.expect("Failed to truncate");
3570            blob.sync().await.expect("Failed to sync");
3571            drop(blob);
3572
3573            // Reinitialize - recovery should detect index entries pointing beyond glob
3574            let mut oversized: Oversized<_, TestEntry, TestValue> =
3575                Oversized::init(context.child("second"), cfg.clone())
3576                    .await
3577                    .expect("Failed to reinit");
3578
3579            // First 2 entries should be valid (index rewound to match glob)
3580            for i in 0..2u8 {
3581                let entry = oversized.get(1, i as u64).await.expect("Failed to get");
3582                assert_eq!(entry.id, i as u64);
3583            }
3584
3585            // Entries 2-4 should be gone (index rewound during recovery)
3586            assert!(oversized.get(1, 2).await.is_err());
3587
3588            // Should be able to append after recovery
3589            let value: TestValue = [99; 16];
3590            let entry = TestEntry::new(100, 0, 0);
3591            let (pos, offset, size);
3592            (oversized, pos, offset, size) = oversized
3593                .append(1, entry, &value)
3594                .await
3595                .expect("Failed to append after recovery");
3596            assert_eq!(pos, 2);
3597
3598            let retrieved = oversized.get(1, 2).await.expect("Failed to get");
3599            assert_eq!(retrieved.id, 100);
3600            let retrieved_value = oversized
3601                .get_value(1, offset, size)
3602                .await
3603                .expect("Failed to get value");
3604            assert_eq!(retrieved_value, value);
3605
3606            oversized.destroy().await.expect("Failed to destroy");
3607        });
3608    }
3609
3610    #[test_traced]
3611    fn test_oversized_get_value_invalid_size() {
3612        let executor = deterministic::Runner::default();
3613        executor.start(|context| async move {
3614            let cfg = test_cfg(&context);
3615            let mut oversized: Oversized<_, TestEntry, TestValue> =
3616                Oversized::init(context, cfg).await.expect("Failed to init");
3617
3618            let value: TestValue = [42; 16];
3619            let entry = TestEntry::new(1, 0, 0);
3620            let (offset, _size);
3621            (oversized, _, offset, _size) = oversized
3622                .append(1, entry, &value)
3623                .await
3624                .expect("Failed to append");
3625            oversized = oversized.sync(1).await.expect("Failed to sync");
3626
3627            // Size 0 - should fail
3628            assert!(oversized.get_value(1, offset, 0).await.is_err());
3629
3630            // Size < value size - should fail with codec error, checksum mismatch, or
3631            // insufficient length (if size < 4 bytes for checksum)
3632            for size in 1..4u32 {
3633                let result = oversized.get_value(1, offset, size).await;
3634                assert!(
3635                    matches!(
3636                        result,
3637                        Err(Error::Codec(_))
3638                            | Err(Error::ChecksumMismatch(_, _))
3639                            | Err(Error::Runtime(_))
3640                    ),
3641                    "expected error, got: {:?}",
3642                    result
3643                );
3644            }
3645
3646            oversized.destroy().await.expect("Failed to destroy");
3647        });
3648    }
3649
3650    #[test_traced]
3651    fn test_oversized_get_value_wrong_size() {
3652        let executor = deterministic::Runner::default();
3653        executor.start(|context| async move {
3654            let cfg = test_cfg(&context);
3655            let mut oversized: Oversized<_, TestEntry, TestValue> =
3656                Oversized::init(context, cfg).await.expect("Failed to init");
3657
3658            let value: TestValue = [42; 16];
3659            let entry = TestEntry::new(1, 0, 0);
3660            let (offset, correct_size);
3661            (oversized, _, offset, correct_size) = oversized
3662                .append(1, entry, &value)
3663                .await
3664                .expect("Failed to append");
3665            oversized = oversized.sync(1).await.expect("Failed to sync");
3666
3667            // Size too small - will fail to decode or checksum mismatch
3668            // (checksum mismatch can occur because we read wrong bytes as the checksum)
3669            let result = oversized.get_value(1, offset, correct_size - 1).await;
3670            assert!(
3671                matches!(
3672                    result,
3673                    Err(Error::Codec(_)) | Err(Error::ChecksumMismatch(_, _))
3674                ),
3675                "expected Codec or ChecksumMismatch error, got: {:?}",
3676                result
3677            );
3678
3679            oversized.destroy().await.expect("Failed to destroy");
3680        });
3681    }
3682
3683    #[test_traced]
3684    fn test_recovery_values_has_orphan_section() {
3685        let executor = deterministic::Runner::default();
3686        executor.start(|context| async move {
3687            let cfg = test_cfg(&context);
3688
3689            // Create and populate with sections 1 and 2
3690            let mut oversized: Oversized<_, TestEntry, TestValue> =
3691                Oversized::init(context.child("first"), cfg.clone())
3692                    .await
3693                    .expect("Failed to init");
3694
3695            for section in 1u64..=2 {
3696                let value: TestValue = [section as u8; 16];
3697                let entry = TestEntry::new(section, 0, 0);
3698                (oversized, _, _, _) = oversized
3699                    .append(section, entry, &value)
3700                    .await
3701                    .expect("Failed to append");
3702                oversized = oversized.sync(section).await.expect("Failed to sync");
3703            }
3704            drop(oversized);
3705
3706            // Manually create an orphan value section (section 3) without corresponding index
3707            let glob_cfg = GlobConfig {
3708                partition: cfg.value_partition.clone(),
3709                compression: cfg.compression,
3710                codec_config: (),
3711                write_buffer: cfg.value_write_buffer,
3712            };
3713            let mut glob: Glob<_, TestValue> = Glob::init(context.child("glob"), glob_cfg)
3714                .await
3715                .expect("Failed to init glob");
3716            let orphan_value: TestValue = [99; 16];
3717            (glob, _, _) = glob
3718                .append(3, &orphan_value)
3719                .await
3720                .expect("Failed to append orphan");
3721            glob = glob.sync(3).await.expect("Failed to sync glob");
3722            drop(glob);
3723
3724            // Reinitialize - should detect and remove the orphan section
3725            let oversized: Oversized<_, TestEntry, TestValue> =
3726                Oversized::init(context.child("second"), cfg.clone())
3727                    .await
3728                    .expect("Failed to reinit");
3729
3730            // Sections 1 and 2 should still be valid
3731            assert!(oversized.get(1, 0).await.is_ok());
3732            assert!(oversized.get(2, 0).await.is_ok());
3733
3734            // Newest section should be 2 (orphan was removed)
3735            assert_eq!(oversized.newest_section(), Some(2));
3736
3737            oversized.destroy().await.expect("Failed to destroy");
3738        });
3739    }
3740
3741    #[test_traced]
3742    fn test_recovery_values_has_multiple_orphan_sections() {
3743        let executor = deterministic::Runner::default();
3744        executor.start(|context| async move {
3745            let cfg = test_cfg(&context);
3746
3747            // Create and populate with only section 1
3748            let mut oversized: Oversized<_, TestEntry, TestValue> =
3749                Oversized::init(context.child("first"), cfg.clone())
3750                    .await
3751                    .expect("Failed to init");
3752
3753            let value: TestValue = [1; 16];
3754            let entry = TestEntry::new(1, 0, 0);
3755            (oversized, _, _, _) = oversized
3756                .append(1, entry, &value)
3757                .await
3758                .expect("Failed to append");
3759            oversized = oversized.sync(1).await.expect("Failed to sync");
3760            drop(oversized);
3761
3762            // Manually create multiple orphan value sections (2, 3, 4)
3763            let glob_cfg = GlobConfig {
3764                partition: cfg.value_partition.clone(),
3765                compression: cfg.compression,
3766                codec_config: (),
3767                write_buffer: cfg.value_write_buffer,
3768            };
3769            let mut glob: Glob<_, TestValue> = Glob::init(context.child("glob"), glob_cfg)
3770                .await
3771                .expect("Failed to init glob");
3772
3773            for section in 2u64..=4 {
3774                let orphan_value: TestValue = [section as u8; 16];
3775                (glob, _, _) = glob
3776                    .append(section, &orphan_value)
3777                    .await
3778                    .expect("Failed to append orphan");
3779                glob = glob.sync(section).await.expect("Failed to sync glob");
3780            }
3781            drop(glob);
3782
3783            // Reinitialize - should detect and remove all orphan sections
3784            let oversized: Oversized<_, TestEntry, TestValue> =
3785                Oversized::init(context.child("second"), cfg.clone())
3786                    .await
3787                    .expect("Failed to reinit");
3788
3789            // Section 1 should still be valid
3790            assert!(oversized.get(1, 0).await.is_ok());
3791
3792            // Newest section should be 1 (orphans removed)
3793            assert_eq!(oversized.newest_section(), Some(1));
3794
3795            oversized.destroy().await.expect("Failed to destroy");
3796        });
3797    }
3798
3799    #[test_traced]
3800    fn test_recovery_index_empty_but_values_exist() {
3801        let executor = deterministic::Runner::default();
3802        executor.start(|context| async move {
3803            let cfg = test_cfg(&context);
3804
3805            // Manually create value sections without any index entries
3806            let glob_cfg = GlobConfig {
3807                partition: cfg.value_partition.clone(),
3808                compression: cfg.compression,
3809                codec_config: (),
3810                write_buffer: cfg.value_write_buffer,
3811            };
3812            let mut glob: Glob<_, TestValue> = Glob::init(context.child("glob"), glob_cfg)
3813                .await
3814                .expect("Failed to init glob");
3815
3816            for section in 1u64..=3 {
3817                let orphan_value: TestValue = [section as u8; 16];
3818                (glob, _, _) = glob
3819                    .append(section, &orphan_value)
3820                    .await
3821                    .expect("Failed to append orphan");
3822                glob = glob.sync(section).await.expect("Failed to sync glob");
3823            }
3824            drop(glob);
3825
3826            // Initialize oversized - should remove all orphan value sections
3827            let oversized: Oversized<_, TestEntry, TestValue> =
3828                Oversized::init(context.child("first"), cfg.clone())
3829                    .await
3830                    .expect("Failed to init");
3831
3832            // No sections should exist
3833            assert_eq!(oversized.newest_section(), None);
3834            assert_eq!(oversized.oldest_section(), None);
3835
3836            oversized.destroy().await.expect("Failed to destroy");
3837        });
3838    }
3839
3840    #[test_traced]
3841    fn test_recovery_orphan_section_append_after() {
3842        let executor = deterministic::Runner::default();
3843        executor.start(|context| async move {
3844            let cfg = test_cfg(&context);
3845
3846            // Create and populate with section 1
3847            let mut oversized: Oversized<_, TestEntry, TestValue> =
3848                Oversized::init(context.child("first"), cfg.clone())
3849                    .await
3850                    .expect("Failed to init");
3851
3852            let value: TestValue = [1; 16];
3853            let entry = TestEntry::new(1, 0, 0);
3854            let (offset1, size1);
3855            (oversized, _, offset1, size1) = oversized
3856                .append(1, entry, &value)
3857                .await
3858                .expect("Failed to append");
3859            oversized = oversized.sync(1).await.expect("Failed to sync");
3860            drop(oversized);
3861
3862            // Manually create orphan value sections (2, 3)
3863            let glob_cfg = GlobConfig {
3864                partition: cfg.value_partition.clone(),
3865                compression: cfg.compression,
3866                codec_config: (),
3867                write_buffer: cfg.value_write_buffer,
3868            };
3869            let mut glob: Glob<_, TestValue> = Glob::init(context.child("glob"), glob_cfg)
3870                .await
3871                .expect("Failed to init glob");
3872
3873            for section in 2u64..=3 {
3874                let orphan_value: TestValue = [section as u8; 16];
3875                (glob, _, _) = glob
3876                    .append(section, &orphan_value)
3877                    .await
3878                    .expect("Failed to append orphan");
3879                glob = glob.sync(section).await.expect("Failed to sync glob");
3880            }
3881            drop(glob);
3882
3883            // Reinitialize - should remove orphan sections
3884            let mut oversized: Oversized<_, TestEntry, TestValue> =
3885                Oversized::init(context.child("second"), cfg.clone())
3886                    .await
3887                    .expect("Failed to reinit");
3888
3889            // Section 1 should still be valid
3890            let entry = oversized.get(1, 0).await.expect("Failed to get");
3891            assert_eq!(entry.id, 1);
3892            let value = oversized
3893                .get_value(1, offset1, size1)
3894                .await
3895                .expect("Failed to get value");
3896            assert_eq!(value, [1; 16]);
3897
3898            // Should be able to append to section 2 after recovery
3899            let new_value: TestValue = [42; 16];
3900            let new_entry = TestEntry::new(42, 0, 0);
3901            let (pos, offset, size);
3902            (oversized, pos, offset, size) = oversized
3903                .append(2, new_entry, &new_value)
3904                .await
3905                .expect("Failed to append after recovery");
3906            assert_eq!(pos, 0);
3907
3908            // Verify the new entry
3909            let retrieved = oversized.get(2, 0).await.expect("Failed to get");
3910            assert_eq!(retrieved.id, 42);
3911            let retrieved_value = oversized
3912                .get_value(2, offset, size)
3913                .await
3914                .expect("Failed to get value");
3915            assert_eq!(retrieved_value, new_value);
3916
3917            // Sync and restart to verify persistence
3918            oversized = oversized.sync(2).await.expect("Failed to sync");
3919            drop(oversized);
3920
3921            let oversized: Oversized<_, TestEntry, TestValue> =
3922                Oversized::init(context.child("third"), cfg)
3923                    .await
3924                    .expect("Failed to reinit after append");
3925
3926            // Both sections should be valid
3927            assert!(oversized.get(1, 0).await.is_ok());
3928            assert!(oversized.get(2, 0).await.is_ok());
3929            assert_eq!(oversized.newest_section(), Some(2));
3930
3931            oversized.destroy().await.expect("Failed to destroy");
3932        });
3933    }
3934
3935    #[test_traced]
3936    fn test_recovery_no_orphan_sections() {
3937        let executor = deterministic::Runner::default();
3938        executor.start(|context| async move {
3939            let cfg = test_cfg(&context);
3940
3941            // Create and populate with sections 1, 2, 3 (no orphans)
3942            let mut oversized: Oversized<_, TestEntry, TestValue> =
3943                Oversized::init(context.child("first"), cfg.clone())
3944                    .await
3945                    .expect("Failed to init");
3946
3947            for section in 1u64..=3 {
3948                let value: TestValue = [section as u8; 16];
3949                let entry = TestEntry::new(section, 0, 0);
3950                (oversized, _, _, _) = oversized
3951                    .append(section, entry, &value)
3952                    .await
3953                    .expect("Failed to append");
3954                oversized = oversized.sync(section).await.expect("Failed to sync");
3955            }
3956            drop(oversized);
3957
3958            // Reinitialize - no orphan cleanup needed
3959            let oversized: Oversized<_, TestEntry, TestValue> =
3960                Oversized::init(context.child("second"), cfg)
3961                    .await
3962                    .expect("Failed to reinit");
3963
3964            // All sections should be valid
3965            for section in 1u64..=3 {
3966                let entry = oversized.get(section, 0).await.expect("Failed to get");
3967                assert_eq!(entry.id, section);
3968            }
3969            assert_eq!(oversized.newest_section(), Some(3));
3970
3971            oversized.destroy().await.expect("Failed to destroy");
3972        });
3973    }
3974
3975    #[test_traced]
3976    fn test_recovery_orphan_with_empty_index_section() {
3977        let executor = deterministic::Runner::default();
3978        executor.start(|context| async move {
3979            let cfg = test_cfg(&context);
3980
3981            // Create and populate section 1 with entries
3982            let mut oversized: Oversized<_, TestEntry, TestValue> =
3983                Oversized::init(context.child("first"), cfg.clone())
3984                    .await
3985                    .expect("Failed to init");
3986
3987            let value: TestValue = [1; 16];
3988            let entry = TestEntry::new(1, 0, 0);
3989            (oversized, _, _, _) = oversized
3990                .append(1, entry, &value)
3991                .await
3992                .expect("Failed to append");
3993            oversized = oversized.sync(1).await.expect("Failed to sync");
3994            drop(oversized);
3995
3996            // Manually create orphan value section 2
3997            let glob_cfg = GlobConfig {
3998                partition: cfg.value_partition.clone(),
3999                compression: cfg.compression,
4000                codec_config: (),
4001                write_buffer: cfg.value_write_buffer,
4002            };
4003            let mut glob: Glob<_, TestValue> = Glob::init(context.child("glob"), glob_cfg)
4004                .await
4005                .expect("Failed to init glob");
4006            let orphan_value: TestValue = [2; 16];
4007            (glob, _, _) = glob
4008                .append(2, &orphan_value)
4009                .await
4010                .expect("Failed to append orphan");
4011            glob = glob.sync(2).await.expect("Failed to sync glob");
4012            drop(glob);
4013
4014            // Now truncate index section 1 to 0 (making it empty but still tracked)
4015            let (blob, _) = context
4016                .open(&cfg.index_partition, &1u64.to_be_bytes())
4017                .await
4018                .expect("Failed to open blob");
4019            blob.resize(0).await.expect("Failed to truncate");
4020            blob.sync().await.expect("Failed to sync");
4021            drop(blob);
4022
4023            // Reinitialize - should handle empty index section and remove orphan value section
4024            let oversized: Oversized<_, TestEntry, TestValue> =
4025                Oversized::init(context.child("second"), cfg)
4026                    .await
4027                    .expect("Failed to reinit");
4028
4029            // Section 1 should exist but have no entries (empty after truncation)
4030            assert!(oversized.get(1, 0).await.is_err());
4031
4032            // Orphan section 2 should be removed
4033            assert_eq!(oversized.newest_section(), Some(1));
4034
4035            oversized.destroy().await.expect("Failed to destroy");
4036        });
4037    }
4038
4039    #[test_traced]
4040    fn test_recovery_orphan_sections_with_gaps() {
4041        // Test non-contiguous sections: index has [1, 3, 5], values has [1, 2, 3, 4, 5, 6]
4042        // Orphan sections 2, 4, 6 should be removed
4043        let executor = deterministic::Runner::default();
4044        executor.start(|context| async move {
4045            let cfg = test_cfg(&context);
4046
4047            // Create index with sections 1, 3, 5 (gaps)
4048            let mut oversized: Oversized<_, TestEntry, TestValue> =
4049                Oversized::init(context.child("first"), cfg.clone())
4050                    .await
4051                    .expect("Failed to init");
4052
4053            for section in [1u64, 3, 5] {
4054                let value: TestValue = [section as u8; 16];
4055                let entry = TestEntry::new(section, 0, 0);
4056                (oversized, _, _, _) = oversized
4057                    .append(section, entry, &value)
4058                    .await
4059                    .expect("Failed to append");
4060                oversized = oversized.sync(section).await.expect("Failed to sync");
4061            }
4062            drop(oversized);
4063
4064            // Manually create orphan value sections 2, 4, 6 (filling gaps and beyond)
4065            let glob_cfg = GlobConfig {
4066                partition: cfg.value_partition.clone(),
4067                compression: cfg.compression,
4068                codec_config: (),
4069                write_buffer: cfg.value_write_buffer,
4070            };
4071            let mut glob: Glob<_, TestValue> = Glob::init(context.child("glob"), glob_cfg)
4072                .await
4073                .expect("Failed to init glob");
4074
4075            for section in [2u64, 4, 6] {
4076                let orphan_value: TestValue = [section as u8; 16];
4077                (glob, _, _) = glob
4078                    .append(section, &orphan_value)
4079                    .await
4080                    .expect("Failed to append orphan");
4081                glob = glob.sync(section).await.expect("Failed to sync glob");
4082            }
4083            drop(glob);
4084
4085            // Reinitialize - should remove orphan sections 2, 4, 6
4086            let oversized: Oversized<_, TestEntry, TestValue> =
4087                Oversized::init(context.child("second"), cfg)
4088                    .await
4089                    .expect("Failed to reinit");
4090
4091            // Sections 1, 3, 5 should still be valid
4092            for section in [1u64, 3, 5] {
4093                let entry = oversized.get(section, 0).await.expect("Failed to get");
4094                assert_eq!(entry.id, section);
4095            }
4096
4097            // Verify only sections 1, 3, 5 exist (orphans removed)
4098            assert_eq!(oversized.oldest_section(), Some(1));
4099            assert_eq!(oversized.newest_section(), Some(5));
4100
4101            oversized.destroy().await.expect("Failed to destroy");
4102        });
4103    }
4104
4105    /// Bytes appended after an index was truncated or removed cannot replace its authenticated
4106    /// prefix.
4107    #[test_traced]
4108    fn test_recovery_discards_index_extension_after_prefix_loss() {
4109        let executor = deterministic::Runner::default();
4110        executor.start(|context| async move {
4111            // Seed two sections with one synced entry each, so both hold durable values.
4112            let cfg = test_cfg(&context);
4113            let mut oversized: Oversized<_, TestEntry, TestValue> =
4114                Oversized::init(context.child("first"), cfg.clone())
4115                    .await
4116                    .expect("Failed to init");
4117            for section in 1..=2 {
4118                (oversized, _, _, _) = oversized
4119                    .append(section, TestEntry::new(section, 0, 0), &[section as u8; 16])
4120                    .await
4121                    .expect("Failed to append");
4122                oversized = oversized.sync(section).await.expect("Failed to sync");
4123            }
4124            drop(oversized);
4125
4126            // Keep both value blobs, but erase one index by truncation and the other by removal.
4127            // Replace each with two complete pages of bytes that have no valid checksum slots.
4128            for section in 1..=2u64 {
4129                let (blob, original_size) = context
4130                    .open(&cfg.index_partition, &section.to_be_bytes())
4131                    .await
4132                    .expect("Failed to open index blob");
4133                if section == 1 {
4134                    blob.resize(0).await.expect("Failed to truncate index");
4135                    blob.sync().await.expect("Failed to sync index truncation");
4136                } else {
4137                    drop(blob);
4138                    context
4139                        .remove(&cfg.index_partition, Some(&section.to_be_bytes()))
4140                        .await
4141                        .expect("Failed to remove index");
4142                }
4143                let (blob, _) = context
4144                    .open(&cfg.index_partition, &section.to_be_bytes())
4145                    .await
4146                    .expect("Failed to recreate index blob");
4147                blob.write_at(
4148                    0,
4149                    vec![0; usize::try_from(original_size * 2).unwrap()],
4150                    WriteOptions::SYNC,
4151                )
4152                .await
4153                .expect("Failed to extend index");
4154            }
4155
4156            // Recovery must not adopt the checksum-less extensions as index data: both
4157            // sections come back empty (the original entries are gone with their prefixes)
4158            // and the truncation is durable.
4159            let mut oversized: Oversized<_, TestEntry, TestValue> =
4160                Oversized::init(context.child("second"), cfg.clone())
4161                    .await
4162                    .expect("Failed to recover extended index");
4163            for section in 1..=2u64 {
4164                assert!(matches!(
4165                    oversized.get(section, 0).await,
4166                    Err(Error::ItemOutOfRange(0))
4167                ));
4168                let (_, recovered_size) = context
4169                    .open(&cfg.index_partition, &section.to_be_bytes())
4170                    .await
4171                    .expect("Failed to reopen index blob");
4172                assert_eq!(recovered_size, 0);
4173
4174                // The recovered sections accept new entries from position zero.
4175                let position;
4176                (oversized, position, _, _) = oversized
4177                    .append(section, TestEntry::new(section, 0, 0), &[section as u8; 16])
4178                    .await
4179                    .expect("Failed to append after recovery");
4180                assert_eq!(position, 0);
4181            }
4182            oversized = oversized.sync_all().await.expect("Failed to sync sentinel");
4183            drop(oversized);
4184
4185            // The sentinel entries written after recovery survive a clean reopen.
4186            let oversized: Oversized<_, TestEntry, TestValue> =
4187                Oversized::init(context.child("third"), cfg)
4188                    .await
4189                    .expect("Failed to reopen sentinel");
4190            for section in 1..=2u64 {
4191                let entry = oversized
4192                    .get(section, 0)
4193                    .await
4194                    .expect("Sentinel index missing");
4195                let (offset, size) = entry.value_location();
4196                assert_eq!(entry.id, section);
4197                assert_eq!(
4198                    oversized
4199                        .get_value(section, offset, size)
4200                        .await
4201                        .expect("Sentinel value missing"),
4202                    [section as u8; 16]
4203                );
4204            }
4205            oversized.destroy().await.expect("Failed to destroy");
4206        });
4207    }
4208
4209    #[test_traced]
4210    fn test_recovery_glob_trailing_garbage_truncated() {
4211        // Tests the bug fix: when value is written to glob but index entry isn't
4212        // (crash after value write, before index write), recovery should truncate
4213        // the glob trailing garbage so subsequent appends start at correct offset.
4214        let executor = deterministic::Runner::default();
4215        executor.start(|context| async move {
4216            let cfg = test_cfg(&context);
4217
4218            // Create and populate
4219            let mut oversized: Oversized<_, TestEntry, TestValue> =
4220                Oversized::init(context.child("first"), cfg.clone())
4221                    .await
4222                    .expect("Failed to init");
4223
4224            // Append 2 entries
4225            let mut locations = Vec::new();
4226            for i in 0..2u8 {
4227                let value: TestValue = [i; 16];
4228                let entry = TestEntry::new(i as u64, 0, 0);
4229                let (position, offset, size);
4230                (oversized, position, offset, size) = oversized
4231                    .append(1, entry, &value)
4232                    .await
4233                    .expect("Failed to append");
4234                locations.push((position, offset, size));
4235            }
4236            oversized = oversized.sync(1).await.expect("Failed to sync");
4237
4238            // Record where next entry SHOULD start (end of entry 1)
4239            let expected_next_offset = byte_end(locations[1].1, locations[1].2);
4240            drop(oversized);
4241
4242            // Simulate crash: write garbage to glob (simulating partial value write)
4243            let (blob, size) = context
4244                .open(&cfg.value_partition, &1u64.to_be_bytes())
4245                .await
4246                .expect("Failed to open blob");
4247            assert_eq!(size, expected_next_offset);
4248
4249            // Write 100 bytes of garbage (simulating partial/failed value write)
4250            let garbage = vec![0xDE; 100];
4251            blob.write_at(size, garbage, WriteOptions::SYNC)
4252                .await
4253                .expect("Failed to write garbage");
4254            drop(blob);
4255
4256            // Verify glob now has trailing garbage
4257            let (blob, new_size) = context
4258                .open(&cfg.value_partition, &1u64.to_be_bytes())
4259                .await
4260                .expect("Failed to open blob");
4261            assert_eq!(new_size, expected_next_offset + 100);
4262            drop(blob);
4263
4264            // Reinitialize - should truncate the trailing garbage
4265            let mut oversized: Oversized<_, TestEntry, TestValue> =
4266                Oversized::init(context.child("second"), cfg.clone())
4267                    .await
4268                    .expect("Failed to reinit");
4269
4270            // First 2 entries should still be valid
4271            for i in 0..2u8 {
4272                let entry = oversized.get(1, i as u64).await.expect("Failed to get");
4273                assert_eq!(entry.id, i as u64);
4274            }
4275
4276            // Append new entry - should start at expected_next_offset, NOT at garbage end
4277            let new_value: TestValue = [99; 16];
4278            let new_entry = TestEntry::new(99, 0, 0);
4279            let (pos, offset, _size);
4280            (oversized, pos, offset, _size) = oversized
4281                .append(1, new_entry, &new_value)
4282                .await
4283                .expect("Failed to append after recovery");
4284
4285            // Verify position is 2 (after the 2 existing entries)
4286            assert_eq!(pos, 2);
4287
4288            // Verify offset is at expected_next_offset (garbage was truncated)
4289            assert_eq!(offset, expected_next_offset);
4290
4291            // Verify we can read the new entry
4292            let retrieved = oversized.get(1, 2).await.expect("Failed to get new entry");
4293            assert_eq!(retrieved.id, 99);
4294
4295            oversized.destroy().await.expect("Failed to destroy");
4296        });
4297    }
4298
4299    #[test_traced]
4300    fn test_recovery_entry_with_overflow_offset() {
4301        // Tests that an entry with offset near u64::MAX that would overflow
4302        // when added to size is detected as invalid during recovery.
4303        let executor = deterministic::Runner::default();
4304        executor.start(|context| async move {
4305            // Use page size = entry size so one entry per page
4306            let cfg = entry_cfg(&context);
4307
4308            // Create and populate with valid entry
4309            let mut oversized: Oversized<_, TestEntry, TestValue> =
4310                Oversized::init(context.child("first"), cfg.clone())
4311                    .await
4312                    .expect("Failed to init");
4313
4314            let value: TestValue = [1; 16];
4315            let entry = TestEntry::new(1, 0, 0);
4316            (oversized, _, _, _) = oversized
4317                .append(1, entry, &value)
4318                .await
4319                .expect("Failed to append");
4320            oversized = oversized.sync(1).await.expect("Failed to sync");
4321            drop(oversized);
4322
4323            // Build a corrupted entry with offset near u64::MAX that would overflow.
4324            // We need to write a valid page (with correct page-level CRC) containing
4325            // the semantically-invalid entry data.
4326            let (blob, _) = context
4327                .open(&cfg.index_partition, &1u64.to_be_bytes())
4328                .await
4329                .expect("Failed to open blob");
4330
4331            // Build entry data: id (8) + value_offset (8) + value_size (4) = 20 bytes
4332            let mut entry_data = Vec::new();
4333            1u64.write(&mut entry_data); // id
4334            (u64::MAX - 10).write(&mut entry_data); // value_offset (near max)
4335            100u32.write(&mut entry_data); // value_size (offset + size overflows)
4336            assert_eq!(entry_data.len(), TestEntry::SIZE);
4337
4338            // Build page-level CRC record (12 bytes):
4339            // len1 (2) + crc1 (4) + len2 (2) + crc2 (4)
4340            let crc = Crc32::checksum(&entry_data);
4341            let len1 = TestEntry::SIZE as u16;
4342            let mut crc_record = Vec::new();
4343            crc_record.extend_from_slice(&len1.to_be_bytes()); // len1
4344            crc_record.extend_from_slice(&crc.to_be_bytes()); // crc1
4345            crc_record.extend_from_slice(&0u16.to_be_bytes()); // len2 (unused)
4346            crc_record.extend_from_slice(&0u32.to_be_bytes()); // crc2 (unused)
4347            assert_eq!(crc_record.len(), 12);
4348
4349            // Write the complete physical page: entry_data + crc_record
4350            let mut page = entry_data;
4351            page.extend_from_slice(&crc_record);
4352            blob.write_at(0, page, WriteOptions::SYNC)
4353                .await
4354                .expect("Failed to write corrupted page");
4355            drop(blob);
4356
4357            // Reinitialize - recovery should detect the invalid entry
4358            // (offset + size would overflow, and even with saturating_add it exceeds glob_size)
4359            let mut oversized: Oversized<_, TestEntry, TestValue> =
4360                Oversized::init(context.child("second"), cfg.clone())
4361                    .await
4362                    .expect("Failed to reinit");
4363
4364            // The corrupted entry should have been rewound (invalid)
4365            assert!(oversized.get(1, 0).await.is_err());
4366
4367            // Should be able to append after recovery
4368            let new_value: TestValue = [99; 16];
4369            let new_entry = TestEntry::new(99, 0, 0);
4370            let (pos, new_offset);
4371            (oversized, pos, new_offset, _) = oversized
4372                .append(1, new_entry, &new_value)
4373                .await
4374                .expect("Failed to append after recovery");
4375
4376            // Position should be 0 (corrupted entry was removed)
4377            assert_eq!(pos, 0);
4378            // Offset should be 0 (glob was truncated to 0)
4379            assert_eq!(new_offset, 0);
4380
4381            oversized.destroy().await.expect("Failed to destroy");
4382        });
4383    }
4384
4385    #[test_traced]
4386    fn test_empty_section_persistence() {
4387        // Tests that sections that become empty (all entries removed/rewound)
4388        // are handled correctly across restart cycles.
4389        let executor = deterministic::Runner::default();
4390        executor.start(|context| async move {
4391            let cfg = test_cfg(&context);
4392
4393            // Create and populate section 1 with entries
4394            let mut oversized: Oversized<_, TestEntry, TestValue> =
4395                Oversized::init(context.child("first"), cfg.clone())
4396                    .await
4397                    .expect("Failed to init");
4398
4399            for i in 0..3u8 {
4400                let value: TestValue = [i; 16];
4401                let entry = TestEntry::new(i as u64, 0, 0);
4402                (oversized, _, _, _) = oversized
4403                    .append(1, entry, &value)
4404                    .await
4405                    .expect("Failed to append");
4406            }
4407            oversized = oversized.sync(1).await.expect("Failed to sync");
4408
4409            // Also create section 2 to ensure it survives
4410            let value2: TestValue = [10; 16];
4411            let entry2 = TestEntry::new(10, 0, 0);
4412            (oversized, _, _, _) = oversized
4413                .append(2, entry2, &value2)
4414                .await
4415                .expect("Failed to append to section 2");
4416            oversized = oversized.sync(2).await.expect("Failed to sync section 2");
4417            drop(oversized);
4418
4419            // Truncate section 1's index to 0 (making it empty)
4420            let (blob, _) = context
4421                .open(&cfg.index_partition, &1u64.to_be_bytes())
4422                .await
4423                .expect("Failed to open blob");
4424            blob.resize(0).await.expect("Failed to truncate");
4425            blob.sync().await.expect("Failed to sync");
4426            drop(blob);
4427
4428            // First restart - recovery should handle empty section 1
4429            let mut oversized: Oversized<_, TestEntry, TestValue> =
4430                Oversized::init(context.child("second"), cfg.clone())
4431                    .await
4432                    .expect("Failed to reinit");
4433
4434            // Section 1 should exist but have no entries
4435            assert!(oversized.get(1, 0).await.is_err());
4436
4437            // Section 2 should still be valid
4438            let entry = oversized.get(2, 0).await.expect("Failed to get section 2");
4439            assert_eq!(entry.id, 10);
4440
4441            // Section 1 should still be tracked (blob exists but is empty)
4442            assert_eq!(oversized.oldest_section(), Some(1));
4443
4444            // Values are reachable only through index entries, so recovery removes the orphaned
4445            // bytes before the section can be reused.
4446            let new_value: TestValue = [99; 16];
4447            let new_entry = TestEntry::new(99, 0, 0);
4448            let (pos, offset, size);
4449            (oversized, pos, offset, size) = oversized
4450                .append(1, new_entry, &new_value)
4451                .await
4452                .expect("Failed to append to empty section");
4453            assert_eq!(pos, 0);
4454            assert_eq!(offset, 0);
4455            oversized = oversized.sync(1).await.expect("Failed to sync");
4456
4457            // Verify the new entry is readable after reusing the section.
4458            let entry = oversized.get(1, 0).await.expect("Failed to get");
4459            assert_eq!(entry.id, 99);
4460            let value = oversized
4461                .get_value(1, offset, size)
4462                .await
4463                .expect("Failed to get value");
4464            assert_eq!(value, new_value);
4465
4466            drop(oversized);
4467
4468            // Second restart - verify persistence
4469            let oversized: Oversized<_, TestEntry, TestValue> =
4470                Oversized::init(context.child("third"), cfg.clone())
4471                    .await
4472                    .expect("Failed to reinit again");
4473
4474            // Section 1's new entry should be valid
4475            let entry = oversized.get(1, 0).await.expect("Failed to get");
4476            assert_eq!(entry.id, 99);
4477
4478            // Section 2 should still be valid
4479            let entry = oversized.get(2, 0).await.expect("Failed to get section 2");
4480            assert_eq!(entry.id, 10);
4481
4482            oversized.destroy().await.expect("Failed to destroy");
4483        });
4484    }
4485
4486    #[test_traced]
4487    fn test_get_value_size_equals_crc_size() {
4488        // Tests the boundary condition where size = 4 (just CRC, no data).
4489        // This should fail because there's no actual data to decode.
4490        let executor = deterministic::Runner::default();
4491        executor.start(|context| async move {
4492            let cfg = test_cfg(&context);
4493            let mut oversized: Oversized<_, TestEntry, TestValue> =
4494                Oversized::init(context, cfg).await.expect("Failed to init");
4495
4496            let value: TestValue = [42; 16];
4497            let entry = TestEntry::new(1, 0, 0);
4498            let offset;
4499            (oversized, _, offset, _) = oversized
4500                .append(1, entry, &value)
4501                .await
4502                .expect("Failed to append");
4503            oversized = oversized.sync(1).await.expect("Failed to sync");
4504
4505            // Size = 4 (exactly CRC_SIZE) means 0 bytes of actual data
4506            // This should fail with ChecksumMismatch or decode error
4507            let result = oversized.get_value(1, offset, 4).await;
4508            assert!(result.is_err());
4509
4510            oversized.destroy().await.expect("Failed to destroy");
4511        });
4512    }
4513
4514    #[test_traced]
4515    fn test_get_value_size_just_over_crc() {
4516        // Tests size = 5 (CRC + 1 byte of data).
4517        // This should fail because the data is too short to decode.
4518        let executor = deterministic::Runner::default();
4519        executor.start(|context| async move {
4520            let cfg = test_cfg(&context);
4521            let mut oversized: Oversized<_, TestEntry, TestValue> =
4522                Oversized::init(context, cfg).await.expect("Failed to init");
4523
4524            let value: TestValue = [42; 16];
4525            let entry = TestEntry::new(1, 0, 0);
4526            let offset;
4527            (oversized, _, offset, _) = oversized
4528                .append(1, entry, &value)
4529                .await
4530                .expect("Failed to append");
4531            oversized = oversized.sync(1).await.expect("Failed to sync");
4532
4533            // Size = 5 means 1 byte of actual data (after stripping CRC)
4534            // This should fail with checksum mismatch since we're reading wrong bytes
4535            let result = oversized.get_value(1, offset, 5).await;
4536            assert!(result.is_err());
4537
4538            oversized.destroy().await.expect("Failed to destroy");
4539        });
4540    }
4541
4542    #[test_traced]
4543    fn test_recovery_maximum_section_numbers() {
4544        // Test recovery with very large section numbers near u64::MAX to check
4545        // for overflow edge cases in section arithmetic.
4546        let executor = deterministic::Runner::default();
4547        executor.start(|context| async move {
4548            let cfg = test_cfg(&context);
4549
4550            // Use section numbers near u64::MAX
4551            let large_sections = [u64::MAX - 3, u64::MAX - 2, u64::MAX - 1];
4552
4553            // Create and populate with large section numbers
4554            let mut oversized: Oversized<_, TestEntry, TestValue> =
4555                Oversized::init(context.child("first"), cfg.clone())
4556                    .await
4557                    .expect("Failed to init");
4558
4559            let mut locations = Vec::new();
4560            for &section in &large_sections {
4561                let value: TestValue = [(section & 0xFF) as u8; 16];
4562                let entry = TestEntry::new(section, 0, 0);
4563                let (position, offset, size);
4564                (oversized, position, offset, size) = oversized
4565                    .append(section, entry, &value)
4566                    .await
4567                    .expect("Failed to append");
4568                locations.push((section, (position, offset, size)));
4569                oversized = oversized.sync(section).await.expect("Failed to sync");
4570            }
4571            drop(oversized);
4572
4573            // Simulate crash: truncate glob for middle section
4574            let middle_section = large_sections[1];
4575            let (blob, size) = context
4576                .open(&cfg.value_partition, &middle_section.to_be_bytes())
4577                .await
4578                .expect("Failed to open blob");
4579            blob.resize(size / 2).await.expect("Failed to truncate");
4580            blob.sync().await.expect("Failed to sync");
4581            drop(blob);
4582
4583            // Reinitialize - should recover without overflow panics
4584            let mut oversized: Oversized<_, TestEntry, TestValue> =
4585                Oversized::init(context.child("second"), cfg.clone())
4586                    .await
4587                    .expect("Failed to reinit");
4588
4589            // First and last sections should still be valid
4590            let entry = oversized
4591                .get(large_sections[0], 0)
4592                .await
4593                .expect("Failed to get first section");
4594            assert_eq!(entry.id, large_sections[0]);
4595
4596            let entry = oversized
4597                .get(large_sections[2], 0)
4598                .await
4599                .expect("Failed to get last section");
4600            assert_eq!(entry.id, large_sections[2]);
4601
4602            // Middle section should have been rewound (no entries)
4603            assert!(oversized.get(middle_section, 0).await.is_err());
4604
4605            // Verify we can still append to these large sections
4606            let new_value: TestValue = [0xAB; 16];
4607            let new_entry = TestEntry::new(999, 0, 0);
4608            (oversized, _, _, _) = oversized
4609                .append(middle_section, new_entry, &new_value)
4610                .await
4611                .expect("Failed to append after recovery");
4612
4613            oversized.destroy().await.expect("Failed to destroy");
4614        });
4615    }
4616
4617    #[test_traced]
4618    fn test_recovery_crash_during_recovery_rewind() {
4619        // Tests a nested crash scenario: initial crash leaves inconsistent state,
4620        // then a second crash occurs during recovery's rewind operation.
4621        // This simulates the worst-case where recovery itself is interrupted.
4622        let executor = deterministic::Runner::default();
4623        executor.start(|context| async move {
4624            let cfg = test_cfg(&context);
4625
4626            // Phase 1: Create valid data with 5 entries
4627            let mut oversized: Oversized<_, TestEntry, TestValue> =
4628                Oversized::init(context.child("first"), cfg.clone())
4629                    .await
4630                    .expect("Failed to init");
4631
4632            let mut locations = Vec::new();
4633            for i in 0..5u8 {
4634                let value: TestValue = [i; 16];
4635                let entry = TestEntry::new(i as u64, 0, 0);
4636                let (position, offset, size);
4637                (oversized, position, offset, size) = oversized
4638                    .append(1, entry, &value)
4639                    .await
4640                    .expect("Failed to append");
4641                locations.push((position, offset, size));
4642            }
4643            oversized = oversized.sync(1).await.expect("Failed to sync");
4644            drop(oversized);
4645
4646            // Phase 2: Simulate first crash - truncate glob to lose last 2 entries
4647            let (blob, _) = context
4648                .open(&cfg.value_partition, &1u64.to_be_bytes())
4649                .await
4650                .expect("Failed to open blob");
4651            let keep_size = byte_end(locations[2].1, locations[2].2);
4652            blob.resize(keep_size).await.expect("Failed to truncate");
4653            blob.sync().await.expect("Failed to sync");
4654            drop(blob);
4655
4656            // Phase 3: Simulate crash during recovery's rewind
4657            // Recovery would try to rewind index from 5 entries to 3 entries.
4658            // Simulate partial rewind by manually truncating index to 4 entries
4659            // (as if crash occurred mid-rewind).
4660            let chunk_size = FixedJournal::<deterministic::Context, TestEntry>::CHUNK_SIZE as u64;
4661            let (index_blob, _) = context
4662                .open(&cfg.index_partition, &1u64.to_be_bytes())
4663                .await
4664                .expect("Failed to open index blob");
4665            let partial_rewind_size = 4 * chunk_size; // 4 entries instead of 3
4666            index_blob
4667                .resize(partial_rewind_size)
4668                .await
4669                .expect("Failed to resize");
4670            index_blob.sync().await.expect("Failed to sync");
4671            drop(index_blob);
4672
4673            // Phase 4: Second recovery attempt should handle the inconsistent state
4674            // Index has 4 entries, but glob only supports 3.
4675            let mut oversized: Oversized<_, TestEntry, TestValue> =
4676                Oversized::init(context.child("second"), cfg.clone())
4677                    .await
4678                    .expect("Failed to reinit after nested crash");
4679
4680            // Only first 3 entries should be valid (recovery should rewind again)
4681            for i in 0..3u8 {
4682                let entry = oversized.get(1, i as u64).await.expect("Failed to get");
4683                assert_eq!(entry.id, i as u64);
4684
4685                let (_, offset, size) = locations[i as usize];
4686                let value = oversized
4687                    .get_value(1, offset, size)
4688                    .await
4689                    .expect("Failed to get value");
4690                assert_eq!(value, [i; 16]);
4691            }
4692
4693            // Entry 3 should not exist (index was rewound to match glob)
4694            assert!(oversized.get(1, 3).await.is_err());
4695
4696            // Verify append works after nested crash recovery
4697            let new_value: TestValue = [0xFF; 16];
4698            let new_entry = TestEntry::new(100, 0, 0);
4699            let (pos, offset, _size);
4700            (oversized, pos, offset, _size) = oversized
4701                .append(1, new_entry, &new_value)
4702                .await
4703                .expect("Failed to append");
4704            assert_eq!(pos, 3); // Should be position 3 (after the 3 valid entries)
4705
4706            // Verify the offset starts where entry 2 ended (no gaps)
4707            assert_eq!(offset, byte_end(locations[2].1, locations[2].2));
4708
4709            oversized.destroy().await.expect("Failed to destroy");
4710        });
4711    }
4712
4713    #[test_traced]
4714    fn test_recovery_crash_during_orphan_cleanup() {
4715        // Tests crash during orphan section cleanup: recovery starts removing
4716        // orphan value sections, but crashes mid-cleanup.
4717        let executor = deterministic::Runner::default();
4718        executor.start(|context| async move {
4719            let cfg = test_cfg(&context);
4720
4721            // Phase 1: Create valid data in section 1
4722            let mut oversized: Oversized<_, TestEntry, TestValue> =
4723                Oversized::init(context.child("first"), cfg.clone())
4724                    .await
4725                    .expect("Failed to init");
4726
4727            let value: TestValue = [1; 16];
4728            let entry = TestEntry::new(1, 0, 0);
4729            let (offset1, size1);
4730            (oversized, _, offset1, size1) = oversized
4731                .append(1, entry, &value)
4732                .await
4733                .expect("Failed to append");
4734            oversized = oversized.sync(1).await.expect("Failed to sync");
4735            drop(oversized);
4736
4737            // Phase 2: Create orphan value sections 2, 3, 4 (no index entries)
4738            let glob_cfg = GlobConfig {
4739                partition: cfg.value_partition.clone(),
4740                compression: cfg.compression,
4741                codec_config: (),
4742                write_buffer: cfg.value_write_buffer,
4743            };
4744            let mut glob: Glob<_, TestValue> = Glob::init(context.child("glob"), glob_cfg)
4745                .await
4746                .expect("Failed to init glob");
4747
4748            for section in 2u64..=4 {
4749                let orphan_value: TestValue = [section as u8; 16];
4750                (glob, _, _) = glob
4751                    .append(section, &orphan_value)
4752                    .await
4753                    .expect("Failed to append orphan");
4754                glob = glob.sync(section).await.expect("Failed to sync glob");
4755            }
4756            drop(glob);
4757
4758            // Phase 3: Simulate partial orphan cleanup (section 2 removed, 3 and 4 remain)
4759            // This simulates a crash during cleanup_orphan_value_sections()
4760            context
4761                .remove(&cfg.value_partition, Some(&2u64.to_be_bytes()))
4762                .await
4763                .expect("Failed to remove section 2");
4764
4765            // Phase 4: Recovery should complete the cleanup
4766            let mut oversized: Oversized<_, TestEntry, TestValue> =
4767                Oversized::init(context.child("second"), cfg.clone())
4768                    .await
4769                    .expect("Failed to reinit");
4770
4771            // Section 1 should still be valid
4772            let entry = oversized.get(1, 0).await.expect("Failed to get");
4773            assert_eq!(entry.id, 1);
4774            let value = oversized
4775                .get_value(1, offset1, size1)
4776                .await
4777                .expect("Failed to get value");
4778            assert_eq!(value, [1; 16]);
4779
4780            // No orphan sections should remain
4781            assert_eq!(oversized.oldest_section(), Some(1));
4782            assert_eq!(oversized.newest_section(), Some(1));
4783
4784            // Should be able to append to section 2 (now clean)
4785            let new_value: TestValue = [42; 16];
4786            let new_entry = TestEntry::new(42, 0, 0);
4787            let pos;
4788            (oversized, pos, _, _) = oversized
4789                .append(2, new_entry, &new_value)
4790                .await
4791                .expect("Failed to append to section 2");
4792            assert_eq!(pos, 0); // First entry in new section
4793
4794            oversized.destroy().await.expect("Failed to destroy");
4795        });
4796    }
4797
4798    #[test_traced]
4799    fn test_rewind_to_zero_index_size() {
4800        let executor = deterministic::Runner::default();
4801        executor.start(|context| async move {
4802            let cfg = test_cfg(&context);
4803            let mut oversized: Oversized<_, TestEntry, TestValue> =
4804                Oversized::init(context, cfg).await.expect("Failed to init");
4805
4806            let value: TestValue = [1; 16];
4807            let entry = TestEntry::new(1, 0, 0);
4808            (oversized, _, _, _) = oversized
4809                .append(0, entry, &value)
4810                .await
4811                .expect("Failed to append");
4812            oversized = oversized.sync(0).await.expect("Failed to sync");
4813
4814            oversized = oversized
4815                .rewind(0, 0)
4816                .await
4817                .expect("rewind to zero index_size must not fail");
4818
4819            assert_eq!(oversized.last(0).await.unwrap(), None);
4820            assert_eq!(oversized.size(0).unwrap(), 0);
4821            assert_eq!(oversized.value_size(0).await.unwrap(), 0);
4822
4823            oversized.destroy().await.expect("Failed to destroy");
4824        });
4825    }
4826
4827    #[test_traced]
4828    fn test_rewind_to_zero_on_missing_section() {
4829        let executor = deterministic::Runner::default();
4830        executor.start(|context| async move {
4831            let cfg = test_cfg(&context);
4832            let mut oversized: Oversized<_, TestEntry, TestValue> =
4833                Oversized::init(context, cfg).await.expect("Failed to init");
4834
4835            oversized = oversized
4836                .rewind(0, 0)
4837                .await
4838                .expect("rewind on missing section must not fail");
4839
4840            assert!(matches!(
4841                oversized.last(0).await,
4842                Err(Error::SectionOutOfRange(0))
4843            ));
4844            assert_eq!(oversized.value_size(0).await.unwrap(), 0);
4845
4846            oversized.destroy().await.expect("Failed to destroy");
4847        });
4848    }
4849
4850    #[test_traced]
4851    fn test_rewind_nonzero_on_missing_section_errors() {
4852        let executor = deterministic::Runner::default();
4853        executor.start(|context| async move {
4854            let cfg = test_cfg(&context);
4855            let oversized: Oversized<_, TestEntry, TestValue> =
4856                Oversized::init(context, cfg).await.expect("Failed to init");
4857
4858            let result = oversized.rewind(0, 1).await;
4859            assert!(
4860                matches!(result, Err(Error::SectionOutOfRange(0))),
4861                "nonzero index_size on missing section must fail, got: {result:?}"
4862            );
4863        });
4864    }
4865
4866    #[test_traced]
4867    fn test_rewind_section_nonzero_on_missing_section_errors() {
4868        let executor = deterministic::Runner::default();
4869        executor.start(|context| async move {
4870            let cfg = test_cfg(&context);
4871            let oversized: Oversized<_, TestEntry, TestValue> =
4872                Oversized::init(context, cfg).await.expect("Failed to init");
4873
4874            let result = oversized.rewind_section(0, 1).await;
4875            assert!(
4876                matches!(result, Err(Error::SectionOutOfRange(0))),
4877                "nonzero index_size on missing section must fail, got: {result:?}"
4878            );
4879        });
4880    }
4881
4882    #[test_traced]
4883    fn test_last_pruned_section_returns_error() {
4884        let executor = deterministic::Runner::default();
4885        executor.start(|context| async move {
4886            let cfg = test_cfg(&context);
4887            let mut oversized: Oversized<_, TestEntry, TestValue> =
4888                Oversized::init(context, cfg).await.expect("Failed to init");
4889
4890            let value: TestValue = [1; 16];
4891            (oversized, _, _, _) = oversized
4892                .append(0, TestEntry::new(1, 0, 0), &value)
4893                .await
4894                .expect("Failed to append");
4895            (oversized, _, _, _) = oversized
4896                .append(1, TestEntry::new(2, 0, 0), &value)
4897                .await
4898                .expect("Failed to append");
4899            oversized = oversized.sync_all().await.expect("Failed to sync");
4900
4901            (oversized, _) = oversized.prune(1).await.expect("Failed to prune");
4902
4903            assert!(matches!(
4904                oversized.last(0).await,
4905                Err(Error::AlreadyPrunedToSection(1))
4906            ));
4907            assert!(oversized.last(1).await.unwrap().is_some());
4908
4909            oversized.destroy().await.expect("Failed to destroy");
4910        });
4911    }
4912}