riegeli 0.2.1

//! `RecordReader` — reads a Riegeli file from any `Read + Seek` source.
//!
//! ## Reading algorithm
//!
//! The reader maintains a "next chunk file position" cursor. On each `read_record()` call:
//! 1. If the current chunk decoder has records remaining, yield the next one.
//! 2. Otherwise, advance to the next chunk: skip any block header at block boundaries,
//!    read a 40-byte `ChunkHeader`, read `data_size` bytes, validate, decode.
//! 3. If a hash validation fails:
//!    - Without recovery: return `Err`.
//!    - With recovery: compute the invalid region, call the callback with
//!      it, and — if the callback returns `true` — resume reading at the
//!      region's end (exactly one bad chunk when its header is trustworthy,
//!      the next block boundary otherwise).

use std::cmp::Ordering;
use std::io::{Read, Seek, SeekFrom};

use crate::block_arithmetic::{is_block_boundary, round_down_to_block_boundary};
use crate::block_header::BlockHeader;
use crate::chunk_header::{ChunkHeader, ChunkType};
use crate::constants::{BLOCK_HEADER_SIZE, BLOCK_SIZE, CHUNK_HEADER_SIZE};
use crate::error::RiegeliError;
use crate::field_projection::FieldProjection;
use crate::record_position::RecordPosition;
use crate::simple_chunk::{Chunk, SimpleChunkDecoder};
use crate::transpose::decoder::TransposeChunkDecoder;

/// Type alias for the optional recovery callback.
///
/// C++ equivalent: `std::function<bool(const SkippedRegion&, RecordReaderBase&)>`.
/// The Rust callback does not receive the reader (it is owned by the reader,
/// so a mutable reference would alias); after a cancelled operation returns,
/// the caller repositions the reader itself, and
/// [`RecordReader::last_skipped_region`] exposes the region that was
/// reported.
type RecoveryCallback = Option<Box<dyn FnMut(&crate::SkippedRegion) -> bool>>;

/// Options for configuring a [`RecordReader`].
pub struct ReaderOptions {
    recovery: RecoveryCallback,
    /// Optional field projection for column pruning in transpose chunks.
    field_projection: Option<FieldProjection>,
}

impl ReaderOptions {
    /// Create `ReaderOptions` with default settings (no recovery, no projection).
    pub fn new() -> Self {
        Self {
            recovery: None,
            field_projection: None,
        }
    }

    /// Set a recovery callback invoked when a corrupted region is
    /// encountered.
    ///
    /// The callback receives the [`SkippedRegion`](crate::SkippedRegion)
    /// about to be skipped. Returning `true` skips the region and continues
    /// the operation; returning `false` cancels it, and the operation
    /// returns the original error.
    ///
    /// The region's `end()` is exactly where the reader resumes: one bad
    /// chunk when its header is trustworthy (hash-valid with
    /// stream-bounded claims), the next block boundary otherwise.
    ///
    /// Divergences from the C++ `set_recovery`
    /// (`std::function<bool(const SkippedRegion&, RecordReaderBase&)>`):
    ///
    /// - No reader parameter: the reader owns the callback, so passing it a
    ///   mutable reference would alias. Use
    ///   [`RecordReader::last_skipped_region`] after a cancelled call.
    /// - Cancelling follows the C++ shape: the region is already skipped
    ///   when the callback runs, so returning `false` reports the error
    ///   once and the NEXT operation continues past the rejected region —
    ///   the callback is never re-invoked for a region it rejected, and a
    ///   retry loop makes progress. A caller that wants to stop AT the
    ///   damage can take [`RecordReader::last_skipped_region`] and
    ///   `seek` to its `begin()`.
    pub fn recovery<F: FnMut(&crate::SkippedRegion) -> bool + 'static>(mut self, f: F) -> Self {
        self.recovery = Some(Box::new(f));
        self
    }

    /// Set a `FieldProjection` to enable column pruning for transpose chunks.
    ///
    /// When set to a non-`all()` projection, the `TransposeChunkDecoder` will
    /// skip data buffers for fields not in the projection and filter decoded
    /// records to contain only the projected fields.
    ///
    /// Non-proto records and simple (non-transpose) chunks are returned unchanged.
    pub fn field_projection(mut self, proj: FieldProjection) -> Self {
        self.field_projection = Some(proj);
        self
    }
}

impl Default for ReaderOptions {
    fn default() -> Self {
        Self::new()
    }
}

/// Active chunk decoder — either simple or transposed.
enum ActiveDecoder {
    Simple(SimpleChunkDecoder),
    Transposed(TransposeChunkDecoder),
}

impl ActiveDecoder {
    // NOTE: the recovery design relies on decoders being structurally
    // infallible after construction — both variants slice pre-validated
    // ranges out of pre-decoded buffers, so the Result below has no
    // reachable Err today. That is what lets every recoverable failure be
    // position-stable at a chunk boundary (the C++ kRecoverChunkDecoder
    // sub-chunk case collapses to the chunk case here). If a streaming or
    // lazy decoder is ever introduced, mid-chunk failures become possible
    // and need their own position-stability and SkippedRegion story.
    fn read_record(&mut self) -> Result<Option<Vec<u8>>, RiegeliError> {
        match self {
            ActiveDecoder::Simple(d) => d.read_record(),
            ActiveDecoder::Transposed(d) => d.read_record(),
        }
    }
}

/// A reader that parses a Riegeli file record by record.
pub struct RecordReader<R: Read + Seek> {
    /// The underlying I/O source.
    reader: R,
    /// Optional recovery callback for corrupted regions.
    recovery: RecoveryCallback,
    /// Optional field projection for column pruning in transpose chunks.
    field_projection: Option<FieldProjection>,
    /// File position of the chunk currently being decoded (its `ChunkHeader` starts here).
    current_chunk_begin: u64,
    /// File position where the NEXT chunk header will be read from.
    next_chunk_file_pos: u64,
    /// The decoder for the current chunk, if one has been loaded.
    current_decoder: Option<ActiveDecoder>,
    /// How many records have been yielded from the current chunk.
    current_record_index: u64,
    /// Logical read-cursor position: points at the next record to be returned.
    pos: RecordPosition,
    /// Position of the last successfully read record.
    last_pos: RecordPosition,
    /// True once we've hit EOF (no more chunks).
    at_eof: bool,
    /// True if the last record was read from a valid (non-recovered) chunk.
    last_record_is_valid: bool,
    /// Stream length as last measured (re-measured on demand if a chunk's
    /// claims exceed it, so a file growing between reads keeps working).
    /// Bounds header-claimed sizes before they drive arithmetic or
    /// allocation — the header hash proves integrity, not honesty.
    stream_len: u64,
    /// The most recent region reported to the recovery callback (whether it
    /// continued or cancelled). `None` until the callback first fires.
    last_skipped_region: Option<crate::SkippedRegion>,
    /// Failure-time classification for recovery: `Some(chunk_end)` when the
    /// failing chunk's header was hash-valid with stream-bounded claims at
    /// the MOMENT of failure (its extent is trustworthy), `None` otherwise.
    /// Set by the error sites, consumed by try_recover_at — never derived
    /// by re-reading, which is a time-of-check/time-of-failure hazard (a
    /// stream that grows between failure and recovery could reclassify an
    /// untrusted failure as trusted and skip a readable chunk).
    pending_trusted_end: Option<u64>,
}

impl<R: Read + Seek> RecordReader<R> {
    /// Open a Riegeli file.
    ///
    /// Validates the initial block header and signature chunk, then positions
    /// the reader at the first data chunk.
    pub fn new(mut reader: R, options: ReaderOptions) -> Result<Self, RiegeliError> {
        let stream_len = reader.seek(SeekFrom::End(0))?;
        reader.seek(SeekFrom::Start(0))?;

        // An empty (0-byte) file is a valid riegeli file with zero records,
        // matching the C++ reference: FailReading only marks truncation
        // when some bytes of a chunk were consumed, so an empty source
        // reads as a clean EOF with the reader still ok() and Close()
        // succeeding. (1..=63 bytes IS an error in both implementations —
        // the validation below rejects it.)
        if stream_len == 0 {
            return Ok(Self {
                reader,
                recovery: options.recovery,
                field_projection: options.field_projection,
                current_chunk_begin: BLOCK_HEADER_SIZE,
                next_chunk_file_pos: BLOCK_HEADER_SIZE + CHUNK_HEADER_SIZE,
                current_decoder: None,
                current_record_index: 0,
                pos: RecordPosition::new(0, 0),
                last_pos: RecordPosition::new(0, 0),
                at_eof: true,
                last_record_is_valid: true,
                stream_len,
                last_skipped_region: None,
                pending_trusted_end: None,
            });
        }

        // File position after the signature chunk: 24 (BH) + 40 (CH) + 0 = 64.
        let next_chunk_file_pos = BLOCK_HEADER_SIZE + CHUNK_HEADER_SIZE;

        // Initial position matches the C++ reference: numeric 0 (the
        // beginning of the file), not the first chunk's canonical address
        // 24 — verified by the differential harness (an earlier criterion
        // documented 24; the reference disagrees and wins).
        let initial_pos = RecordPosition::new(0, 0);

        let mut this = Self {
            reader,
            recovery: options.recovery,
            field_projection: options.field_projection,
            current_chunk_begin: BLOCK_HEADER_SIZE,
            next_chunk_file_pos,
            current_decoder: None,
            current_record_index: 0,
            pos: initial_pos,
            last_pos: initial_pos,
            at_eof: false,
            last_record_is_valid: true,
            stream_len,
            last_skipped_region: None,
            pending_trusted_end: None,
        };

        if let Err(e) = this.validate_file_preamble() {
            // C++ treats both preamble failures as recoverable corruption
            // (Recoverable::kFindChunk): with a recovery function the
            // skipped region [0, resync) is reported and reading continues
            // at the resync position — the recovery contract applies to the
            // most common real-world corruption location (the beginning of
            // the file) too. Without a callback, the error stands.
            if !this.try_recover_at(0, &e) {
                return Err(e);
            }
        }

        Ok(this)
    }

    /// Validate the leading block header (offset 0) and the file signature
    /// chunk (offset 24). The damage classes C++ recovers from are reported
    /// as `MalformedData`; only I/O failures other than a short file
    /// surface as `IoError`.
    fn validate_file_preamble(&mut self) -> Result<(), RiegeliError> {
        self.reader.seek(SeekFrom::Start(0))?;

        // Read and validate the first block header at offset 0. A short
        // file (1..=63 bytes) is malformed — and, like every other
        // truncation-at-EOF, recoverable as a region running to EOF.
        let mut bh_bytes = [0u8; 24]; // BLOCK_HEADER_SIZE
        if read_full(&mut self.reader, &mut bh_bytes)? < bh_bytes.len() {
            return Err(RiegeliError::MalformedData(
                "truncated Riegeli/records file: incomplete file preamble".into(),
            ));
        }
        let block_hdr = BlockHeader::from_bytes(bh_bytes);
        if !block_hdr.is_valid() {
            return Err(RiegeliError::MalformedData(
                "invalid block header hash at offset 0".into(),
            ));
        }

        // Validate the signature chunk at offset 24 by exact comparison: the
        // riegeli file signature is a fixed 40-byte constant (empty data,
        // zero records). Comparing bytes — rather than checking the hash and
        // type and then trusting the header's claimed sizes — does no
        // arithmetic on attacker-controlled values at all: a hash-valid
        // signature header claiming a huge data_size used to overflow the
        // position sum in debug and seek backward through the i64 cast in
        // release. This matches the C++ reader, which verifies the
        // signature bytes.
        let mut ch_bytes = [0u8; 40]; // CHUNK_HEADER_SIZE
        if read_full(&mut self.reader, &mut ch_bytes)? < ch_bytes.len() {
            return Err(RiegeliError::MalformedData(
                "truncated Riegeli/records file: incomplete file preamble".into(),
            ));
        }
        let canonical = ChunkHeader::from_parts(&[], ChunkType::FileSignature, 0, 0).to_bytes();
        if ch_bytes != canonical {
            return Err(RiegeliError::MalformedData(
                "invalid file signature chunk at offset 24".into(),
            ));
        }

        Ok(())
    }

    /// Read the next record from the file.
    ///
    /// Returns `Ok(Some(bytes))` for a record, `Ok(None)` at EOF, or `Err` on
    /// unrecoverable corruption (when no recovery callback is set).
    pub fn read_record(&mut self) -> Result<Option<Vec<u8>>, RiegeliError> {
        self.pending_trusted_end = None;
        loop {
            // End of file is deliberately NOT latched here: C++
            // PullChunkHeader re-polls the source on every call (it has an
            // explicit "source has grown" branch), so records appended by a
            // concurrent writer after a clean EOF are returned by later
            // calls — the standard tailing loop works. When nothing was
            // appended, load_next_chunk below reports the clean EOF again
            // cheaply.

            // If we have an active decoder, try to get a record from it.
            if let Some(decoder) = &mut self.current_decoder {
                match decoder.read_record()? {
                    Some(rec) => {
                        // Record successfully read.
                        let rec_pos = RecordPosition::new(
                            self.current_chunk_begin,
                            self.current_record_index,
                        );
                        self.last_pos = rec_pos;
                        self.current_record_index += 1;
                        self.pos = RecordPosition::new(
                            self.current_chunk_begin,
                            self.current_record_index,
                        );
                        self.last_record_is_valid = true;
                        return Ok(Some(rec));
                    }
                    None => {
                        // Current chunk exhausted; fall through to load next chunk.
                        self.current_decoder = None;
                    }
                }
            }

            // Try to load the next chunk.
            match self.load_next_chunk() {
                Ok(true) => {
                    // Chunk loaded; loop back to read from it.
                    self.at_eof = false;
                }
                Ok(false) => {
                    // EOF reached.
                    self.at_eof = true;
                    return Ok(None);
                }
                Err(e) => {
                    // Corruption detected. With recovery: report the region
                    // and — on `true` — resume at its end (C++: ReadRecord is
                    // retried). On `false` or without recovery, return the
                    // original error (the reader stays position-stable at
                    // the bad region; a bare retry re-reports it).
                    let at = self.next_chunk_file_pos;
                    if self.try_recover_at(at, &e) {
                        if self.at_eof {
                            // Resync seek failed (region end past a
                            // shrunken stream) — clean end of file.
                            return Ok(None);
                        }
                        // Continue reading from the region end.
                    } else {
                        return Err(e);
                    }
                }
            }
        }
    }

    /// Returns the current logical read position.
    ///
    /// Before the first `read_record()` call, returns the initial position
    /// `{ chunk_begin: 0, record_index: 0 }` (numeric 0, matching the C++
    /// reference). After reading records, points at the next record to be
    /// returned.
    pub fn pos(&self) -> RecordPosition {
        self.pos
    }

    /// Returns the position of the last successfully read record.
    ///
    /// Before any records have been read, returns the initial position
    /// `{ chunk_begin: 0, record_index: 0 }` (numeric 0).
    pub fn last_pos(&self) -> RecordPosition {
        self.last_pos
    }

    /// Seek to a specific record position.
    ///
    /// Loads the chunk at `pos.chunk_begin` and skips `pos.record_index` records.
    pub fn seek(&mut self, pos: RecordPosition) -> Result<(), RiegeliError> {
        self.pending_trusted_end = None;
        // Canonicalize the chunk address: boundary + 24 is an accepted alias
        // of a boundary-coincident chunk's canonical address (the block
        // boundary). All position bookkeeping below stores the canonical
        // form, so pos() after an alias seek agrees with last_pos() and
        // sequential reads of the same record — mixing the two sources must
        // not produce two different numerics for one record.
        let chunk_file_pos = crate::block_arithmetic::canonical_chunk_address(pos.chunk_begin);

        self.reader.seek(SeekFrom::Start(chunk_file_pos))?;

        // Load that chunk.
        self.current_decoder = None;
        self.at_eof = false;
        self.next_chunk_file_pos = chunk_file_pos;
        self.current_chunk_begin = chunk_file_pos;
        self.current_record_index = 0;

        // Load the chunk at this position.
        match self.load_chunk_at(chunk_file_pos) {
            Ok(Some(decoder)) => {
                self.current_decoder = Some(decoder);
            }
            Ok(None) => {
                self.at_eof = true;
                self.pos = RecordPosition::new(chunk_file_pos, pos.record_index);
                // last_pos deliberately NOT updated: it tracks the last
                // successfully READ record, and a seek reads nothing.
                return Ok(());
            }
            Err(e) => {
                // C++: Seek returns the result of the recovery function.
                // On `true` the reader is positioned at the region end.
                let at = self.next_chunk_file_pos;
                if self.try_recover_at(at, &e) {
                    self.pos = RecordPosition::new(self.next_chunk_file_pos, 0);
                    return Ok(());
                }
                return Err(e);
            }
        }

        // Skip record_index records. C++ `ChunkDecoder::SetIndex` clamps:
        // an index past the chunk's end positions the reader at the end of
        // the chunk (the next read continues with the following chunk)
        // rather than failing — a persisted position must stay usable
        // against a file that was re-chunked with fewer records per chunk.
        //
        // The index is interpreted within the ADDRESSED chunk only. When
        // the addressed chunk is a non-data chunk (padding, metadata,
        // signature), load_chunk_at scanned forward to the next data chunk;
        // C++ SetIndex clamps the index against the addressed chunk's zero
        // records, so it must NOT skip records of that following chunk —
        // that silently dropped record_index records.
        let record_index = if self.current_chunk_begin == chunk_file_pos {
            pos.record_index
        } else {
            0
        };
        for _ in 0..record_index {
            if let Some(ref mut dec) = self.current_decoder {
                match dec.read_record()? {
                    Some(_) => {
                        self.current_record_index += 1;
                    }
                    None => break, // clamp to num_records
                }
            }
        }

        self.pos = RecordPosition::new(chunk_file_pos, self.current_record_index);
        // last_pos deliberately NOT updated — see the EOF arm above.
        Ok(())
    }

    /// Seek to the record at or after file position `numeric`.
    ///
    /// Interprets `numeric` as `chunk_begin + record_index` (from `RecordPosition::numeric()`).
    /// Scans forward through the file to find the chunk where `chunk_begin <= numeric`
    /// and returns positioned at `record_index = numeric - chunk_begin` within that chunk.
    pub fn seek_numeric(&mut self, numeric: u64) -> Result<(), RiegeliError> {
        self.pending_trusted_end = None;
        // Resolve the scan origin from the block header at the block
        // boundary at or below the target (C++ SeekToChunkContaining):
        // numeric seeks jump straight to the target's block instead of
        // walking every chunk from the beginning of the file, so corruption
        // in earlier, unrelated regions neither fails the seek nor fires
        // the recovery callback.
        let mut scan_pos = self.resolve_scan_origin(numeric)?;

        loop {
            // peek_chunk_header canonicalizes and skips block headers itself,
            // so scan_pos stays a canonical chunk address. (Pre-skipping here
            // would turn a boundary-coincident chunk's address into the
            // boundary+24 alias and shift its numeric positions by 24.)
            match self.peek_chunk_header(scan_pos) {
                Ok(None) => {
                    // EOF — seek to end.
                    self.at_eof = true;
                    self.pos = RecordPosition::new(scan_pos, 0);
                    // last_pos keeps the last successfully read record.
                    self.current_decoder = None;
                    self.current_chunk_begin = scan_pos;
                    self.current_record_index = 0;
                    // EOF is retriable (a concurrent writer may append), so
                    // the next-chunk cursor must point at the position where
                    // appended chunks would begin.
                    self.next_chunk_file_pos = scan_pos;
                    return Ok(());
                }
                Err(e) if self.recovery.is_some() => {
                    // Skip the invalid region and keep scanning, or cancel
                    // with the original error.
                    if self.try_recover_at(scan_pos, &e) {
                        scan_pos = self.next_chunk_file_pos;
                        continue;
                    }
                    return Err(e);
                }
                Ok(Some(ch)) => {
                    let chunk_begin = scan_pos;
                    let num_records = ch.num_records();
                    let data_size = ch.data_size();

                    // The walk is type-blind, matching the C++
                    // SeekToChunkContaining: it stops at the first chunk
                    // position at/after the target, or at the chunk whose
                    // record range contains it, whatever the chunk type.
                    // Type errors (an unknown type carrying records, an
                    // invalid special chunk) surface when seek() actually
                    // loads the chunk — silently scanning past such a chunk
                    // would resolve the position to the wrong record.
                    if chunk_begin >= numeric {
                        return self.seek(RecordPosition::new(chunk_begin, 0));
                    }
                    if chunk_begin + num_records > numeric {
                        let record_index = numeric - chunk_begin;
                        return self.seek(RecordPosition::new(chunk_begin, record_index));
                    }

                    // Advance to the next chunk.
                    scan_pos =
                        crate::block_arithmetic::chunk_end(chunk_begin, data_size, num_records);
                }
                Err(e) => return Err(e),
            }
        }
    }

    /// Resolve where a position-based scan should begin: the canonical
    /// address of a chunk at or before `target`, obtained from the block
    /// header at the block boundary at or below it (the navigation step of
    /// C++ `DefaultChunkReaderBase::SeekToChunk`). Falls back to the first
    /// chunk after the file preamble whenever the boundary's block header
    /// cannot be used (boundary 0, header unreadable or hash-invalid,
    /// implausible pointers) — the fallback is never less correct than
    /// scanning from the beginning, only slower.
    fn resolve_scan_origin(&mut self, target: u64) -> Result<u64, RiegeliError> {
        let first_data_chunk = BLOCK_HEADER_SIZE + CHUNK_HEADER_SIZE; // = 64

        // A target past the last measured stream length may be valid now:
        // a growing file is a supported way to keep reading.
        if target > self.stream_len {
            self.stream_len = self.reader.seek(SeekFrom::End(0))?;
        }
        let block_begin = round_down_to_block_boundary(target.min(self.stream_len));
        if block_begin == 0 {
            return Ok(first_data_chunk);
        }

        self.reader.seek(SeekFrom::Start(block_begin))?;
        let mut bh_bytes = [0u8; 24]; // BLOCK_HEADER_SIZE
        let n = match read_full(&mut self.reader, &mut bh_bytes) {
            Ok(n) => n,
            Err(_) => return Ok(first_data_chunk),
        };
        if n < bh_bytes.len() {
            return Ok(first_data_chunk);
        }
        let bh = BlockHeader::from_bytes(bh_bytes);
        if !bh.is_valid() {
            return Ok(first_data_chunk);
        }

        let chunk_begin = if bh.previous_chunk() == 0 {
            // A chunk boundary coincides with the block boundary.
            block_begin
        } else {
            let next = block_begin.saturating_add(bh.next_chunk());
            if next > target {
                // The target is inside the chunk spanning this boundary, so
                // the walk must start at that chunk's beginning.
                match block_begin.checked_sub(bh.previous_chunk()) {
                    Some(begin) => begin,
                    None => return Ok(first_data_chunk),
                }
            } else {
                next
            }
        };
        if chunk_begin < first_data_chunk
            || !crate::block_arithmetic::is_possible_chunk_boundary(chunk_begin)
        {
            return Ok(first_data_chunk);
        }
        Ok(chunk_begin)
    }

    /// Locate the chunk whose span contains `target`
    /// (`chunk_begin <= target < chunk_end`), the analogue of C++
    /// `SeekToChunkBefore`. Returns `Ok(None)` when the file ends at or
    /// before `target`.
    fn find_chunk_before(
        &mut self,
        target: u64,
    ) -> Result<Option<(u64, ChunkHeader)>, RiegeliError> {
        let mut scan_pos = self.resolve_scan_origin(target)?;
        loop {
            match self.peek_chunk_header(scan_pos)? {
                None => return Ok(None),
                Some(ch) => {
                    let end = crate::block_arithmetic::chunk_end(
                        scan_pos,
                        ch.data_size(),
                        ch.num_records(),
                    );
                    if end > target {
                        return Ok(Some((scan_pos, ch)));
                    }
                    scan_pos = end;
                }
            }
        }
    }

    /// Returns `true` since file-based I/O supports seeking.
    pub fn supports_random_access(&self) -> bool {
        true
    }

    /// Read the file metadata chunk as a typed `RecordsMetadata` proto, if present.
    ///
    /// Peeks at the chunk immediately after the file signature (offset 64) to check
    /// if it is a `ChunkType::FileMetadata` chunk. If so, parses and returns the
    /// `RecordsMetadata` message. Does not change the current read position.
    pub fn read_metadata(&mut self) -> Result<Option<crate::RecordsMetadata>, RiegeliError> {
        use protobuf::Parse;
        match self.read_serialized_metadata()? {
            Some(bytes) => {
                let msg = crate::RecordsMetadata::parse(&bytes).map_err(|e| {
                    RiegeliError::MalformedData(
                        format!("failed to parse RecordsMetadata: {e}").into(),
                    )
                })?;
                Ok(Some(msg))
            }
            None => Ok(None),
        }
    }

    /// Read the file metadata chunk as raw bytes, if present.
    ///
    /// Like [`read_metadata`](Self::read_metadata), but returns the raw serialized
    /// proto bytes without parsing. Does not change the current read position —
    /// including under recovery: corruption at the metadata position is
    /// reported to the callback but the region is never consumed, so a
    /// repeated call reports the same region again (the one deliberate
    /// exception to the consumed-once cancel contract, because consuming
    /// would move a position this method promises not to touch).
    ///
    /// The chunk data is transpose-encoded (the C++ writer runs the serialized
    /// `RecordsMetadata` through `TransposeEncoder`); this decodes it back to
    /// the serialized proto bytes, mirroring `RecordReaderBase::ParseMetadata`.
    pub fn read_serialized_metadata(&mut self) -> Result<Option<Vec<u8>>, RiegeliError> {
        self.pending_trusted_end = None;
        // The metadata chunk, if present, is at offset 64 (right after signature).
        let metadata_chunk_pos = BLOCK_HEADER_SIZE + CHUNK_HEADER_SIZE; // = 64

        // Peek at the chunk header at offset 64. A clean EOF (file ends
        // before any chunk) means no metadata; a real error — hash-invalid
        // header, impossible claims, I/O failure — is corruption and must
        // not be reported as "no metadata": a caller inspecting metadata
        // first would proceed as if the file were clean.
        let ch = match self.peek_chunk_header(metadata_chunk_pos) {
            Ok(Some(ch)) => ch,
            Ok(None) => return Ok(None),
            Err(e) => {
                // C++: ReadMetadata returns the result of the recovery
                // function — on `true` the file simply has no (readable)
                // metadata. REPORT-ONLY: this method's contract is that it
                // does not change the read position (C++'s ReadMetadata is
                // a sequential read with no such promise, so there is no
                // reference position behavior to match — ours must honor
                // our own documentation). Repositioning here rewound a
                // mid-stream reader to the skipped region's end, replaying
                // or dropping records.
                if self.report_region_at(metadata_chunk_pos, &e) == Some(true) {
                    return Ok(None);
                }
                return Err(e);
            }
        };

        if !matches!(ch.chunk_type(), Ok(ChunkType::FileMetadata)) {
            return Ok(None);
        }

        // Every post-peek failure mode — structural invalidity, unreadable
        // or hash-mismatching chunk data, transpose decode failure — also
        // consults the recovery callback, matching the C++ reference:
        // ReadSerializedMetadata routes a ReadChunk failure through the
        // recovery function (kRecoverChunkReader) and a ParseMetadata
        // failure through kRecoverChunkDecoder, and on recovery returns
        // success with no metadata. REPORT-ONLY, like the peek failure
        // above: the read position is never moved.
        match self.read_metadata_chunk_payload(metadata_chunk_pos, &ch) {
            Ok(metadata) => Ok(Some(metadata)),
            Err(e) => {
                if self.report_region_at(metadata_chunk_pos, &e) == Some(true) {
                    return Ok(None);
                }
                Err(e)
            }
        }
    }

    /// Read, validate, and transpose-decode the FileMetadata chunk at
    /// `metadata_chunk_pos` (its hash-valid header is `ch`), returning the
    /// serialized `RecordsMetadata` bytes.
    fn read_metadata_chunk_payload(
        &mut self,
        metadata_chunk_pos: u64,
        ch: &ChunkHeader,
    ) -> Result<Vec<u8>, RiegeliError> {
        // A metadata chunk claiming records is structurally invalid
        // (C++ ChunkDecoder::Parse rejects it).
        validate_special_chunk(ch, metadata_chunk_pos, ChunkType::FileMetadata)?;

        // Read the chunk data. The metadata chunk header is at offset 64, far
        // from any block boundary, so its data always begins at 64 + 40.
        let metadata_chunk_end = crate::block_arithmetic::chunk_end(
            metadata_chunk_pos,
            ch.data_size(),
            ch.num_records(),
        );
        let data = self.read_chunk_data(
            metadata_chunk_pos,
            metadata_chunk_end,
            metadata_chunk_pos + CHUNK_HEADER_SIZE,
            ch.data_size(),
        )?;

        // Header hash-valid, claims stream-bounded, data physically present:
        // the chunk's extent is trustworthy for any failure from here on, so
        // the region reported to the recovery callback is exactly this chunk.
        self.pending_trusted_end = Some(metadata_chunk_end);

        // Validate data hash.
        if !ch.is_data_valid(&data) {
            return Err(RiegeliError::MalformedData(
                "metadata chunk data hash mismatch".into(),
            ));
        }

        // Decode the transpose-encoded chunk data as a single record whose
        // bytes are the serialized RecordsMetadata (the C++ reader calls
        // TransposeDecoder::Decode with num_records = 1 and the chunk header's
        // decoded_data_size).
        let decode_header = crate::chunk_header::ChunkHeader::from_parts(
            &data,
            ChunkType::Transposed,
            1,
            ch.decoded_data_size(),
        );
        let chunk = crate::simple_chunk::Chunk {
            header: decode_header,
            data,
        };
        let mut decoder = TransposeChunkDecoder::new(chunk)?;
        decoder.read_record()?.ok_or_else(|| {
            RiegeliError::MalformedData("file metadata chunk decoded to no records".into())
        })
    }

    /// Change the active field projection, taking effect at the next chunk boundary.
    ///
    /// The current chunk decoder (if any) continues with the old projection until
    /// it is exhausted. New chunks loaded after this call will use the new projection.
    ///
    /// To switch back to returning all fields, pass `FieldProjection::all()`.
    pub fn set_field_projection(&mut self, proj: FieldProjection) {
        self.field_projection = if proj.is_all() { None } else { Some(proj) };
    }

    /// Binary search for a record in a sorted file.
    ///
    /// `test` is called with the bytes of individual records; it should return
    /// `Ordering::Less` if the target is after this record, `Ordering::Greater`
    /// if before, and `Ordering::Equal` if this is the target record.
    ///
    /// After a successful search, the reader is positioned so that the next
    /// `read_record()` returns the found record.
    ///
    /// Returns `Ok(true)` if a record for which `test` returns `Equal` was found,
    /// `Ok(false)` if the target does not exist in the file.
    ///
    /// The search reads at most O(log N) records where N is the total number of records.
    pub fn search<F>(&mut self, mut test: F) -> Result<bool, RiegeliError>
    where
        F: FnMut(&[u8]) -> Ordering,
    {
        // Collect all data chunk positions and their record counts.
        let chunks = self.collect_data_chunks()?;

        if chunks.is_empty() {
            self.park_at_end_of_stream();
            return Ok(false);
        }

        // Binary search over chunks using the first record of each chunk as a pivot.
        // Invariant: if the target exists, it is in chunks[lo..hi].
        let mut lo = 0usize;
        let mut hi = chunks.len();

        while lo < hi {
            let mid = lo + (hi - lo) / 2;

            // Probe the first record of chunks[mid]. An unreadable pivot
            // (e.g. a chunk skipped under recovery because its data hash
            // does not match) is UNORDERED — C++ SearchImpl declares the
            // skipped region unordered and the binary search shrinks around
            // it — so scan right for the nearest readable pivot instead of
            // judging fabricated record content.
            let mut probe = mid;
            let mut ordering = None;
            while probe < hi {
                let (chunk_pos, _num_records) = chunks[probe];
                match self.read_record_at(chunk_pos, 0)? {
                    Some(first_record) => {
                        ordering = Some((probe, chunk_pos, test(&first_record)));
                        break;
                    }
                    None => probe += 1,
                }
            }

            match ordering {
                None => {
                    // Everything in [mid, hi) is unordered (unreadable): the
                    // target, if findable at all, is in the left half.
                    hi = mid;
                }
                Some((probe, _, Ordering::Less)) => {
                    // Target is after this chunk's first record → search
                    // right of the probe (chunks [mid, probe) are unordered).
                    lo = probe + 1;
                }
                Some((_, _, Ordering::Greater)) => {
                    // Target is before this chunk's first record → search
                    // left half. Unordered chunks in [mid, probe) cannot be
                    // judged, so a target inside them is reported absent
                    // (the same outcome C++ gives for skipped regions).
                    hi = mid;
                }
                Some((_, chunk_pos, Ordering::Equal)) => {
                    // First record of this chunk matches. Seek to it and return.
                    let target = crate::record_position::RecordPosition::new(chunk_pos, 0);
                    self.seek(target)?;
                    return Ok(true);
                }
            }
        }

        // lo == hi: the target might be inside chunks[lo-1].
        // That chunk's first record is < target (test returned Less), but a later
        // record in that chunk might equal the target.
        if lo > 0 {
            let (chunk_pos, num_records) = chunks[lo - 1];
            let found = self.binary_search_within_chunk(chunk_pos, num_records, &mut test)?;
            if found {
                return Ok(true);
            }
        }

        // Target not found in the file.
        self.park_at_end_of_stream();
        Ok(false)
    }

    /// Position the reader at the end of the stream. End of file is a
    /// retriable condition for `read_record` (a concurrent writer may
    /// append), so the next-chunk cursor must point where appended chunks
    /// would begin — not at whatever position an internal scan or probe
    /// left it.
    fn park_at_end_of_stream(&mut self) {
        self.at_eof = true;
        self.current_decoder = None;
        self.next_chunk_file_pos = self.stream_len.max(BLOCK_HEADER_SIZE + CHUNK_HEADER_SIZE);
        self.current_chunk_begin = self.next_chunk_file_pos;
        self.current_record_index = 0;
    }

    /// Collect (file_pos, num_records) for all Simple and Transposed data chunks.
    ///
    /// Scans the entire file, reading only chunk headers (no data decompression).
    fn collect_data_chunks(&mut self) -> Result<Vec<(u64, u64)>, RiegeliError> {
        self.pending_trusted_end = None;
        let first_data_chunk = BLOCK_HEADER_SIZE + CHUNK_HEADER_SIZE; // = 64
        let mut scan_pos = first_data_chunk;
        let mut chunks = Vec::new();

        // Read chunk headers until EOF (skipping leading and interleaved
        // block headers). With recovery set, invalid regions are skipped and
        // the scan continues (C++: Search skips invalid regions while the
        // recovery function returns true).
        loop {
            let (ch, chunk_begin, _) = match self.read_chunk_header_at(scan_pos) {
                Ok(Some(v)) => v,
                Ok(None) => break,
                Err(e) => {
                    if self.try_recover_at(scan_pos, &e) {
                        scan_pos = self.next_chunk_file_pos;
                        continue;
                    }
                    return Err(e);
                }
            };
            if !ch.is_header_valid() {
                let e = RiegeliError::MalformedData(
                    format!("invalid chunk header at {chunk_begin} during search scan").into(),
                );
                if self.try_recover_at(scan_pos, &e) {
                    scan_pos = self.next_chunk_file_pos;
                    continue;
                }
                return Err(e);
            }

            let data_size = ch.data_size();
            let num_records = ch.num_records();

            // Zero-record chunks cannot serve as search pivots: probing one
            // via read_record_at falls through to the NEXT chunk's first
            // record, so the comparator sees a misattributed record and the
            // search can report a present target as absent.
            if num_records > 0
                && matches!(
                    ch.chunk_type(),
                    Ok(ChunkType::Simple) | Ok(ChunkType::Transposed)
                )
            {
                chunks.push((chunk_begin, num_records));
            }

            scan_pos = crate::block_arithmetic::chunk_end(chunk_begin, data_size, num_records);
        }

        Ok(chunks)
    }

    /// Read the record at `record_index` within the chunk at `chunk_pos`.
    ///
    /// Uses `seek()` to position at the exact record. Does NOT preserve reader state.
    ///
    /// Returns `Ok(None)` when the requested record could not be read AS a
    /// record of that chunk — the chunk was skipped under recovery or the
    /// read fell through to a different chunk. `None` is distinct from an
    /// empty record (empty records are legal), so callers can treat the
    /// probe as unordered instead of judging fabricated content.
    fn read_record_at(
        &mut self,
        chunk_pos: u64,
        record_index: u64,
    ) -> Result<Option<Vec<u8>>, RiegeliError> {
        let target = crate::record_position::RecordPosition::new(chunk_pos, record_index);
        self.seek(target)?;
        match self.read_record()? {
            // Guard against misattribution: if the read fell through to a
            // different chunk (e.g. the requested chunk had no records or
            // was skipped by recovery), report absence rather than another
            // chunk's record.
            Some(rec) if self.current_chunk_begin == chunk_pos => Ok(Some(rec)),
            _ => Ok(None),
        }
    }

    /// Binary search within a single chunk for a matching record.
    ///
    /// Uses O(log num_records) reads by seeking to specific record indices.
    /// On success, positions the reader at the matching record.
    fn binary_search_within_chunk<F>(
        &mut self,
        chunk_pos: u64,
        num_records: u64,
        test: &mut F,
    ) -> Result<bool, RiegeliError>
    where
        F: FnMut(&[u8]) -> Ordering,
    {
        if num_records == 0 {
            return Ok(false);
        }

        // Binary search over record indices [0, num_records).
        // Invariant: if the target is in this chunk, it is at index [lo, hi).
        // We already know record 0 gives test == Less (from the outer binary search).
        let mut lo = 1u64; // record 0 was already checked and returned Less
        let mut hi = num_records;

        while lo < hi {
            let mid = lo + (hi - lo) / 2;
            let rec = match self.read_record_at(chunk_pos, mid)? {
                Some(rec) => rec,
                // The chunk became unreadable (skipped under recovery):
                // chunks decode all-or-nothing, so none of its records can
                // be judged and the target cannot be found here.
                None => return Ok(false),
            };

            match test(&rec) {
                Ordering::Less => {
                    lo = mid + 1;
                }
                Ordering::Greater => {
                    hi = mid;
                }
                Ordering::Equal => {
                    // Found the target. Position the reader at this record.
                    let target = crate::record_position::RecordPosition::new(chunk_pos, mid);
                    self.seek(target)?;
                    return Ok(true);
                }
            }
        }

        // Target not in this chunk.
        Ok(false)
    }

    /// Returns `true` if the most recently returned record came from a valid
    /// (non-recovered) chunk.
    ///
    /// Returns `true` initially (before any record is read) and after each
    /// successful record read. Returns `false` after a recovery callback fires
    /// due to a corrupted chunk.
    pub fn last_record_is_valid(&self) -> bool {
        self.last_record_is_valid
    }

    /// Seek one record back from the current position (C++ `SeekBack`).
    ///
    /// If the current position is past the first record of the current
    /// chunk, steps to the previous record of that chunk; otherwise walks
    /// back chunk by chunk to the last record of the nearest preceding
    /// chunk that has records. Repeated calls without intervening reads
    /// step backward through the file, one record per call.
    ///
    /// Immediately after a successful `read_record()`, the next
    /// `read_record()` after `seek_back()` returns that same record again.
    ///
    /// Returns `Ok(true)` if there is a previous record to seek to.
    /// Returns `Ok(false)` if positioned at or before the first record.
    pub fn seek_back(&mut self) -> Result<bool, RiegeliError> {
        self.pending_trusted_end = None;

        // One record back within the current chunk
        // (C++: SetIndex(index - 1)).
        if self.current_record_index > 0 {
            let target =
                RecordPosition::new(self.current_chunk_begin, self.current_record_index - 1);
            self.seek(target)?;
            return Ok(true);
        }

        // At the beginning of a chunk: walk back chunk by chunk (C++
        // SeekToChunkBefore) until a chunk that has records, and position
        // at its last record. Only the file preamble — which never carries
        // records — precedes the first chunk at offset 64.
        let first_data_chunk = BLOCK_HEADER_SIZE + CHUNK_HEADER_SIZE; // = 64
        let mut chunk_pos = self.current_chunk_begin;
        while chunk_pos > first_data_chunk {
            let (chunk_begin, ch) = match self.find_chunk_before(chunk_pos - 1)? {
                Some(v) => v,
                None => return Ok(false),
            };
            if ch.num_records() > 0 {
                self.seek(RecordPosition::new(chunk_begin, ch.num_records() - 1))?;
                return Ok(true);
            }
            if chunk_begin >= chunk_pos {
                // No backward progress (corrupt layout); stop rather than spin.
                return Ok(false);
            }
            chunk_pos = chunk_begin;
        }
        Ok(false)
    }

    /// Return the total number of records in the file.
    ///
    /// Scans all chunk headers summing `num_records` without decompressing any
    /// record data. The current read position is preserved — the next
    /// `read_record()` after `size()` returns the same record it would have
    /// without the `size()` call.
    pub fn size(&mut self) -> Result<u64, RiegeliError> {
        self.pending_trusted_end = None;
        // The scan below reads chunk headers directly from the underlying
        // reader and touches NO logical reader state: the current chunk's
        // decoder survives untouched, so the read position, last_pos, and
        // the projection the in-flight chunk was decoded with are all
        // preserved (set_field_projection's next-chunk-boundary timing
        // included). Every read path re-seeks the underlying reader
        // explicitly, so the physical position needs no restoring either.
        //
        // Corruption honors the recovery contract the same way the search
        // scan does, but REPORT-ONLY (like metadata reads): the region is
        // reported to the callback and the scan continues past it, without
        // consuming the region or moving the read position — size() must
        // not be stricter than read_record() on a file the reader can read
        // under its configured recovery policy.
        let first_data_chunk = BLOCK_HEADER_SIZE + CHUNK_HEADER_SIZE; // = 64
        let mut scan_pos = first_data_chunk;
        let mut total_records: u64 = 0;

        loop {
            // Read chunk header (skipping leading and interleaved block headers).
            let (ch, chunk_begin) = match self.read_chunk_header_at(scan_pos) {
                Ok(Some((ch, chunk_begin, _))) if ch.is_header_valid() => (ch, chunk_begin),
                Ok(Some((_, chunk_begin, _))) => {
                    let e = RiegeliError::MalformedData(
                        format!("invalid chunk header at {chunk_begin} during size scan").into(),
                    );
                    if self.report_region_at(scan_pos, &e) == Some(true) {
                        scan_pos = self
                            .last_skipped_region
                            .as_ref()
                            .expect("set by report_region_at")
                            .end();
                        continue;
                    }
                    return Err(e);
                }
                Ok(None) => break, // EOF
                Err(e) => {
                    if self.report_region_at(scan_pos, &e) == Some(true) {
                        scan_pos = self
                            .last_skipped_region
                            .as_ref()
                            .expect("set by report_region_at")
                            .end();
                        continue;
                    }
                    return Err(e);
                }
            };

            let data_size = ch.data_size();
            let num_records = ch.num_records();

            if matches!(
                ch.chunk_type(),
                Ok(ChunkType::Simple) | Ok(ChunkType::Transposed)
            ) {
                total_records += num_records;
            }

            // Advance past this chunk.
            scan_pos = crate::block_arithmetic::chunk_end(chunk_begin, data_size, num_records);
        }

        Ok(total_records)
    }

    /// Validate all block and chunk headers and data hashes in the file.
    ///
    /// Does not decompress any record data — only validates the raw (possibly
    /// compressed) chunk data against the stored hash. Returns `Ok(())` if all
    /// headers and data hashes are valid, or `Err(RiegeliError::MalformedData(_))`
    /// on the first validation failure.
    ///
    /// The current read position is not changed by this method.
    pub fn check_file_format(&mut self) -> Result<(), RiegeliError> {
        // Validate the initial block header at offset 0.
        self.reader.seek(SeekFrom::Start(0))?;
        let mut bh_bytes = [0u8; 24];
        self.reader.read_exact(&mut bh_bytes)?;
        let bh = BlockHeader::from_bytes(bh_bytes);
        if !bh.is_valid() {
            return Err(RiegeliError::MalformedData(
                "invalid block header hash at offset 0".into(),
            ));
        }

        // Scan all chunks starting from the signature chunk (offset 24).
        let mut scan_pos: u64 = BLOCK_HEADER_SIZE; // = 24

        // Read chunk headers until EOF (skipping leading and interleaved block headers).
        while let Some((ch, chunk_begin, data_begin)) = self.read_chunk_header_at(scan_pos)? {
            if !ch.is_header_valid() {
                return Err(RiegeliError::MalformedData(
                    format!("invalid chunk header hash at offset {chunk_begin}").into(),
                ));
            }

            let data_size = ch.data_size();
            let num_records = ch.num_records();
            let chunk_end = crate::block_arithmetic::chunk_end(chunk_begin, data_size, num_records);

            // Read the raw chunk data (without decompressing) and validate data hash.
            let chunk_data = self.read_chunk_data(chunk_begin, chunk_end, data_begin, data_size)?;
            if !ch.is_data_valid(&chunk_data) {
                return Err(RiegeliError::MalformedData(
                    format!("chunk data hash mismatch at offset {chunk_begin}").into(),
                ));
            }

            // Advance past this chunk.
            scan_pos = chunk_end;
        }

        Ok(())
    }

    // -------------------------------------------------------------------------
    // Internal helpers
    // -------------------------------------------------------------------------

    /// The most recent region reported to the recovery callback, whether the
    /// callback continued or cancelled. `None` if the callback never fired.
    ///
    /// This is the Rust stand-in for the reader parameter the C++ recovery
    /// callback receives: after a cancelled operation returns its error, the
    /// caller can inspect the region here and reposition explicitly.
    pub fn last_skipped_region(&self) -> Option<&crate::SkippedRegion> {
        self.last_skipped_region.as_ref()
    }

    /// Attempt recovery for `error`, which left the reader position-stable
    /// at `at` (every error path rewinds to the failed chunk — the
    /// persistence invariant).
    ///
    /// Computes the skipped region — COUPLED to the resync target by
    /// construction: `region.end` is exactly where reading resumes.
    /// A readable, hash-valid header (whose claims passed the stream bound)
    /// gives a trustworthy extent, so exactly that chunk is skipped
    /// (preserving siblings in the same block — more precise than a
    /// boundary skip). Otherwise the claims cannot be trusted and the
    /// resync is the next block boundary.
    ///
    /// Returns `true` if a callback is set, it returned `true`, and the
    /// reader was repositioned to the region end; `false` otherwise (the
    /// caller returns the original error).
    /// Compute the region for `error` and invoke the callback WITHOUT any
    /// repositioning side effects — for operations whose contract promises
    /// not to move the read cursor (metadata reads). Returns `None` when no
    /// callback is set, otherwise `Some(callback verdict)`.
    fn report_region_at(&mut self, at: u64, error: &RiegeliError) -> Option<bool> {
        // Consume the failure-time classification unconditionally so it can
        // never go stale across attempts or operations.
        let trusted_end = self.pending_trusted_end.take();
        self.recovery.as_ref()?;
        let begin = crate::block_arithmetic::canonical_chunk_address(at);
        // Canonicalizing a boundary+24 failure address steps 24 bytes back
        // across the block header — bytes the PREVIOUS region already covers
        // when its resync landed exactly on that alias (a block header whose
        // next_chunk is 24). Clamp to the previous region's end so regions
        // reported during forward reads never overlap; an explicit backward
        // seek has prev.end() > at and is unaffected.
        let begin = match &self.last_skipped_region {
            Some(prev) if prev.end() <= at && begin < prev.end() => prev.end(),
            _ => begin,
        };
        let end = match trusted_end {
            // The failing chunk's extent was trustworthy WHEN it failed
            // (hash-valid header, stream-bounded claims): skip exactly it.
            Some(end) => end,
            // Header unreadable, hash-invalid, claims unvalidated, or I/O
            // failure: nothing about the extent can be trusted — resync via
            // the next block boundary's header, clamped to the stream
            // length (a region cannot extend past the file — except by the
            // minimal progress margin when corruption sits at EOF — and
            // C++ reports EOF-ended regions the same way).
            None => {
                let boundary = next_block_boundary(begin);
                let boundary = if boundary == begin {
                    begin + BLOCK_SIZE
                } else {
                    boundary
                };
                self.resolve_boundary_resync(boundary)
                    .min(self.stream_len.max(begin))
            }
        };
        // Forward progress no matter what the arithmetic said — measured
        // from the RESYNC ORIGIN `at`, not just `begin`: when `at` is the
        // boundary+24 alias of a canonical `begin` 24 bytes earlier, an
        // EOF-clamped end can satisfy end > begin while still equaling
        // `at`, and the reader would spin on the same region forever (a
        // past-EOF end is fine — the resync seek lands at EOF and reads
        // terminate).
        let end = end.max(begin + 1).max(at.saturating_add(1));
        // The resume position must never be a canonical ALIAS of this
        // region's own begin: an EOF-clamped end landing on begin's
        // boundary+24 alias would re-read the same corrupt bytes at
        // canonical begin and double-report overlapping regions (breaking
        // begin-monotonicity). The condition only triggers for that exact
        // alias, so the bump is at most one byte and cannot loop.
        let end = if crate::block_arithmetic::canonical_chunk_address(end) <= begin {
            end + 1
        } else {
            end
        };
        let region = crate::SkippedRegion::new(begin, end, error.to_string());
        self.last_skipped_region = Some(region);
        let cb = self.recovery.as_mut().expect("checked above");
        let region_ref = self.last_skipped_region.as_ref().expect("just set");
        Some(cb(region_ref))
    }

    /// Resolve the resync position for an untrusted failure, starting from
    /// `boundary` (a block boundary strictly after the failed region's
    /// begin). Mirrors the C++ `DefaultChunkReaderBase::Recover` kFindChunk
    /// path: chunks do not generally start at block boundaries — after the
    /// boundary's 24-byte block header, the bytes are normally the TAIL of
    /// a chunk spanning the boundary — so the block header's `next_chunk`
    /// pointer is followed to land exactly on the next chunk start.
    /// `previous_chunk == 0` means a chunk begins at the boundary itself.
    /// A hash-invalid block header or an implausible pointer pushes the
    /// walk to the next boundary; EOF ends the walk (the caller clamps the
    /// result to the stream length).
    ///
    /// Only the underlying reader's physical position moves; all logical
    /// state is untouched (report-only callers rely on that, and every
    /// read path re-seeks explicitly).
    fn resolve_boundary_resync(&mut self, mut boundary: u64) -> u64 {
        loop {
            if self.reader.seek(SeekFrom::Start(boundary)).is_err() {
                return boundary;
            }
            let mut bh_bytes = [0u8; 24]; // BLOCK_HEADER_SIZE
            let n = match read_full(&mut self.reader, &mut bh_bytes) {
                Ok(n) => n,
                Err(_) => return boundary,
            };
            if n < bh_bytes.len() {
                // EOF at or inside the boundary's block header: the region
                // runs to the end of the stream.
                return boundary;
            }
            let bh = BlockHeader::from_bytes(bh_bytes);
            if !bh.is_valid() {
                // Untrustworthy block header: try the next boundary
                // (C++ goes back to find_chunk).
                boundary += BLOCK_SIZE;
                continue;
            }
            if bh.previous_chunk() == 0 {
                // A chunk boundary coincides with the block boundary.
                return boundary;
            }
            let target = boundary.saturating_add(bh.next_chunk());
            if target > self.stream_len {
                // Past the end of the stream; the caller clamps to EOF
                // (C++: the resync seek fails and the region ends there).
                return target;
            }
            if crate::block_arithmetic::is_possible_chunk_boundary(target) {
                return target;
            }
            // Implausible pointer (lands inside a block header): continue
            // from the next boundary after it (C++ goto find_chunk).
            boundary = next_block_boundary(target);
        }
    }

    fn try_recover_at(&mut self, at: u64, error: &RiegeliError) -> bool {
        let Some(go) = self.report_region_at(at, error) else {
            // No callback configured: leave the reader untouched so the
            // error stays position-stable and persistent on retry.
            return false;
        };
        // The region is consumed REGARDLESS of the verdict, matching the
        // C++ reference (Recover() repositions before the callback is
        // consulted): cancel reports the error once, but the next
        // operation continues past the rejected region. This is what
        // makes a naive retry loop around a cancelling callback make
        // progress instead of re-firing the same region forever.
        let end = self
            .last_skipped_region
            .as_ref()
            .expect("set by report_region_at")
            .end();
        self.last_record_is_valid = false;
        self.current_decoder = None;
        self.at_eof = false;
        self.next_chunk_file_pos = end;
        if self.reader.seek(SeekFrom::Start(end)).is_err() {
            self.at_eof = true;
        }
        go
    }

    /// Load the next chunk from `self.next_chunk_file_pos`.
    ///
    /// Returns `Ok(true)` if a chunk was loaded into `self.current_decoder`.
    /// Returns `Ok(false)` at EOF.
    /// Returns `Err` on corruption (without recovery).
    fn load_next_chunk(&mut self) -> Result<bool, RiegeliError> {
        loop {
            // Each chunk attempt re-classifies from scratch.
            self.pending_trusted_end = None;
            let pos = self.next_chunk_file_pos;

            // Read the chunk header, skipping any leading block header and any
            // block header interleaved within the 40-byte header span.
            // `chunk_begin` is the position of the header's first byte — the
            // chunk's canonical address.
            let (ch, chunk_begin, data_begin) = match self.read_chunk_header_at(pos)? {
                Some(v) => v,
                None => return Ok(false), // EOF
            };

            if !ch.is_header_valid() {
                return Err(RiegeliError::MalformedData(
                    format!("invalid chunk header hash at file position {chunk_begin}").into(),
                ));
            }

            let data_size = ch.data_size();
            let num_records = ch.num_records();

            // Compute where the chunk data ends in the file (accounting for block headers).
            let data_file_end =
                crate::block_arithmetic::chunk_end(chunk_begin, data_size, num_records);

            // Read the chunk data (skipping block headers).
            let chunk_data =
                self.read_chunk_data(chunk_begin, data_file_end, data_begin, data_size)?;

            // Past this point the header is hash-valid, its claims are
            // stream-bounded, and the data bytes were physically present:
            // the chunk's extent is trustworthy at failure time, whatever
            // the failure (bad data hash, unknown type, construction).
            self.pending_trusted_end = Some(data_file_end);

            // Validate data hash.
            if !ch.is_data_valid(&chunk_data) {
                // Leave next_chunk_file_pos at the chunk start so the error
                // persists on retry and recovery scans from the right place.
                self.next_chunk_file_pos = chunk_begin;
                return Err(RiegeliError::MalformedData(
                    format!("chunk data hash mismatch at file position {chunk_begin}").into(),
                ));
            }

            // Update state for the next chunk.
            self.next_chunk_file_pos = data_file_end;

            // Resolve the chunk type only after next_chunk_file_pos has been
            // advanced: an unknown type must skip the whole chunk (forward
            // compatibility), and skipping requires the loop to make progress.
            //
            // Matching the C++ ChunkDecoder: an unknown chunk type is ignored
            // only when it carries no records; skipping a chunk with records
            // would lose data silently, so that case is an error.
            let chunk_type = match ch.chunk_type() {
                Ok(ct) => ct,
                Err(_) if num_records == 0 => continue,
                Err(e) => {
                    // Mirror the hash-mismatch convention: reset to the chunk
                    // start so the error is persistent — a bare retry must
                    // re-hit this chunk, not silently resume past its
                    // (dropped) records at the next chunk.
                    self.next_chunk_file_pos = chunk_begin;
                    return Err(e);
                }
            };

            match chunk_type {
                ChunkType::Simple => {
                    let chunk = Chunk {
                        header: ch,
                        data: chunk_data,
                    };
                    // Persistence: construction failures (malformed chunk
                    // interior behind valid hashes) must rewind like the
                    // hash-mismatch and unknown-type paths do — a bare retry
                    // must re-hit this chunk, not silently skip its records.
                    let decoder = match SimpleChunkDecoder::new(chunk) {
                        Ok(d) => d,
                        Err(e) => {
                            self.next_chunk_file_pos = chunk_begin;
                            return Err(e);
                        }
                    };
                    self.current_chunk_begin = chunk_begin;
                    self.current_record_index = 0;
                    self.pos = RecordPosition::new(chunk_begin, 0);
                    self.current_decoder = Some(ActiveDecoder::Simple(decoder));
                    let _ = num_records;
                    return Ok(true);
                }
                ChunkType::Transposed => {
                    let chunk = Chunk {
                        header: ch,
                        data: chunk_data,
                    };
                    // Same persistence convention as the Simple arm above.
                    let decoder = match TransposeChunkDecoder::new_with_projection(
                        chunk,
                        self.field_projection.as_ref(),
                    ) {
                        Ok(d) => d,
                        Err(e) => {
                            self.next_chunk_file_pos = chunk_begin;
                            return Err(e);
                        }
                    };
                    self.current_chunk_begin = chunk_begin;
                    self.current_record_index = 0;
                    self.pos = RecordPosition::new(chunk_begin, 0);
                    self.current_decoder = Some(ActiveDecoder::Transposed(decoder));
                    let _ = num_records;
                    return Ok(true);
                }
                ChunkType::FileSignature | ChunkType::Padding | ChunkType::FileMetadata => {
                    // Validate the structural invariants before skipping
                    // (C++ ChunkDecoder::Parse rejects these). Persistence:
                    // rewind like the other post-trust error paths so a
                    // bare retry re-hits this chunk.
                    if let Err(e) = validate_special_chunk(&ch, chunk_begin, chunk_type) {
                        self.next_chunk_file_pos = chunk_begin;
                        return Err(e);
                    }
                    continue;
                }
            }
        }
    }

    /// Read the 40-byte chunk header at `pos`, skipping and validating the
    /// block headers that the writer interleaves at every block boundary —
    /// both a block header directly at `pos` and one falling inside the
    /// 40-byte span (a chunk header may straddle a block boundary).
    ///
    /// Returns `Ok(None)` on a clean EOF. Otherwise returns
    /// `(header, chunk_begin, data_begin)`: `chunk_begin` is the position of
    /// the header's first byte (after any leading block header) — the value
    /// to use for record positions and `advance_past_chunk` — and
    /// `data_begin` is the position of the first chunk-data byte.
    fn read_chunk_header_at(
        &mut self,
        pos: u64,
    ) -> Result<Option<(ChunkHeader, u64, u64)>, RiegeliError> {
        // Canonicalize: a chunk whose header physically follows a block
        // header is addressed AT the block boundary; the first-header-byte
        // form (boundary + 24) is accepted as an alias of the same chunk.
        let chunk_begin = crate::block_arithmetic::canonical_chunk_address(pos);
        let mut file_pos = chunk_begin;
        let mut bytes = [0u8; 40]; // CHUNK_HEADER_SIZE
        let mut filled: usize = 0;
        // Whether any byte of this chunk's span (including an interleaved
        // block header) has been consumed. Distinguishes a clean EOF (the
        // stream ends exactly where the chunk would begin — C++ ReadRecord
        // returns false with the reader still ok()) from TRUNCATION (the
        // stream ends after part of the chunk was read — C++ sets
        // truncated_ in FailReading and fails with "Truncated
        // Riegeli/records file"). Mapping the partial case to EOF made a
        // file cut off mid-header read back as a complete, shorter file.
        let mut consumed_any = false;
        // A block header read strictly INSIDE the 40-byte header span (the
        // straddle case), kept for the next_chunk cross-check once the
        // chunk header's own claims are hash-validated below.
        let mut interior_block_header: Option<(u64, BlockHeader)> = None;

        while filled < bytes.len() {
            if is_block_boundary(file_pos) {
                self.reader.seek(SeekFrom::Start(file_pos))?;
                let mut bh_bytes = [0u8; 24]; // BLOCK_HEADER_SIZE
                let n = read_full(&mut self.reader, &mut bh_bytes)?;
                if n < bh_bytes.len() {
                    if n == 0 && !consumed_any {
                        return Ok(None); // clean EOF at the chunk's beginning
                    }
                    return Err(RiegeliError::MalformedData(
                        format!(
                            "truncated Riegeli/records file: incomplete chunk at {chunk_begin}"
                        )
                        .into(),
                    ));
                }
                let bh = BlockHeader::from_bytes(bh_bytes);
                if !bh.is_valid() {
                    return Err(RiegeliError::MalformedData(
                        format!("invalid block header hash at file position {file_pos}").into(),
                    ));
                }
                // Cross-validate against the chunk layout (C++
                // ReadChunkHeader): the block header's previous_chunk must
                // point back exactly to this chunk's beginning.
                if file_pos.checked_sub(bh.previous_chunk()) != Some(chunk_begin) {
                    return Err(RiegeliError::MalformedData(
                        format!(
                            "chunk boundary is {chunk_begin} but block header at {file_pos} \
                             implies a different previous chunk boundary"
                        )
                        .into(),
                    ));
                }
                if filled > 0 {
                    interior_block_header = Some((file_pos, bh));
                }
                consumed_any = true;
                file_pos += BLOCK_HEADER_SIZE;
            }
            let until_boundary = BLOCK_SIZE - (file_pos % BLOCK_SIZE);
            let to_read = ((bytes.len() - filled) as u64).min(until_boundary) as usize;
            self.reader.seek(SeekFrom::Start(file_pos))?;
            let n = read_full(&mut self.reader, &mut bytes[filled..filled + to_read])?;
            if n < to_read {
                if n == 0 && !consumed_any {
                    return Ok(None); // clean EOF at the chunk's beginning
                }
                return Err(RiegeliError::MalformedData(
                    format!("truncated Riegeli/records file: incomplete chunk at {chunk_begin}")
                        .into(),
                ));
            }
            consumed_any = true;
            filled += n;
            file_pos += n as u64;
        }

        let ch = ChunkHeader::from_bytes(bytes);

        // Validate header-claimed sizes against the physical stream before
        // any caller does arithmetic, allocation, or overhead walking with
        // them. The header hash only proves integrity, not honesty — anyone
        // authoring a file can hash arbitrary claims, and unchecked claims
        // reach u64 arithmetic (overflow), Vec::with_capacity (allocation
        // bombs), and an O(claim) block-overhead walk. No well-formed file
        // is rejected: chunk data cannot extend past end of file, and the
        // format guarantees a chunk spans at least num_records file bytes.
        // Claims of a hash-invalid header are not checked here — callers
        // report those with their own header-hash errors.
        if ch.is_header_valid() {
            let data_begin = file_pos;
            if ch.data_size() > self.stream_len.saturating_sub(data_begin)
                || ch.num_records() > self.stream_len.saturating_sub(chunk_begin)
            {
                // Re-measure before rejecting: a seek resets EOF, so a file
                // that grew since the last measurement is a supported way
                // to keep reading — the bound must track the growth.
                self.stream_len = self.reader.seek(SeekFrom::End(0))?;
                if ch.data_size() > self.stream_len.saturating_sub(data_begin) {
                    return Err(RiegeliError::MalformedData(format!(
                        "chunk at {chunk_begin} claims {} data bytes with only {} bytes left in the stream",
                        ch.data_size(),
                        self.stream_len.saturating_sub(data_begin)
                    ).into()));
                }
                if ch.num_records() > self.stream_len.saturating_sub(chunk_begin) {
                    return Err(RiegeliError::MalformedData(format!(
                        "chunk at {chunk_begin} claims {} records with only {} bytes left in the stream",
                        ch.num_records(),
                        self.stream_len.saturating_sub(chunk_begin)
                    ).into()));
                }
            }
            // The chunk header straddled a block boundary: both headers
            // have been read, so verify they agree (C++ ReadChunkHeader):
            // the block header's next_chunk must point exactly to this
            // chunk's end. Only meaningful once the chunk header is
            // hash-valid and its claims are stream-bounded (chunk_end does
            // arithmetic on the claims).
            if let Some((block_begin, bh)) = interior_block_header {
                let chunk_end = crate::block_arithmetic::chunk_end(
                    chunk_begin,
                    ch.data_size(),
                    ch.num_records(),
                );
                if block_begin.checked_add(bh.next_chunk()) != Some(chunk_end) {
                    return Err(RiegeliError::MalformedData(
                        format!(
                            "chunk boundary is {chunk_end} but block header at {block_begin} \
                             implies a different next chunk boundary"
                        )
                        .into(),
                    ));
                }
            }
        }

        Ok(Some((ch, chunk_begin, file_pos)))
    }

    /// Read `data_size` bytes of chunk data starting at `data_begin`,
    /// skipping block headers at boundaries. `data_begin` must be the
    /// position of the first data byte (as returned by
    /// `read_chunk_header_at`), which is not `chunk_begin + 40` when the
    /// chunk header straddles a block boundary.
    ///
    /// `chunk_begin` and `chunk_end` describe the chunk being read; every
    /// block header encountered inside the data is cross-validated against
    /// them (C++ ReadChunk): previous_chunk must point back to
    /// `chunk_begin` and next_chunk forward to `chunk_end`.
    fn read_chunk_data(
        &mut self,
        chunk_begin: u64,
        chunk_end: u64,
        data_begin: u64,
        data_size: u64,
    ) -> Result<Vec<u8>, RiegeliError> {
        // data_size is validated against the stream by read_chunk_header_at;
        // the min is defense in depth for any future unvalidated caller.
        let mut result = Vec::with_capacity(data_size.min(self.stream_len) as usize);
        let mut remaining = data_size;
        let mut file_pos = data_begin;

        // Always position explicitly: callers cannot guarantee the reader's
        // physical position (the header read may have re-measured the stream
        // length against a growing file, which seeks to the end).
        self.reader.seek(SeekFrom::Start(file_pos))?;

        while remaining > 0 {
            // Skip block header if at boundary.
            if is_block_boundary(file_pos) {
                let mut bh_bytes = [0u8; 24]; // BLOCK_HEADER_SIZE
                self.reader.seek(SeekFrom::Start(file_pos))?;
                self.reader.read_exact(&mut bh_bytes)?;
                let bh = BlockHeader::from_bytes(bh_bytes);
                if !bh.is_valid() {
                    return Err(RiegeliError::MalformedData(
                        format!(
                        "invalid block header hash at file position {file_pos} (during data read)"
                    )
                        .into(),
                    ));
                }
                // Cross-validate against the chunk layout (C++ ReadChunk):
                // the block header must agree about where this chunk begins
                // and ends, or the walk has desynchronized from the
                // writer's actual layout.
                if file_pos.checked_sub(bh.previous_chunk()) != Some(chunk_begin) {
                    return Err(RiegeliError::MalformedData(
                        format!(
                            "chunk boundary is {chunk_begin} but block header at {file_pos} \
                             implies a different previous chunk boundary"
                        )
                        .into(),
                    ));
                }
                if file_pos.checked_add(bh.next_chunk()) != Some(chunk_end) {
                    return Err(RiegeliError::MalformedData(
                        format!(
                            "chunk boundary is {chunk_end} but block header at {file_pos} \
                             implies a different next chunk boundary"
                        )
                        .into(),
                    ));
                }
                file_pos += BLOCK_HEADER_SIZE;
                // Seek to data position after the block header.
                self.reader.seek(SeekFrom::Start(file_pos))?;
            }

            // How many bytes can we read before hitting the next block boundary?
            let bytes_until_boundary = BLOCK_SIZE - (file_pos % BLOCK_SIZE);
            let to_read = remaining.min(bytes_until_boundary) as usize;

            let old_len = result.len();
            result.resize(old_len + to_read, 0);
            self.reader.read_exact(&mut result[old_len..])?;

            file_pos += to_read as u64;
            remaining -= to_read as u64;
        }

        Ok(result)
    }

    /// Load and decode a chunk at the given file position, returning the decoder.
    ///
    /// Returns `Ok(None)` at EOF.
    fn load_chunk_at(&mut self, file_pos: u64) -> Result<Option<ActiveDecoder>, RiegeliError> {
        let mut file_pos = file_pos;
        loop {
            // Each chunk attempt re-classifies from scratch.
            self.pending_trusted_end = None;
            // Read the chunk header (skipping leading and interleaved block headers).
            let (ch, chunk_begin, data_begin) = match self.read_chunk_header_at(file_pos)? {
                Some(v) => v,
                None => return Ok(None), // EOF
            };

            if !ch.is_header_valid() {
                return Err(RiegeliError::MalformedData(
                    format!("invalid chunk header hash at file position {chunk_begin}").into(),
                ));
            }

            let data_size = ch.data_size();

            // Advance only on success: if decoder construction below fails, the
            // position stays at this chunk so the error is persistent on retry
            // (same convention as load_next_chunk).
            let chunk_end_pos =
                crate::block_arithmetic::chunk_end(chunk_begin, data_size, ch.num_records());

            let chunk_data =
                self.read_chunk_data(chunk_begin, chunk_end_pos, data_begin, data_size)?;

            // Header valid, claims bounded, data present: trustworthy extent
            // for any failure from here on (see load_next_chunk).
            self.pending_trusted_end = Some(chunk_end_pos);

            if !ch.is_data_valid(&chunk_data) {
                return Err(RiegeliError::MalformedData(
                    format!("chunk data hash mismatch at file position {chunk_begin}").into(),
                ));
            }

            self.current_chunk_begin = chunk_begin;
            self.current_record_index = 0;

            match ch.chunk_type() {
                Ok(ChunkType::Simple) => {
                    let chunk = Chunk {
                        header: ch,
                        data: chunk_data,
                    };
                    let decoder = SimpleChunkDecoder::new(chunk)?;
                    self.next_chunk_file_pos = chunk_end_pos;
                    return Ok(Some(ActiveDecoder::Simple(decoder)));
                }
                Ok(ChunkType::Transposed) => {
                    let chunk = Chunk {
                        header: ch,
                        data: chunk_data,
                    };
                    let decoder = TransposeChunkDecoder::new_with_projection(
                        chunk,
                        self.field_projection.as_ref(),
                    )?;
                    self.next_chunk_file_pos = chunk_end_pos;
                    return Ok(Some(ActiveDecoder::Transposed(decoder)));
                }
                other => {
                    match other {
                        // Validate the structural invariants before skipping
                        // (C++ ChunkDecoder::Parse rejects these).
                        Ok(ct) => {
                            if let Err(e) = validate_special_chunk(&ch, chunk_begin, ct) {
                                self.next_chunk_file_pos = chunk_begin;
                                return Err(e);
                            }
                        }
                        // Matching the C++ ChunkDecoder (and load_next_chunk
                        // on the sequential path): an unknown chunk type is
                        // skippable only when it carries no records —
                        // skipping records silently would lose data, so a
                        // seek that lands on such a chunk is an error, not a
                        // silent jump to the next chunk.
                        Err(_) if ch.num_records() == 0 => {}
                        Err(e) => {
                            // Persistence convention: rewind so a bare retry
                            // re-hits this chunk.
                            self.next_chunk_file_pos = chunk_begin;
                            return Err(e);
                        }
                    }
                    // A non-data chunk (signature, metadata, padding) is not
                    // EOF: scan forward to the next data chunk, the same way
                    // seek_numeric does. Conflating the two wedged the reader
                    // at EOF for every read after a seek to such an address.
                    // Iterative on purpose: a long run of tiny padding chunks
                    // must cost O(file size) scanning, not a stack frame per
                    // chunk (a crafted 2 MB padding run overflowed the stack
                    // when this was recursive).
                    self.next_chunk_file_pos = chunk_end_pos;
                    file_pos = chunk_end_pos;
                }
            }
        }
    }

    /// Peek at the chunk header at file_pos without advancing state.
    fn peek_chunk_header(&mut self, file_pos: u64) -> Result<Option<ChunkHeader>, RiegeliError> {
        match self.read_chunk_header_at(file_pos)? {
            None => Ok(None),
            // Hash-invalid headers carry unvalidated claims (the stream-bound
            // check in read_chunk_header_at only covers hash-valid headers,
            // since the other callers report hash failures themselves). A
            // peek must not hand such claims to seek scans or metadata reads
            // — a corrupted header claiming a huge size would drive an
            // O(claim) overhead walk or an oversized read.
            Some((ch, chunk_begin, _)) => {
                if !ch.is_header_valid() {
                    return Err(RiegeliError::MalformedData(
                        format!("invalid chunk header hash at file position {chunk_begin}").into(),
                    ));
                }
                Ok(Some(ch))
            }
        }
    }
}

/// Validate the structural invariants of non-data chunk types, matching
/// the C++ `ChunkDecoder::Parse`:
///
/// - `FileSignature`: data size, record count, and decoded data size are 0.
/// - `FileMetadata`: record count is 0.
/// - `Padding`: record count and decoded data size are 0.
///
/// Skipping such chunks unchecked diverges from the reference's
/// accept/reject decisions, and a nonzero `num_records` additionally skews
/// the chunk-end arithmetic (it participates in the zero-padding term), so
/// the reader would resume at a different position than C++.
fn validate_special_chunk(
    ch: &ChunkHeader,
    chunk_begin: u64,
    chunk_type: ChunkType,
) -> Result<(), RiegeliError> {
    match chunk_type {
        ChunkType::FileSignature => {
            if ch.data_size() != 0 || ch.num_records() != 0 || ch.decoded_data_size() != 0 {
                return Err(RiegeliError::MalformedData(
                    format!(
                        "invalid file signature chunk at {chunk_begin}: data size {}, \
                         number of records {}, and decoded data size {} must all be zero",
                        ch.data_size(),
                        ch.num_records(),
                        ch.decoded_data_size()
                    )
                    .into(),
                ));
            }
        }
        ChunkType::FileMetadata => {
            if ch.num_records() != 0 {
                return Err(RiegeliError::MalformedData(
                    format!(
                        "invalid file metadata chunk at {chunk_begin}: \
                         number of records is not zero: {}",
                        ch.num_records()
                    )
                    .into(),
                ));
            }
        }
        ChunkType::Padding => {
            if ch.num_records() != 0 || ch.decoded_data_size() != 0 {
                return Err(RiegeliError::MalformedData(
                    format!(
                        "invalid padding chunk at {chunk_begin}: number of records {} \
                         and decoded data size {} must both be zero",
                        ch.num_records(),
                        ch.decoded_data_size()
                    )
                    .into(),
                ));
            }
        }
        ChunkType::Simple | ChunkType::Transposed => {}
    }
    Ok(())
}

/// Read as many bytes as are available into `buf`, returning the count.
/// Unlike `read_exact`, a short read at end of stream is reported through
/// the count instead of an `UnexpectedEof` error, so callers can tell a
/// clean EOF (0 bytes) from a truncated structure (partial bytes).
fn read_full<R: Read>(reader: &mut R, buf: &mut [u8]) -> std::io::Result<usize> {
    let mut filled = 0;
    while filled < buf.len() {
        match reader.read(&mut buf[filled..]) {
            Ok(0) => break,
            Ok(n) => filled += n,
            Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
            Err(e) => return Err(e),
        }
    }
    Ok(filled)
}

/// Return the next block boundary strictly after `pos`.
fn next_block_boundary(pos: u64) -> u64 {
    if is_block_boundary(pos) {
        pos
    } else {
        round_down_to_block_boundary(pos) + BLOCK_SIZE
    }
}

#[cfg(test)]
mod tests {
    use std::cell::RefCell;
    use std::io::Cursor;
    use std::rc::Rc;

    use super::*;
    use crate::compression::CompressionType;
    use crate::record_writer::{RecordWriter, WriterOptions};

    /// Write records to a Vec<u8> and return the bytes.
    fn write_records(records: &[&[u8]], opts: WriterOptions) -> Vec<u8> {
        let mut buf = Cursor::new(Vec::<u8>::new());
        {
            let mut w = RecordWriter::new(&mut buf, opts).expect("new ok");
            for rec in records {
                w.write_record(rec).expect("write ok");
            }
            w.flush().expect("flush ok");
        }
        buf.into_inner()
    }

    // -------------------------------------------------------------------------
    // read back a RecordWriter-written file
    // -------------------------------------------------------------------------
    #[test]
    fn roundtrip_basic() {
        let records: &[&[u8]] = &[b"hello", b"world", b"riegeli"];
        let data = write_records(records, WriterOptions::new());
        let cursor = Cursor::new(data);
        let mut reader = RecordReader::new(cursor, ReaderOptions::new()).expect("reader new ok");

        let mut got = Vec::new();
        while let Some(rec) = reader.read_record().expect("read ok") {
            got.push(rec);
        }
        assert_eq!(got.len(), records.len());
        for (i, (got, expected)) in got.iter().zip(records.iter()).enumerate() {
            assert_eq!(got.as_slice(), *expected, "record {i} mismatch");
        }
    }

    // -------------------------------------------------------------------------
    // 100 records
    // -------------------------------------------------------------------------
    #[test]
    fn roundtrip_100_records() {
        let record_data: Vec<u8> = (0..100u8).collect();
        let records: Vec<&[u8]> = (0..100).map(|_| record_data.as_slice()).collect();
        let data = write_records(&records, WriterOptions::new());
        let cursor = Cursor::new(data);
        let mut reader = RecordReader::new(cursor, ReaderOptions::new()).expect("reader new ok");

        let mut count = 0usize;
        while let Some(rec) = reader.read_record().expect("read ok") {
            assert_eq!(rec, record_data, "record {count} mismatch");
            count += 1;
        }
        assert_eq!(count, 100);
    }

    // -------------------------------------------------------------------------
    // pos() at start
    // -------------------------------------------------------------------------
    #[test]
    fn pos_at_start() {
        let data = write_records(&[b"test"], WriterOptions::new());
        let cursor = Cursor::new(data);
        let reader = RecordReader::new(cursor, ReaderOptions::new()).expect("reader new ok");

        let pos = reader.pos();
        // Matches the C++ reference: initial position is numeric 0 (the
        // earlier draft said 24; the differential harness showed the
        // reference returns 0 and the reference wins).
        assert_eq!(pos.numeric(), 0, "initial position is numeric 0");
        assert_eq!(pos.record_index, 0, "record_index should be 0");
    }

    // -------------------------------------------------------------------------
    // last_pos().numeric() → seek_numeric → same record
    // -------------------------------------------------------------------------
    #[test]
    fn seek_numeric_roundtrip() {
        let records: Vec<Vec<u8>> = (0..10u8).map(|i| vec![i; 50]).collect();
        let record_refs: Vec<&[u8]> = records.iter().map(|r| r.as_slice()).collect();
        let data = write_records(&record_refs, WriterOptions::new().chunk_size(200));
        let data = std::sync::Arc::new(data);

        let cursor = Cursor::new((*data).clone());
        let mut reader = RecordReader::new(cursor, ReaderOptions::new()).expect("reader new ok");

        // Read a few records and verify seek_numeric can re-read them.
        let mut positions = Vec::new();
        let mut read_records_vec = Vec::new();
        while let Some(rec) = reader.read_record().expect("read ok") {
            positions.push(reader.last_pos());
            read_records_vec.push(rec);
        }

        // Now for each position, seek_numeric and re-read.
        for (i, (&pos, expected)) in positions.iter().zip(read_records_vec.iter()).enumerate() {
            let cursor2 = Cursor::new((*data).clone());
            let mut reader2 =
                RecordReader::new(cursor2, ReaderOptions::new()).expect("reader new ok");
            reader2
                .seek_numeric(pos.numeric())
                .expect("seek_numeric ok");
            let rec = reader2
                .read_record()
                .expect("read ok after seek")
                .expect("should have record");
            assert_eq!(&rec, expected, "record {i} mismatch after seek_numeric");
        }
    }

    // -------------------------------------------------------------------------
    // corruption handling
    // -------------------------------------------------------------------------
    #[test]
    fn corruption_no_recovery() {
        let records: &[&[u8]] = &[b"before", b"during", b"after"];
        let mut data = write_records(records, WriterOptions::new().chunk_size(10));

        // Corrupt the second chunk's data (skip header at 0, sig chunk at 24..64, first data chunk starts at 64).
        // The first data chunk header is at 64 (40 bytes), data starts at 104.
        // Let's find the second data chunk by reading the first chunk's size.
        // For simplicity, just corrupt some bytes in the middle of the file.
        let mid = data.len() / 2;
        // Flip some bytes in the middle, making sure we're not in a block header.
        for i in mid..mid + 4 {
            if i < data.len() {
                data[i] ^= 0xFF;
            }
        }

        let cursor = Cursor::new(data);
        let mut reader = RecordReader::new(cursor, ReaderOptions::new()).expect("reader new ok");

        // Without recovery, should return Err at some point.
        let mut found_err = false;
        for _ in 0..10 {
            match reader.read_record() {
                Err(_) => {
                    found_err = true;
                    break;
                }
                Ok(None) => break,
                Ok(Some(_)) => {}
            }
        }
        assert!(
            found_err,
            "expected an error when reading corrupted file without recovery"
        );
    }

    #[test]
    fn corruption_with_recovery() {
        // Write many records spread across multiple chunks.
        let records: Vec<Vec<u8>> = (0..50u8).map(|i| vec![i; 100]).collect();
        let record_refs: Vec<&[u8]> = records.iter().map(|r| r.as_slice()).collect();
        let mut data = write_records(&record_refs, WriterOptions::new().chunk_size(200));

        // Corrupt the middle of the file (past the first block of data, so
        // there are records before and after the corruption).
        // Find a good spot: skip initial headers and corrupt something in the data area.
        // We need to corrupt inside a chunk (not a block header) to trigger recovery.
        let mid = (data.len() / 2).max(100);
        // Make sure we're not corrupting a block header position.
        let mid = if mid % 65536 < 24 { mid + 24 } else { mid };
        if mid + 8 < data.len() {
            for b in &mut data[mid..mid + 8] {
                *b ^= 0xFF;
            }
        }

        let recovered_positions: Rc<RefCell<Vec<u64>>> = Rc::new(RefCell::new(Vec::new()));
        let recovered_clone = Rc::clone(&recovered_positions);

        let cursor = Cursor::new(data);
        let opts = ReaderOptions::new().recovery(move |region| {
            recovered_clone.borrow_mut().push(region.begin());
            true
        });
        let mut reader = RecordReader::new(cursor, opts).expect("reader new ok");

        // Read all records (with recovery, should not return Err).
        let mut all_records = Vec::new();
        loop {
            match reader.read_record() {
                Ok(Some(rec)) => all_records.push(rec),
                Ok(None) => break,
                Err(e) => panic!("unexpected error with recovery: {e}"),
            }
        }

        // Recovery should have been triggered (some records recovered or skipped).
        // We should have read at least some records.
        assert!(
            !all_records.is_empty(),
            "should have read some records with recovery"
        );
        // Recovery callback should have been called at least once.
        assert!(
            !recovered_positions.borrow().is_empty(),
            "recovery callback should have been called"
        );
    }

    // -------------------------------------------------------------------------
    // seek_numeric to middle of chunk
    // -------------------------------------------------------------------------
    #[test]
    fn seek_numeric_mid_chunk() {
        // Write records into a single chunk (large chunk_size so all go in one).
        let records: Vec<Vec<u8>> = (0..10u8).map(|i| vec![i; 20]).collect();
        let record_refs: Vec<&[u8]> = records.iter().map(|r| r.as_slice()).collect();
        let data = write_records(&record_refs, WriterOptions::new().chunk_size(1 << 20));

        // All 10 records are in one chunk starting at 64.
        // chunk_begin = 64, record_index 0..9.
        // numeric for record 5 = 64 + 5 = 69.
        // seek_numeric(67) should resolve to the record at chunk_begin=64, record_index=3 (67-64=3).
        // That is records[3] = vec![3; 20].
        let cursor = Cursor::new(data);
        let mut reader = RecordReader::new(cursor, ReaderOptions::new()).expect("reader new ok");

        reader.seek_numeric(64 + 3).expect("seek_numeric ok");
        let rec = reader
            .read_record()
            .expect("read ok")
            .expect("should have record");
        assert_eq!(rec, vec![3u8; 20], "expected record[3]");
    }

    // -------------------------------------------------------------------------
    // read_metadata returns None
    // -------------------------------------------------------------------------
    #[test]
    fn read_metadata_returns_none() {
        let data = write_records(&[b"x"], WriterOptions::new());
        let cursor = Cursor::new(data);
        let mut reader = RecordReader::new(cursor, ReaderOptions::new()).expect("reader new ok");
        let meta = reader.read_metadata().expect("read_metadata ok");
        assert!(meta.is_none(), "expected None from read_metadata");
    }

    // -------------------------------------------------------------------------
    // EOF returns Ok(None), then Ok(None) again
    // -------------------------------------------------------------------------
    #[test]
    fn eof_returns_none_repeatedly() {
        let data = write_records(&[b"only"], WriterOptions::new());
        let cursor = Cursor::new(data);
        let mut reader = RecordReader::new(cursor, ReaderOptions::new()).expect("reader new ok");

        // Read the one record.
        let rec = reader
            .read_record()
            .expect("first read ok")
            .expect("should have record");
        assert_eq!(rec, b"only");

        // EOF.
        let r1 = reader.read_record().expect("second read ok");
        assert!(r1.is_none(), "expected None at EOF");

        // EOF again.
        let r2 = reader.read_record().expect("third read ok");
        assert!(r2.is_none(), "expected None again");
    }

    // -------------------------------------------------------------------------
    // Multi-block roundtrip
    // -------------------------------------------------------------------------
    #[test]
    fn roundtrip_multi_block() {
        // Write enough data to span multiple blocks.
        let record: Vec<u8> = vec![0xAB; 1000];
        let records: Vec<&[u8]> = (0..100).map(|_| record.as_slice()).collect();
        let data = write_records(&records, WriterOptions::new().chunk_size(4096));

        let cursor = Cursor::new(data);
        let mut reader = RecordReader::new(cursor, ReaderOptions::new()).expect("reader new ok");

        let mut count = 0usize;
        while let Some(rec) = reader.read_record().expect("read ok") {
            assert_eq!(rec, record, "record {count} mismatch");
            count += 1;
        }
        assert_eq!(count, 100, "should read exactly 100 records");
    }

    // -------------------------------------------------------------------------
    // Interleaved simple and transposed chunks
    // -------------------------------------------------------------------------
    #[test]
    fn interleaved_simple_and_transposed() {
        // Build a file by hand: signature + simple chunk + transposed chunk.
        // We use the record_writer to write a normal file (simple chunks only),
        // then manually splice in a transposed chunk.
        //
        // For simplicity, we write a file with simple chunk records, then create
        // a separate transposed chunk and concatenate them into a valid file.
        use crate::block_header::BlockHeader;
        use crate::chunk_header::{ChunkHeader, ChunkType};
        use crate::simple_chunk::SimpleChunkEncoder;
        use crate::transpose::internal::message_id;
        use crate::varint::{encode_u32, encode_u64};

        // Build the file manually:
        // [BlockHeader at 0] [FileSignature ChunkHeader at 24] [Simple ChunkHeader] [Simple Data] [Transposed ChunkHeader] [Transposed Data]

        let mut file_data: Vec<u8> = Vec::new();

        // Block header at offset 0.
        // We'll fill it in later once we know sizes.
        let bh_placeholder = [0u8; 24];
        file_data.extend_from_slice(&bh_placeholder);

        // File signature chunk.
        let sig_header = ChunkHeader::from_parts(&[], ChunkType::FileSignature, 0, 0);
        file_data.extend_from_slice(&sig_header.to_bytes());

        // Simple chunk with 2 records.
        let mut simple_enc = SimpleChunkEncoder::new();
        simple_enc.add_record(b"simple_one");
        simple_enc.add_record(b"simple_two");
        let simple_chunk = simple_enc.encode().unwrap();
        file_data.extend_from_slice(&simple_chunk.header.to_bytes());
        file_data.extend_from_slice(&simple_chunk.data);

        // Transposed chunk with 1 nonproto record "transposed".
        let nonproto_data = b"transposed".to_vec();
        let mut nonproto_lengths = Vec::new();
        nonproto_lengths.extend_from_slice(&encode_u32(10));

        // Build transpose header.
        let mut header_bytes: Vec<u8> = Vec::new();
        header_bytes.extend_from_slice(&encode_u32(1)); // num_buckets
        header_bytes.extend_from_slice(&encode_u32(2)); // num_buffers
        let total_buf: usize = nonproto_data.len() + nonproto_lengths.len();
        header_bytes.extend_from_slice(&encode_u64(total_buf as u64)); // bucket compressed size
        header_bytes.extend_from_slice(&encode_u64(nonproto_data.len() as u64)); // buf 0 size
        header_bytes.extend_from_slice(&encode_u64(nonproto_lengths.len() as u64)); // buf 1 size
        header_bytes.extend_from_slice(&encode_u32(1)); // num_states
        header_bytes.extend_from_slice(&encode_u32(message_id::NON_PROTO)); // tag for state 0
        header_bytes.extend_from_slice(&encode_u32(0)); // next_node for state 0
        // NonProto reads buffer_index:
        header_bytes.extend_from_slice(&encode_u32(0)); // buffer_index = 0 (nonproto data)
        header_bytes.extend_from_slice(&encode_u32(0)); // first_node

        let mut trans_data: Vec<u8> = Vec::new();
        trans_data.push(0x00); // CompressionType::None
        trans_data.extend_from_slice(&encode_u64(header_bytes.len() as u64));
        trans_data.extend_from_slice(&header_bytes);
        trans_data.extend_from_slice(&nonproto_data);
        trans_data.extend_from_slice(&nonproto_lengths);
        // no transitions

        let trans_header = ChunkHeader::from_parts(&trans_data, ChunkType::Transposed, 1, 10);

        file_data.extend_from_slice(&trans_header.to_bytes());
        file_data.extend_from_slice(&trans_data);

        // Fix the block header.
        // next_chunk = distance from 0 to end of signature chunk = 64.
        // previous_chunk = 0.
        let bh = BlockHeader::from_parts(0, 64);
        let bh_bytes = bh.to_bytes();
        file_data[..24].copy_from_slice(&bh_bytes);

        // Read all records.
        let cursor = Cursor::new(file_data);
        let mut reader = RecordReader::new(cursor, ReaderOptions::new()).expect("reader new ok");

        let mut got = Vec::new();
        while let Some(rec) = reader.read_record().expect("read ok") {
            got.push(rec);
        }

        assert_eq!(got.len(), 3, "should have 3 records total");
        assert_eq!(got[0], b"simple_one");
        assert_eq!(got[1], b"simple_two");
        assert_eq!(got[2], b"transposed");
    }

    // -------------------------------------------------------------------------
    // Brotli roundtrip (when feature enabled)
    // -------------------------------------------------------------------------
    #[test]
    #[cfg(feature = "brotli")]
    fn roundtrip_brotli() {
        let records: &[&[u8]] = &[b"compressed1", b"compressed2", b"compressed3"];
        let data = write_records(
            records,
            WriterOptions::new().compression(CompressionType::Brotli),
        );
        let cursor = Cursor::new(data);
        let mut reader = RecordReader::new(cursor, ReaderOptions::new()).expect("reader new ok");

        let mut got = Vec::new();
        while let Some(rec) = reader.read_record().expect("read ok") {
            got.push(rec);
        }
        assert_eq!(got.len(), records.len());
        for (i, (got, expected)) in got.iter().zip(records.iter()).enumerate() {
            assert_eq!(got.as_slice(), *expected, "brotli record {i} mismatch");
        }
    }

    // -------------------------------------------------------------------------
    // Unknown chunk types must be skipped, not re-read forever
    // -------------------------------------------------------------------------

    /// A well-formed 40-byte chunk header whose type byte is not any known
    /// `ChunkType` discriminant, with zero data bytes. Both hashes are valid,
    /// so only the type is unrecognized — the forward-compatibility case.
    fn unknown_type_chunk() -> Vec<u8> {
        unknown_type_chunk_with_records(0)
    }

    /// A 40-byte Simple-chunk header with valid hashes but hostile claimed
    /// sizes, and no data bytes. The hash proves integrity, not honesty —
    /// these claims must be rejected against the physical stream.
    fn hostile_simple_chunk(data_size: u64, num_records: u64) -> Vec<u8> {
        let data_hash = crate::hash::highway_hash_64(&[]);
        let chunk_type_and_num_records: u64 = (num_records << 8) | (ChunkType::Simple as u8 as u64);
        let decoded_data_size: u64 = 0;

        let mut body = [0u8; 32];
        body[0..8].copy_from_slice(&data_size.to_le_bytes());
        body[8..16].copy_from_slice(&data_hash.to_le_bytes());
        body[16..24].copy_from_slice(&chunk_type_and_num_records.to_le_bytes());
        body[24..32].copy_from_slice(&decoded_data_size.to_le_bytes());
        let header_hash = crate::hash::highway_hash_64(&body);

        let mut out = Vec::with_capacity(40);
        out.extend_from_slice(&header_hash.to_le_bytes());
        out.extend_from_slice(&body);
        out
    }

    fn unknown_type_chunk_with_records(num_records: u64) -> Vec<u8> {
        let data_size: u64 = 0;
        let data_hash = crate::hash::highway_hash_64(&[]);
        let chunk_type_and_num_records: u64 = (num_records << 8) | b'z' as u64; // 'z' is not a known type
        let decoded_data_size: u64 = 0;

        let mut body = [0u8; 32];
        body[0..8].copy_from_slice(&data_size.to_le_bytes());
        body[8..16].copy_from_slice(&data_hash.to_le_bytes());
        body[16..24].copy_from_slice(&chunk_type_and_num_records.to_le_bytes());
        body[24..32].copy_from_slice(&decoded_data_size.to_le_bytes());
        let header_hash = crate::hash::highway_hash_64(&body);

        let mut out = Vec::with_capacity(40);
        out.extend_from_slice(&header_hash.to_le_bytes());
        out.extend_from_slice(&body);
        out
    }

    #[test]
    fn unknown_chunk_type_at_end_is_skipped() {
        let mut data = write_records(&[b"only"], WriterOptions::new());
        data.extend_from_slice(&unknown_type_chunk());

        let mut reader =
            RecordReader::new(Cursor::new(data), ReaderOptions::new()).expect("reader new ok");
        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(&b"only"[..])
        );
        // Before the fix this call never returned: load_next_chunk() hit the
        // unknown chunk and retried the same file position forever.
        assert_eq!(reader.read_record().expect("read ok"), None);
    }

    // -------------------------------------------------------------------------
    // Chunk headers that straddle a 64 KiB block boundary
    // -------------------------------------------------------------------------

    /// Write `n` single-record chunks of `rec_size` incompressible-layout
    /// bytes (flush per record, no compression) and return the file bytes.
    /// With rec_size around 16 KiB the fourth chunk's header lands near the
    /// first 64 KiB block boundary.
    fn write_chunks_past_first_block(rec_size: usize, n: usize) -> Vec<u8> {
        let mut buf = Cursor::new(Vec::<u8>::new());
        {
            let mut w = RecordWriter::new(
                &mut buf,
                WriterOptions::new().compression(CompressionType::None),
            )
            .expect("writer new ok");
            for i in 0..n {
                let rec = vec![(i % 251) as u8; rec_size];
                w.write_record(&rec).expect("write ok");
                w.flush().expect("flush ok");
            }
        }
        buf.into_inner()
    }

    /// Returns true if any chunk in the file has a header straddling a block
    /// boundary (header begins within 40 bytes below a 64 KiB multiple).
    ///
    /// Detected by direct inspection of the raw block-header bytes the
    /// writer emitted — deliberately NOT via the reader under test, whose
    /// position bookkeeping is part of what the straddle tests exercise.
    /// The block header at a boundary stores the distance back to the start
    /// of the chunk in progress there; a 40-byte chunk header straddles the
    /// boundary iff that distance is in (0, 40).
    fn has_straddling_chunk_header(data: &[u8]) -> bool {
        let block = BLOCK_SIZE as usize;
        let mut boundary = block;
        let mut straddles = false;
        while boundary + BLOCK_HEADER_SIZE as usize <= data.len() {
            let prev = u64::from_le_bytes(data[boundary + 8..boundary + 16].try_into().unwrap());
            if 0 < prev && prev < CHUNK_HEADER_SIZE {
                straddles = true;
            }
            boundary += block;
        }
        straddles
    }

    /// The writer interleaves a 24-byte block header inside the 40-byte chunk
    /// header when a chunk begins within 40 bytes of a block boundary. Before
    /// the fix, every read path fetched the header with one contiguous
    /// read_exact and failed with "invalid chunk header hash" on such files.
    #[test]
    fn chunk_header_straddling_block_boundary_roundtrip() {
        let n = 6;
        let mut exercised_straddle = false;
        // Sweep the record size so the fourth chunk's header walks across the
        // 64 KiB boundary; the exact straddling sizes shift if writer overhead
        // changes, which is why this is a sweep and not a single size.
        for rec_size in 16300..16340usize {
            let data = write_chunks_past_first_block(rec_size, n);
            exercised_straddle |= has_straddling_chunk_header(&data);

            let mut reader =
                RecordReader::new(Cursor::new(data), ReaderOptions::new()).expect("reader new ok");
            let mut count = 0;
            while let Some(rec) = reader.read_record().expect("read ok") {
                assert_eq!(rec.len(), rec_size, "rec_size={rec_size} record {count}");
                // The fill byte identifies the record, so a read that comes
                // back the right length but from the wrong place still fails.
                let fill = (count % 251) as u8;
                assert!(
                    rec.iter().all(|&b| b == fill),
                    "rec_size={rec_size} record {count}: content mismatch"
                );
                count += 1;
            }
            assert_eq!(count, n, "rec_size={rec_size}: wrong record count");
        }
        assert!(
            exercised_straddle,
            "sweep never produced a straddling chunk header; widen the range"
        );
    }

    /// size() and seek() walk chunk headers with their own scan loops; they
    /// must handle straddling headers too.
    #[test]
    fn chunk_header_straddling_block_boundary_size_and_seek() {
        let n = 6;
        // Find a straddling layout within the sweep window.
        let data = (16300..16340usize)
            .map(|rec_size| write_chunks_past_first_block(rec_size, n))
            .find(|data| has_straddling_chunk_header(data))
            .expect("no straddling layout found in sweep; widen the range");

        let mut reader =
            RecordReader::new(Cursor::new(data), ReaderOptions::new()).expect("reader new ok");
        assert_eq!(reader.size().expect("size ok"), n as u64);

        // Collect record positions, then seek back to each and re-read.
        let mut positions = Vec::new();
        while reader.read_record().expect("read ok").is_some() {
            positions.push(reader.last_pos());
        }
        assert_eq!(positions.len(), n);
        for (i, pos) in positions.into_iter().enumerate() {
            reader.seek(pos).expect("seek ok");
            let rec = reader
                .read_record()
                .expect("read after seek ok")
                .unwrap_or_else(|| panic!("record {i} missing after seek"));
            assert_eq!(rec[0], (i % 251) as u8, "record {i} content after seek");
        }
    }

    /// Block headers carry previous_chunk/next_chunk pointers that the C++
    /// reader cross-validates against the actual chunk layout whenever a
    /// chunk spans a block boundary ("block header ... implies a different
    /// previous/next chunk boundary"). Hash-valid but layout-inconsistent
    /// pointers (a spliced file, a buggy writer) must be rejected like the
    /// reference does — only checking the hash let the reader keep parsing
    /// from a desynchronized position.
    #[test]
    fn inconsistent_block_header_pointers_are_rejected() {
        let n = 6;
        let data = write_chunks_past_first_block(16320, n);
        let b = BLOCK_SIZE as usize;
        assert!(data.len() > b + 24, "file must span the first boundary");
        let prev = u64::from_le_bytes(data[b + 8..b + 16].try_into().unwrap());
        let next = u64::from_le_bytes(data[b + 16..b + 24].try_into().unwrap());

        let read_all = |data: Vec<u8>| -> Result<usize, RiegeliError> {
            let mut reader = RecordReader::new(Cursor::new(data), ReaderOptions::new())?;
            let mut count = 0;
            while reader.read_record()?.is_some() {
                count += 1;
            }
            Ok(count)
        };

        // Sanity: the untampered file reads fully.
        assert_eq!(read_all(data.clone()).expect("clean file reads"), n);

        // Tamper each pointer; from_parts recomputes the hash, so only the
        // layout cross-check can catch the inconsistency.
        for (label, bad_prev, bad_next) in [
            ("next_chunk", prev, next + 17),
            ("previous_chunk", prev + 17, next),
        ] {
            let mut bad = data.clone();
            bad[b..b + 24].copy_from_slice(&BlockHeader::from_parts(bad_prev, bad_next).to_bytes());
            let err = read_all(bad).expect_err(&format!(
                "inconsistent {label} pointer must be rejected, not silently accepted"
            ));
            assert!(
                err.to_string().contains("chunk boundary"),
                "{label}: unexpected error: {err}"
            );
        }
    }

    #[test]
    fn unknown_chunk_type_mid_stream_is_skipped() {
        // Build two single-record files and splice an unknown chunk between
        // the first file and the second file's record chunk. Layout of each
        // writer output: block header (24) | signature chunk (40) | record chunk.
        let first = write_records(&[b"first"], WriterOptions::new());
        let second = write_records(&[b"second"], WriterOptions::new());

        let mut data = first;
        data.extend_from_slice(&unknown_type_chunk());
        data.extend_from_slice(&second[64..]); // strip block header + signature chunk

        let mut reader =
            RecordReader::new(Cursor::new(data), ReaderOptions::new()).expect("reader new ok");
        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(&b"first"[..])
        );
        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(&b"second"[..])
        );
        assert_eq!(reader.read_record().expect("read ok"), None);
    }

    /// Matching the C++ ChunkDecoder: an unknown chunk type that claims to
    /// carry records cannot be skipped — its records would be silently lost —
    /// so it is an error rather than a forward-compatibility skip.
    #[test]
    fn unknown_chunk_type_with_records_is_an_error() {
        let mut data = write_records(&[b"only"], WriterOptions::new());
        data.extend_from_slice(&unknown_type_chunk_with_records(3));

        let mut reader =
            RecordReader::new(Cursor::new(data), ReaderOptions::new()).expect("reader new ok");
        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(&b"only"[..])
        );
        let err = reader
            .read_record()
            .expect_err("unknown chunk with records must error");
        assert!(
            err.to_string().contains("unknown chunk type")
                || err.to_string().contains("chunk type"),
            "unexpected error: {err}"
        );
    }

    /// The unknown-chunk-with-records error must be persistent: a caller
    /// that ignores the error and calls `read_record` again must hit the
    /// same error, not silently resume at the next chunk — that would skip
    /// the unknown chunk's claimed records after all, defeating the guard.
    /// Layout: [chunk "first"] [unknown type, num_records=3] [chunk "second"].
    #[test]
    fn unknown_chunk_type_error_is_persistent_and_does_not_skip_to_next_chunk() {
        // chunk_size(1) forces one chunk per record, so the two-record file
        // is [block hdr][signature][chunk "first"][chunk "second"] and the
        // one-record file is the same minus the last chunk. Their shared
        // prefix lets us splice an unknown chunk between the two chunks;
        // chunks are position-independent below the first block boundary.
        let one = write_records(&[b"first"], WriterOptions::new().chunk_size(1));
        let two = write_records(&[b"first", b"second"], WriterOptions::new().chunk_size(1));
        assert_eq!(
            &two[..one.len()],
            &one[..],
            "one-record file must be a prefix"
        );
        let second_chunk = &two[one.len()..];

        let mut data = one.clone();
        data.extend_from_slice(&unknown_type_chunk_with_records(3));
        data.extend_from_slice(second_chunk);
        assert!(
            data.len() < 65536,
            "test assumes no block boundary is crossed"
        );

        let mut reader =
            RecordReader::new(Cursor::new(data), ReaderOptions::new()).expect("reader new ok");
        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(&b"first"[..])
        );

        // First attempt errors on the unknown chunk.
        let err = reader
            .read_record()
            .expect_err("unknown chunk with records must error");
        assert!(
            err.to_string().contains("chunk type"),
            "unexpected error: {err}"
        );

        // Retries must keep erroring on the same chunk; "second" must never
        // surface past the guard.
        for attempt in 0..3 {
            match reader.read_record() {
                Err(e) => assert!(
                    e.to_string().contains("chunk type"),
                    "attempt {attempt}: unexpected error: {e}"
                ),
                Ok(rec) => panic!(
                    "attempt {attempt}: error was not persistent; got {:?}",
                    rec.as_deref()
                        .map(|r| String::from_utf8_lossy(r).into_owned())
                ),
            }
        }
    }

    /// A hash-valid chunk header with an arbitrary type/claims, followed by
    /// `data` bytes (whose hash is stored in the header).
    fn special_chunk(
        chunk_type: ChunkType,
        data: &[u8],
        num_records: u64,
        decoded_data_size: u64,
    ) -> Vec<u8> {
        let mut out = ChunkHeader::from_parts(data, chunk_type, num_records, decoded_data_size)
            .to_bytes()
            .to_vec();
        out.extend_from_slice(data);
        out
    }

    /// Non-data chunks must satisfy the structural invariants the C++
    /// ChunkDecoder::Parse enforces: a file signature chunk has zero
    /// data/records/decoded size, a metadata chunk has zero records, a
    /// padding chunk has zero records and decoded size. Skipping them
    /// unchecked diverged from the reference's accept/reject decisions,
    /// and a hostile padding num_records skewed the chunk-end arithmetic
    /// (it participates in the zero-padding term), desynchronizing the
    /// reader's position from C++'s after recovery.
    #[test]
    fn special_chunks_with_nonzero_claims_are_rejected() {
        for (label, bad_chunk) in [
            (
                "padding with records",
                special_chunk(ChunkType::Padding, &[], 1, 0),
            ),
            (
                "padding with decoded size",
                special_chunk(ChunkType::Padding, &[], 0, 7),
            ),
            (
                "mid-file signature with records",
                special_chunk(ChunkType::FileSignature, &[], 2, 0),
            ),
            (
                "metadata with records",
                special_chunk(ChunkType::FileMetadata, &[], 1, 0),
            ),
        ] {
            let mut data = write_records(&[b"only"], WriterOptions::new());
            data.extend_from_slice(&bad_chunk);

            let mut reader =
                RecordReader::new(Cursor::new(data), ReaderOptions::new()).expect("reader new ok");
            assert_eq!(
                reader.read_record().expect("read ok").as_deref(),
                Some(&b"only"[..]),
                "{label}"
            );
            assert!(
                reader.read_record().is_err(),
                "{label}: structurally invalid special chunk must be rejected, not skipped"
            );
        }
    }

    /// The metadata read path shares the validation: a metadata chunk
    /// claiming records is invalid (C++ ChunkDecoder::Parse).
    #[test]
    fn metadata_chunk_with_records_is_rejected_by_metadata_read() {
        // [bh][sig][metadata chunk claiming 1 record][data chunk "x"]
        let base = write_records(&[b"x"], WriterOptions::new());
        let mut data = base[..64].to_vec();
        data.extend_from_slice(&special_chunk(ChunkType::FileMetadata, b"\x0a\x00", 1, 2));
        data.extend_from_slice(&base[64..]);

        let mut reader =
            RecordReader::new(Cursor::new(data), ReaderOptions::new()).expect("reader new ok");
        assert!(
            reader.read_serialized_metadata().is_err(),
            "metadata chunk claiming records must be rejected"
        );
    }

    /// A file truncated INSIDE a chunk header must surface as corruption,
    /// not as a clean EOF: C++ marks the source truncated as soon as any
    /// byte of a chunk was consumed (FailReading) and fails with
    /// "Truncated Riegeli/records file" at close. Mapping the partial
    /// header to EOF made an interrupted download indistinguishable from a
    /// complete file — silent data loss.
    #[test]
    fn truncation_inside_chunk_header_is_an_error_not_eof() {
        let one = write_records(&[b"a"], WriterOptions::new().chunk_size(1));
        let two = write_records(&[b"a", b"b"], WriterOptions::new().chunk_size(1));
        assert_eq!(&two[..one.len()], &one[..]);

        for kept in [1usize, 20, 39] {
            let data = two[..one.len() + kept].to_vec();
            let mut reader =
                RecordReader::new(Cursor::new(data), ReaderOptions::new()).expect("reader new ok");
            assert_eq!(
                reader.read_record().expect("read ok").as_deref(),
                Some(&b"a"[..]),
                "kept={kept}"
            );
            let err = reader
                .read_record()
                .expect_err("a partial trailing chunk header is truncation, not EOF");
            assert!(
                err.to_string().to_lowercase().contains("truncated"),
                "kept={kept}: unexpected error: {err}"
            );
        }

        // Sanity: the file cut exactly at the chunk boundary is a clean EOF.
        let data = two[..one.len()].to_vec();
        let mut reader =
            RecordReader::new(Cursor::new(data), ReaderOptions::new()).expect("reader new ok");
        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(&b"a"[..])
        );
        assert_eq!(reader.read_record().expect("read ok"), None, "clean EOF");
    }

    /// Same for a partial 24-byte block header at a block boundary.
    #[test]
    fn truncation_inside_block_header_is_an_error_not_eof() {
        let mut data = write_records(&[b"a"], WriterOptions::new().final_padding(BLOCK_SIZE));
        assert_eq!(data.len() as u64, BLOCK_SIZE);
        let second = write_records(&[b"b"], WriterOptions::new());
        data.extend_from_slice(&second[..10]); // 10 of the 24 block-header bytes

        let mut reader =
            RecordReader::new(Cursor::new(data), ReaderOptions::new()).expect("reader new ok");
        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(&b"a"[..])
        );
        let err = reader
            .read_record()
            .expect_err("a partial block header is truncation, not EOF");
        assert!(
            err.to_string().to_lowercase().contains("truncated"),
            "unexpected error: {err}"
        );
    }

    /// Cancel semantics (C++ shape, adopted after the empirical trace):
    /// the region is consumed BEFORE the callback runs — `false` makes the
    /// failing operation return the ORIGINAL error once, but the next
    /// operation continues past the rejected region and the callback is
    /// never re-invoked for it. A LATER corrupt region fires the callback
    /// afresh.
    #[test]
    fn recovery_cancel_reports_once_and_consumes_the_region() {
        // [a][b CORRUPT][c][d CORRUPT][e]
        let mut lens = Vec::new();
        let recs: Vec<&[u8]> = vec![b"a", b"b", b"c", b"d", b"e"];
        for k in 1..=recs.len() {
            lens.push(write_records(&recs[..k], WriterOptions::new().chunk_size(1)).len());
        }
        let mut data = write_records(&recs, WriterOptions::new().chunk_size(1));
        data[lens[1] - 1] ^= 0xFF; // chunk "b" data
        data[lens[3] - 1] ^= 0xFF; // chunk "d" data

        let regions: Rc<RefCell<Vec<crate::SkippedRegion>>> = Rc::new(RefCell::new(Vec::new()));
        let rc = Rc::clone(&regions);
        let opts = ReaderOptions::new().recovery(move |region| {
            rc.borrow_mut().push(region.clone());
            false // cancel every region
        });
        let mut reader = RecordReader::new(Cursor::new(data), opts).expect("reader new ok");

        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(&b"a"[..])
        );
        let e1 = reader
            .read_record()
            .expect_err("cancel returns the original error");
        // The region was consumed: the next read continues past "b" to "c".
        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(&b"c"[..]),
            "rejected region is already skipped; reading continues"
        );
        let e2 = reader
            .read_record()
            .expect_err("the LATER corrupt region errors afresh");
        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(&b"e"[..])
        );
        assert_eq!(reader.read_record().expect("read ok"), None);

        let regions = regions.borrow();
        assert_eq!(regions.len(), 2, "one callback per region — never re-fired");
        assert_eq!(regions[0].begin(), lens[0] as u64, "region 1 = chunk b");
        assert_eq!(regions[0].end(), lens[1] as u64);
        assert_eq!(regions[1].begin(), lens[2] as u64, "region 2 = chunk d");
        assert_eq!(regions[1].end(), lens[3] as u64);
        assert_ne!(e1.to_string(), "", "errors carry messages");
        let _ = e2;
        // The accessor exposes the most recent reported region.
        assert_eq!(reader.last_skipped_region(), Some(&regions[1]));
    }

    /// Coupled region/resync precision: corrupting one chunk's DATA (its
    /// header stays hash-valid, so its extent is trustworthy) must skip
    /// exactly that chunk — the region is [chunk_begin, chunk_end) and the
    /// SIBLING chunk in the same block is recovered. The old boundary skip
    /// threw the sibling away.
    #[test]
    fn recovery_skips_exactly_one_chunk_when_header_is_valid() {
        let one = write_records(&[b"a"], WriterOptions::new().chunk_size(1));
        let three = write_records(&[b"a", b"b", b"c"], WriterOptions::new().chunk_size(1));
        let two = write_records(&[b"a", b"b"], WriterOptions::new().chunk_size(1));
        assert_eq!(&three[..two.len()], &two[..]);

        let mut data = three.clone();
        data[two.len() - 1] ^= 0xFF; // corrupt chunk "b"'s final data byte only
        let chunk_b_begin = one.len() as u64;
        let chunk_c_begin = two.len() as u64;

        let regions: Rc<RefCell<Vec<crate::SkippedRegion>>> = Rc::new(RefCell::new(Vec::new()));
        let rc = Rc::clone(&regions);
        let opts = ReaderOptions::new().recovery(move |region| {
            rc.borrow_mut().push(region.clone());
            true
        });
        let mut reader = RecordReader::new(Cursor::new(data), opts).expect("reader new ok");
        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(&b"a"[..])
        );
        // "b" is skipped; "c" — a sibling in the same block — is recovered.
        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(&b"c"[..]),
            "sibling after the bad chunk must be recovered"
        );
        assert_eq!(reader.read_record().expect("read ok"), None);
        // last_record_is_valid is per-record: "c" came from a valid chunk,
        // so the flag is true again after it returns.
        assert!(reader.last_record_is_valid());

        let regions = regions.borrow();
        assert_eq!(regions.len(), 1);
        assert_eq!(
            regions[0].begin(),
            chunk_b_begin,
            "region begins at the bad chunk"
        );
        assert_eq!(
            regions[0].end(),
            chunk_c_begin,
            "region ends exactly where the next chunk begins (the resync position)"
        );
        assert!(regions[0].message().contains("hash mismatch"));
    }

    /// When the chunk HEADER is hash-invalid, its claims cannot be trusted
    /// to compute an extent — the region ends at the next block boundary.
    #[test]
    fn recovery_resyncs_to_boundary_when_header_is_invalid() {
        let one = write_records(&[b"first"], WriterOptions::new().chunk_size(1));
        let two = write_records(&[b"first", b"second"], WriterOptions::new().chunk_size(1));
        let mut data = two.clone();
        data[one.len()] ^= 0xFF; // corrupt chunk 2's HEADER hash
        let data_len = data.len() as u64;

        let regions: Rc<RefCell<Vec<crate::SkippedRegion>>> = Rc::new(RefCell::new(Vec::new()));
        let rc = Rc::clone(&regions);
        let opts = ReaderOptions::new().recovery(move |region| {
            rc.borrow_mut().push(region.clone());
            true
        });
        let mut reader = RecordReader::new(Cursor::new(data), opts).expect("reader new ok");
        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(&b"first"[..])
        );
        assert_eq!(
            reader.read_record().expect("read ok"),
            None,
            "rest of block skipped"
        );

        let regions = regions.borrow();
        assert_eq!(regions.len(), 1);
        assert_eq!(regions[0].begin(), one.len() as u64);
        // Boundary-class, clamped to the stream length (the file ends well
        // before the next 64 KiB boundary; a region cannot extend past the
        // file — C++ reports EOF-ended regions the same way).
        assert_eq!(
            regions[0].end(),
            data_len,
            "boundary resync clamps to EOF for untrusted claims"
        );
    }

    /// Classification must happen at FAILURE time, not recovery time: a
    /// stream that grows between the two (every stream-length probe is a
    /// growth opportunity) must not reclassify an untrusted failure as
    /// trusted — that mislabeled a readable chunk as a precisely-skipped
    /// region. With failure-time classification the claim failure stays
    /// untrusted and the resync is the boundary, never a chunk_end computed
    /// from claims that were unvalidatable when the error happened.
    #[test]
    fn recovery_classifies_at_failure_time_not_recovery_time() {
        struct GrowOnNthEndSeek {
            full: Vec<u8>,
            visible: usize,
            pos: u64,
            end_seeks: u32,
            grow_at: u32,
        }
        impl std::io::Read for GrowOnNthEndSeek {
            fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
                let avail = &self.full[..self.visible];
                let start = (self.pos as usize).min(avail.len());
                let n = buf.len().min(avail.len() - start);
                buf[..n].copy_from_slice(&avail[start..start + n]);
                self.pos += n as u64;
                Ok(n)
            }
        }
        impl std::io::Seek for GrowOnNthEndSeek {
            fn seek(&mut self, pos: std::io::SeekFrom) -> std::io::Result<u64> {
                match pos {
                    std::io::SeekFrom::Start(p) => self.pos = p,
                    std::io::SeekFrom::End(off) => {
                        self.end_seeks += 1;
                        if self.end_seeks >= self.grow_at {
                            self.visible = self.full.len();
                        }
                        self.pos = (self.visible as i64 + off).max(0) as u64;
                    }
                    std::io::SeekFrom::Current(off) => {
                        self.pos = (self.pos as i64 + off).max(0) as u64;
                    }
                }
                Ok(self.pos)
            }
        }

        let one = write_records(&[b"first"], WriterOptions::new().chunk_size(1));
        let full = write_records(&[b"first", b"second"], WriterOptions::new().chunk_size(1));
        let cut = one.len() + 40; // chunk B header readable, data truncated

        let regions: Rc<RefCell<Vec<crate::SkippedRegion>>> = Rc::new(RefCell::new(Vec::new()));
        let rc = Rc::clone(&regions);
        // Grow generously late so construction-time probes don't trigger it,
        // but any recovery-time re-read (the bug) would.
        for grow_at in [3u32, 4, 5] {
            regions.borrow_mut().clear();
            let rc2 = Rc::clone(&rc);
            let reader_src = GrowOnNthEndSeek {
                full: full.clone(),
                visible: cut,
                pos: 0,
                end_seeks: 0,
                grow_at,
            };
            let opts = ReaderOptions::new().recovery(move |region| {
                rc2.borrow_mut().push(region.clone());
                true
            });
            let mut reader = RecordReader::new(reader_src, opts).expect("reader new ok");
            assert_eq!(
                reader.read_record().expect("read ok").as_deref(),
                Some(&b"first"[..])
            );
            let _ = reader.read_record(); // recovery fires on the truncated chunk

            for region in regions.borrow().iter() {
                assert!(
                    region.end() == BLOCK_SIZE || region.end() <= cut as u64 + BLOCK_SIZE,
                    "grow_at={grow_at}: untrusted failure must resync at a boundary"
                );
                assert_ne!(
                    region.end(),
                    full.len() as u64,
                    "grow_at={grow_at}: region end must not be a chunk_end derived \
                     from claims that were unvalidatable at failure time"
                );
            }
        }
    }

    /// A long run of corrupt-data chunks must produce one callback per
    /// chunk with contiguous, exactly-one-chunk regions (the coupling
    /// invariant universally, not just for a single chunk), terminating at
    /// EOF — no loop, no boundary fallback for trusted failures.
    #[test]
    fn recovery_walks_corrupt_chain_with_contiguous_regions() {
        const N: usize = 200; // corrupt chunks between two good ones
        // Record k..=N+1 prefix files give every chunk span.
        let mut lens = Vec::with_capacity(N + 3);
        let mut recs: Vec<Vec<u8>> = Vec::new();
        for k in 0..(N + 2) {
            recs.push(format!("r{k:03}").into_bytes());
            let refs: Vec<&[u8]> = recs.iter().map(|r| r.as_slice()).collect();
            lens.push(write_records(&refs, WriterOptions::new().chunk_size(1)).len());
        }
        let refs: Vec<&[u8]> = recs.iter().map(|r| r.as_slice()).collect();
        let mut data = write_records(&refs, WriterOptions::new().chunk_size(1));
        assert!(
            data.len() < BLOCK_SIZE as usize,
            "keep it single-block for span math"
        );
        // Corrupt the final data byte of chunks 1..=N (leave first and last).
        for &len in &lens[1..=N] {
            data[len - 1] ^= 0xFF;
        }

        let regions: Rc<RefCell<Vec<crate::SkippedRegion>>> = Rc::new(RefCell::new(Vec::new()));
        let rc = Rc::clone(&regions);
        let opts = ReaderOptions::new().recovery(move |region| {
            rc.borrow_mut().push(region.clone());
            true
        });
        let mut reader = RecordReader::new(Cursor::new(data), opts).expect("reader new ok");
        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(&b"r000"[..])
        );
        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(format!("r{:03}", N + 1).as_bytes()),
            "the good chunk after the corrupt run must be reached"
        );
        assert_eq!(reader.read_record().expect("read ok"), None);

        let regions = regions.borrow();
        assert_eq!(regions.len(), N, "exactly one callback per corrupt chunk");
        for (i, region) in regions.iter().enumerate() {
            let k = i + 1;
            assert_eq!(region.begin(), lens[k - 1] as u64, "chunk {k} begin");
            assert_eq!(region.end(), lens[k] as u64, "chunk {k} end == next begin");
        }
    }

    /// A corrupt 24-byte block-header tail at an exact block boundary: the
    /// reader's raw position (boundary+24) canonicalizes 24 bytes BACK, and
    /// the EOF-clamped region end equals the raw position — forward
    /// progress must be measured from the raw position or the reader spins
    /// forever re-reporting the same region (CWE-835, found in review).
    #[test]
    fn recovery_terminates_on_corrupt_block_header_tail_at_boundary() {
        // Fill block 0 exactly: a padded file of BLOCK_SIZE bytes, then a
        // corrupt 24-byte pseudo block header.
        let mut data = write_records(&[b"a"], WriterOptions::new().final_padding(BLOCK_SIZE));
        assert_eq!(data.len() as u64, BLOCK_SIZE);
        data.extend_from_slice(&[0xFFu8; 24]); // hash-invalid block header
        assert_eq!(data.len() as u64, BLOCK_SIZE + 24);

        let regions: Rc<RefCell<Vec<crate::SkippedRegion>>> = Rc::new(RefCell::new(Vec::new()));
        let rr = Rc::clone(&regions);
        let opts = ReaderOptions::new().recovery(move |r| {
            rr.borrow_mut().push(r.clone());
            true
        });
        let mut reader = RecordReader::new(Cursor::new(data), opts).expect("reader new ok");
        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(&b"a"[..])
        );
        // Must terminate (the old clamp spun forever here).
        assert_eq!(reader.read_record().expect("read ok"), None, "clean EOF");
        let regions = regions.borrow();
        assert_eq!(
            regions.len(),
            1,
            "exactly ONE region — the alias-end double-report is repaired (got {regions:?})"
        );
        assert_eq!(regions[0].begin(), BLOCK_SIZE, "canonical begin");
        assert_eq!(
            regions[0].end(),
            BLOCK_SIZE + 25,
            "minimal progress margin: one byte past the 24-byte corrupt tail"
        );
    }

    /// An untrusted failure resyncs at a block boundary — but chunks do not
    /// generally START at boundaries: the bytes after the boundary's block
    /// header are normally the tail of a chunk spanning it. The block
    /// header's next_chunk pointer exists precisely so a reader landing at
    /// a boundary can find the next chunk start, and C++ Recover follows
    /// it (pos_ += block_header_.next_chunk()). Resuming AT the boundary
    /// instead made every subsequent block parse mid-chunk bytes as a
    /// header, cascade new regions, and silently skip the whole rest of
    /// the file.
    #[test]
    fn recovery_resync_follows_block_header_next_chunk() {
        let n = 6;
        let rec_size = 16320usize;
        // Prefix lengths give every chunk's begin (flush-per-record output
        // is deterministic, so shorter runs are prefixes of longer ones).
        let lens: Vec<u64> = (0..=n)
            .map(|k| write_chunks_past_first_block(rec_size, k).len() as u64)
            .collect();
        let mut data = write_chunks_past_first_block(rec_size, n);
        assert!(
            data.len() as u64 > BLOCK_SIZE,
            "file must span the first block boundary"
        );
        // The chunk containing the first boundary (it does not start there).
        let c = (0..n)
            .find(|&k| lens[k] < BLOCK_SIZE && BLOCK_SIZE < lens[k + 1])
            .expect("some chunk must straddle the boundary");
        assert!(c >= 2, "need a chunk to corrupt before the straddler");

        // Corrupt chunk 1's HEADER hash: an untrusted failure at lens[1].
        data[lens[1] as usize] ^= 0xFF;

        let regions: Rc<RefCell<Vec<crate::SkippedRegion>>> = Rc::new(RefCell::new(Vec::new()));
        let rc = Rc::clone(&regions);
        let opts = ReaderOptions::new().recovery(move |region| {
            rc.borrow_mut().push(region.clone());
            true
        });
        let mut reader = RecordReader::new(Cursor::new(data), opts).expect("reader new ok");

        let mut got = Vec::new();
        while let Some(rec) = reader.read_record().expect("read ok") {
            got.push(rec[0]);
        }

        // Record 0 reads normally; the skip resumes at the chunk AFTER the
        // straddler (boundary + next_chunk), recovering records c+1..n.
        let mut expected = vec![0u8];
        expected.extend((c + 1..n).map(|i| (i % 251) as u8));
        assert_eq!(
            got, expected,
            "records after the boundary resync must be recovered"
        );

        let regions = regions.borrow();
        assert_eq!(regions.len(), 1, "exactly one region, no cascade");
        assert_eq!(
            regions[0].begin(),
            lens[1],
            "region begins at the bad chunk"
        );
        assert_eq!(
            regions[0].end(),
            lens[c + 1],
            "region ends at boundary + next_chunk (the next real chunk start)"
        );
    }

    /// Metadata reads promise not to move the read position — recovery
    /// firing inside one must be report-only. Repositioning here rewound a
    /// mid-stream reader to the region end (replaying records) or skipped
    /// block-0 chunks at the start.
    #[test]
    fn metadata_recovery_does_not_move_the_read_position() {
        // [sig][chunkA@64][padding to 64 KiB][fileB: sig][chunkB "b"]
        let mut data = write_records(&[b"a"], WriterOptions::new().final_padding(BLOCK_SIZE));
        assert_eq!(data.len() as u64, BLOCK_SIZE);
        data.extend_from_slice(&write_records(&[b"b"], WriterOptions::new()));
        data[64] ^= 0xFF; // corrupt chunk A's header (the metadata position)

        let count: Rc<RefCell<u32>> = Rc::new(RefCell::new(0));
        let rc = Rc::clone(&count);
        let opts = ReaderOptions::new().recovery(move |_r| {
            *rc.borrow_mut() += 1;
            true
        });
        let mut reader = RecordReader::new(Cursor::new(data), opts).expect("reader new ok");

        let pos_before = reader.pos();
        assert_eq!(
            reader.read_serialized_metadata().expect("metadata ok"),
            None,
            "skipped region reads as absent metadata"
        );
        assert_eq!(*count.borrow(), 1, "callback reported the region");
        assert_eq!(
            reader.pos(),
            pos_before,
            "metadata read must not move the read position"
        );

        // The record stream is undisturbed: reading proceeds from the start,
        // recovers past the corrupt chunk normally, and reaches "b".
        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(&b"b"[..])
        );
        assert_eq!(*count.borrow(), 2, "read-path recovery fired separately");
    }

    /// Seek with recovery: C++ `Seek` returns the result of the recovery
    /// function — on `true` the reader is positioned past the region.
    #[test]
    fn seek_recovers_past_invalid_region() {
        let one = write_records(&[b"a"], WriterOptions::new().chunk_size(1));
        let three = write_records(&[b"a", b"b", b"c"], WriterOptions::new().chunk_size(1));
        let two = write_records(&[b"a", b"b"], WriterOptions::new().chunk_size(1));
        let mut data = three.clone();
        data[two.len() - 1] ^= 0xFF; // corrupt chunk "b"'s final data byte

        let opts = ReaderOptions::new().recovery(|_region| true);
        let mut reader = RecordReader::new(Cursor::new(data), opts).expect("reader new ok");
        reader
            .seek(RecordPosition::new(one.len() as u64, 0))
            .expect("seek with recovery must succeed");
        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(&b"c"[..]),
            "positioned past the skipped region"
        );
        let _ = two;
    }

    /// Search with recovery skips invalid regions during the scan (C++
    /// contract) and still finds targets beyond them.
    #[test]
    fn search_recovers_past_invalid_region() {
        let one = write_records(&[b"a"], WriterOptions::new().chunk_size(1));
        let three = write_records(&[b"a", b"b", b"c"], WriterOptions::new().chunk_size(1));
        let mut data = three;
        data[one.len()] ^= 0xFF; // corrupt chunk "b"'s HEADER (scan-visible)

        let opts = ReaderOptions::new().recovery(|_region| true);
        let mut reader = RecordReader::new(Cursor::new(data), opts).expect("reader new ok");
        let found = reader.search(|rec| rec.cmp(&b"c"[..])).expect("search ok");
        // "c" sits in the region skipped by the boundary resync? No: the
        // corrupt header forces a boundary skip during the scan, and "c"
        // lives below the boundary too, so the honest outcome is NOT
        // FOUND without error — the scan completed, the region was
        // skipped, and the target was inside it.
        assert!(
            !found,
            "target inside the skipped region is reported absent"
        );

        // A target in an intact chunk before the corruption is still found.
        let mut reader2 = {
            let one = write_records(&[b"a"], WriterOptions::new().chunk_size(1));
            let three = write_records(&[b"a", b"b", b"c"], WriterOptions::new().chunk_size(1));
            let mut data = three;
            data[one.len()] ^= 0xFF;
            RecordReader::new(
                Cursor::new(data),
                ReaderOptions::new().recovery(|_region| true),
            )
            .expect("reader new ok")
        };
        assert!(reader2.search(|rec| rec.cmp(&b"a"[..])).expect("search ok"));
    }

    /// A chunk whose header is valid but whose DATA is corrupt passes the
    /// header-only collection scan, so the binary search can pick it as a
    /// pivot. Probing it under recovery used to fabricate an empty record
    /// for the comparator (`b""` compares Less than every non-empty
    /// target), discarding the left half of the search — and a present
    /// target was reported absent. C++ declares such a probe unordered and
    /// the search shrinks around it without judging fabricated content.
    #[test]
    fn search_does_not_judge_fabricated_records_for_unreadable_pivots() {
        let two = write_records(&[b"a", b"b"], WriterOptions::new().chunk_size(1));
        let three = write_records(&[b"a", b"b", b"c"], WriterOptions::new().chunk_size(1));
        let mut data = three.clone();
        data[two.len() - 1] ^= 0xFF; // corrupt chunk "b"'s DATA (header stays valid)

        let opts = ReaderOptions::new().recovery(|_region| true);
        let mut reader = RecordReader::new(Cursor::new(data), opts).expect("reader new ok");
        let found = reader.search(|rec| rec.cmp(&b"a"[..])).expect("search ok");
        assert!(
            found,
            "target in a readable chunk left of the unreadable pivot must be found"
        );
        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(&b"a"[..])
        );
    }

    /// Seeking to a non-data chunk's address (here: the signature chunk of
    /// a concatenated second file) must scan forward to the next data chunk
    /// — it used to wedge the reader at EOF, hiding all following records.
    #[test]
    fn seek_to_non_data_chunk_scans_forward() {
        let mut data = write_records(&[b"a"], WriterOptions::new().final_padding(BLOCK_SIZE));
        assert_eq!(data.len() as u64, BLOCK_SIZE);
        let second = write_records(&[b"b"], WriterOptions::new());
        data.extend_from_slice(&second);

        let mut reader =
            RecordReader::new(Cursor::new(data), ReaderOptions::new()).expect("reader new ok");
        // The concatenated file's signature chunk is addressed at the block
        // boundary — a non-data chunk.
        reader
            .seek(RecordPosition::new(BLOCK_SIZE, 0))
            .expect("seek ok");
        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(&b"b"[..]),
            "seek to a non-data chunk must land on the next data chunk, not EOF"
        );
    }

    /// A long run of consecutive tiny padding chunks must be scanned
    /// iteratively: the forward scan over non-data chunks used to recurse
    /// once per chunk, and a crafted ~2 MB padding run overflowed the
    /// stack (SIGABRT). With the loop this completes in milliseconds.
    #[test]
    fn seek_across_long_padding_run_does_not_overflow_stack() {
        fn padding_run_file(n: usize) -> Vec<u8> {
            let mut file = write_records(&[], WriterOptions::new());
            assert_eq!(file.len() as u64, BLOCK_HEADER_SIZE + CHUNK_HEADER_SIZE);
            let header =
                crate::chunk_header::ChunkHeader::from_parts(&[], ChunkType::Padding, 0, 0)
                    .to_bytes();
            for _ in 0..n {
                let chunk_begin =
                    crate::block_arithmetic::canonical_chunk_address(file.len() as u64);
                let chunk_end = crate::block_arithmetic::chunk_end(chunk_begin, 0, 0);
                let mut i = 0usize;
                while i < header.len() {
                    let pos = file.len() as u64;
                    if pos.is_multiple_of(BLOCK_SIZE) {
                        let bh = crate::block_header::BlockHeader::from_parts(
                            pos - chunk_begin,
                            chunk_end - pos,
                        );
                        file.extend_from_slice(&bh.to_bytes());
                        continue;
                    }
                    let until = (BLOCK_SIZE - pos % BLOCK_SIZE) as usize;
                    let take = until.min(header.len() - i);
                    file.extend_from_slice(&header[i..i + take]);
                    i += take;
                }
            }
            file
        }

        let data = padding_run_file(50_000);
        let mut reader =
            RecordReader::new(Cursor::new(data), ReaderOptions::new()).expect("reader new ok");
        reader
            .seek(RecordPosition::new(
                BLOCK_HEADER_SIZE + CHUNK_HEADER_SIZE,
                0,
            ))
            .expect("seek across the padding run must not crash");
        assert_eq!(reader.read_record().expect("read ok"), None, "clean EOF");
    }

    /// Zero-record data chunks must not serve as binary-search pivots:
    /// probing one falls through to the NEXT chunk's record, corrupting
    /// pivot decisions (a present target could be reported absent).
    #[test]
    fn search_skips_zero_record_chunks() {
        /// A valid empty Simple chunk: zero records, uncompressed, empty
        /// sizes section.
        fn empty_simple_chunk() -> Vec<u8> {
            let data = [0x00u8, 0x00]; // compression none, sizes_byte_len 0
            let data_hash = crate::hash::highway_hash_64(&data);
            let chunk_type_and_num_records: u64 = ChunkType::Simple as u8 as u64; // 0 records
            let mut body = [0u8; 32];
            body[0..8].copy_from_slice(&(data.len() as u64).to_le_bytes());
            body[8..16].copy_from_slice(&data_hash.to_le_bytes());
            body[16..24].copy_from_slice(&chunk_type_and_num_records.to_le_bytes());
            body[24..32].copy_from_slice(&0u64.to_le_bytes());
            let header_hash = crate::hash::highway_hash_64(&body);
            let mut out = Vec::with_capacity(42);
            out.extend_from_slice(&header_hash.to_le_bytes());
            out.extend_from_slice(&body);
            out.extend_from_slice(&data);
            out
        }

        // Convergence layout: the target lives in the LAST real chunk and a
        // trailing empty chunk becomes the final pivot — probing it at EOF
        // yields an empty record, the comparator steers into the empty
        // chunk, and the within-chunk search reports the target absent.
        let mut data = write_records(&[b"a", b"m"], WriterOptions::new());
        data.extend_from_slice(&empty_simple_chunk());

        let mut reader =
            RecordReader::new(Cursor::new(data), ReaderOptions::new()).expect("reader new ok");
        let found = reader.search(|rec| rec.cmp(&b"m"[..])).expect("search ok");
        assert!(
            found,
            "present record must be found despite an empty pivot chunk"
        );
        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(&b"m"[..])
        );
    }

    /// seek() must not clobber last_pos — it tracks the last successfully
    /// READ record, and seek_back() is documented to return there.
    #[test]
    fn seek_back_returns_to_last_read_record_not_seek_target() {
        let data = write_records(&[b"first", b"second"], WriterOptions::new().chunk_size(1));
        let mut reader =
            RecordReader::new(Cursor::new(data), ReaderOptions::new()).expect("reader new ok");

        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(&b"first"[..])
        );
        let second_pos = {
            // Find the second record's position by reading it, then rewind.
            let p = reader.pos();
            assert_eq!(
                reader.read_record().expect("read ok").as_deref(),
                Some(&b"second"[..])
            );
            let sp = reader.last_pos();
            // Restore: last read should again be "first" for the real test.
            reader.seek(p).expect("seek ok");
            sp
        };

        // Fresh reader: read "first", then seek AWAY to second's position
        // without reading, then seek_back — must land on "first".
        let data2 = write_records(&[b"first", b"second"], WriterOptions::new().chunk_size(1));
        let mut reader2 =
            RecordReader::new(Cursor::new(data2), ReaderOptions::new()).expect("reader new ok");
        assert_eq!(
            reader2.read_record().expect("read ok").as_deref(),
            Some(&b"first"[..])
        );
        reader2.seek(second_pos).expect("seek ok");
        assert!(reader2.seek_back().expect("seek_back ok"));
        assert_eq!(
            reader2.read_record().expect("read ok").as_deref(),
            Some(&b"first"[..]),
            "seek_back must return to the last READ record, not the last seek target"
        );
    }

    /// Decoder-construction failures (malformed chunk interior behind valid
    /// hashes) must be persistent like every other read error: a bare retry
    /// must re-hit the same chunk, not silently resume at the next one.
    /// Layout: [chunk "first"][valid-hash chunk with hostile interior]
    /// [chunk "second"].
    #[test]
    fn decoder_construction_error_is_persistent() {
        // A Simple chunk whose hashes are valid but whose single data byte
        // is an unknown compression type — SimpleChunkDecoder::new fails.
        fn hostile_interior_chunk() -> Vec<u8> {
            let data = [0xFFu8]; // unknown compression byte
            let data_hash = crate::hash::highway_hash_64(&data);
            let chunk_type_and_num_records: u64 = (1u64 << 8) | ChunkType::Simple as u8 as u64;
            let decoded_data_size: u64 = 0;
            let mut body = [0u8; 32];
            body[0..8].copy_from_slice(&(data.len() as u64).to_le_bytes());
            body[8..16].copy_from_slice(&data_hash.to_le_bytes());
            body[16..24].copy_from_slice(&chunk_type_and_num_records.to_le_bytes());
            body[24..32].copy_from_slice(&decoded_data_size.to_le_bytes());
            let header_hash = crate::hash::highway_hash_64(&body);
            let mut out = Vec::with_capacity(41);
            out.extend_from_slice(&header_hash.to_le_bytes());
            out.extend_from_slice(&body);
            out.extend_from_slice(&data);
            out
        }

        let one = write_records(&[b"first"], WriterOptions::new().chunk_size(1));
        let two = write_records(&[b"first", b"second"], WriterOptions::new().chunk_size(1));
        assert_eq!(&two[..one.len()], &one[..]);
        let second_chunk = &two[one.len()..];

        let mut data = one.clone();
        data.extend_from_slice(&hostile_interior_chunk());
        data.extend_from_slice(second_chunk);
        assert!(data.len() < 65536);

        let mut reader =
            RecordReader::new(Cursor::new(data), ReaderOptions::new()).expect("reader new ok");
        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(&b"first"[..])
        );
        assert!(
            reader.read_record().is_err(),
            "construction failure must error"
        );
        for attempt in 0..3 {
            match reader.read_record() {
                Err(_) => {}
                Ok(rec) => panic!(
                    "attempt {attempt}: construction error was not persistent; got {:?}",
                    rec.as_deref()
                        .map(|r| String::from_utf8_lossy(r).into_owned())
                ),
            }
        }
    }

    /// A corrupt chunk header where the metadata chunk would live must
    /// surface as an error from the metadata APIs, not as "no metadata" —
    /// a caller inspecting metadata first must not conclude the file is
    /// clean.
    #[test]
    fn metadata_peek_propagates_corruption() {
        let mut data = write_records(&[b"only"], WriterOptions::new());
        data[64] ^= 0xFF; // corrupt the first chunk header after the signature

        let mut reader =
            RecordReader::new(Cursor::new(data), ReaderOptions::new()).expect("reader new ok");
        assert!(
            reader.read_serialized_metadata().is_err(),
            "corruption at the metadata position must not read as absent metadata"
        );

        // Sanity: an intact file without a metadata chunk still reports None.
        let clean = write_records(&[b"only"], WriterOptions::new());
        let mut reader =
            RecordReader::new(Cursor::new(clean), ReaderOptions::new()).expect("reader new ok");
        assert!(reader.read_serialized_metadata().expect("ok").is_none());
    }

    /// An empty (0-byte) file is a valid riegeli file with zero records:
    /// the C++ reference returns false from ReadRecord with the reader
    /// still ok() (FailReading only marks truncation when some bytes of a
    /// chunk were consumed), and Close() succeeds. Hard-failing in the
    /// constructor broke readers pointed at files a writer has created
    /// but not yet flushed.
    #[test]
    fn empty_file_is_a_valid_file_with_zero_records() {
        let mut reader = RecordReader::new(Cursor::new(Vec::<u8>::new()), ReaderOptions::new())
            .expect("an empty file must open as a valid zero-record file");
        assert_eq!(reader.read_record().expect("read ok"), None);
        assert_eq!(reader.read_record().expect("read ok"), None);
        assert_eq!(reader.size().expect("size ok"), 0);

        // A partial preamble (1..=63 bytes) is an error in both
        // implementations — only the exactly-empty case is valid.
        let partial = write_records(&[b"x"], WriterOptions::new())[..10].to_vec();
        assert!(
            RecordReader::new(Cursor::new(partial), ReaderOptions::new()).is_err(),
            "a truncated preamble is still rejected"
        );
    }

    /// Corruption in the first 64 bytes (leading block header or file
    /// signature) must be recoverable when a recovery callback is set: C++
    /// classifies both failures as Recoverable::kFindChunk (\"missing file
    /// signature\" / block-header hash mismatch), reports one skipped
    /// region, and reads the rest of the file. Hard-failing in new()
    /// silently ignored the documented recovery contract for the most
    /// common real-world corruption location.
    #[test]
    fn corrupt_file_preamble_is_recoverable_with_a_callback() {
        let n = 6;
        let rec_size = 16320usize;
        let lens: Vec<u64> = (0..=n)
            .map(|k| write_chunks_past_first_block(rec_size, k).len() as u64)
            .collect();
        let clean = write_chunks_past_first_block(rec_size, n);
        assert!(clean.len() as u64 > BLOCK_SIZE);
        // The chunk straddling the first boundary; resync lands after it.
        let c = (0..n)
            .find(|&k| lens[k] < BLOCK_SIZE && BLOCK_SIZE < lens[k + 1])
            .expect("some chunk must straddle the boundary");

        // Corrupt (a) the leading block header's hash, (b) the signature
        // chunk header.
        for (label, corrupt_at) in [("block header", 2usize), ("file signature", 30usize)] {
            let mut data = clean.clone();
            data[corrupt_at] ^= 0xFF;

            // Without recovery the constructor still fails.
            assert!(
                RecordReader::new(Cursor::new(data.clone()), ReaderOptions::new()).is_err(),
                "{label}: hard error without a recovery callback"
            );

            let regions: Rc<RefCell<Vec<crate::SkippedRegion>>> = Rc::new(RefCell::new(Vec::new()));
            let rc = Rc::clone(&regions);
            let opts = ReaderOptions::new().recovery(move |region| {
                rc.borrow_mut().push(region.clone());
                true
            });
            let mut reader = RecordReader::new(Cursor::new(data), opts)
                .unwrap_or_else(|e| panic!("{label}: must be recoverable, got {e}"));

            let mut got = Vec::new();
            while let Some(rec) = reader.read_record().expect("read ok") {
                got.push(rec[0]);
            }
            let expected: Vec<u8> = (c + 1..n).map(|i| (i % 251) as u8).collect();
            assert_eq!(
                got, expected,
                "{label}: records beyond the resync must be readable"
            );

            let regions = regions.borrow();
            assert_eq!(regions.len(), 1, "{label}: one region, no cascade");
            assert_eq!(regions[0].begin(), 0, "{label}: region begins at 0");
            assert_eq!(
                regions[0].end(),
                lens[c + 1],
                "{label}: region ends at the post-boundary chunk start"
            );
        }
    }

    /// The signature chunk is a fixed constant; a hash-valid signature
    /// header with nonzero claimed sizes must be rejected by the exact
    /// byte comparison — trusting its data_size used to overflow the
    /// position arithmetic (debug panic) or seek backward through the
    /// i64 cast (release).
    #[test]
    fn hostile_signature_chunk_claims_are_rejected() {
        fn hostile_signature(data_size: u64) -> Vec<u8> {
            let data_hash = crate::hash::highway_hash_64(&[]);
            let chunk_type_and_num_records: u64 = ChunkType::FileSignature as u8 as u64;
            let decoded_data_size: u64 = 0;
            let mut body = [0u8; 32];
            body[0..8].copy_from_slice(&data_size.to_le_bytes());
            body[8..16].copy_from_slice(&data_hash.to_le_bytes());
            body[16..24].copy_from_slice(&chunk_type_and_num_records.to_le_bytes());
            body[24..32].copy_from_slice(&decoded_data_size.to_le_bytes());
            let header_hash = crate::hash::highway_hash_64(&body);
            let mut out = Vec::with_capacity(40);
            out.extend_from_slice(&header_hash.to_le_bytes());
            out.extend_from_slice(&body);
            out
        }

        let valid = write_records(&[b"x"], WriterOptions::new());
        for data_size in [u64::MAX, 1000u64] {
            let mut data = valid[..24].to_vec(); // keep the valid block header
            data.extend_from_slice(&hostile_signature(data_size));

            let err = RecordReader::new(Cursor::new(data), ReaderOptions::new())
                .err()
                .expect("hostile signature chunk must be rejected");
            assert!(
                err.to_string().contains("file signature"),
                "data_size={data_size}: unexpected error: {err}"
            );
        }

        // Sanity: a writer-produced file still opens (its signature IS the
        // canonical constant).
        RecordReader::new(Cursor::new(valid), ReaderOptions::new()).expect("valid file opens");
    }

    /// Seeking to a record_index past the end of a chunk must clamp to the
    /// end of that chunk, matching C++ `ChunkDecoder::SetIndex` ("If
    /// `index > num_records()`, the current index is set to
    /// `num_records()`"): the seek succeeds and the next read continues
    /// with the following chunk. Erroring here mislabeled a valid file as
    /// malformed when a persisted position was applied to a file that was
    /// re-chunked differently.
    #[test]
    fn seek_with_out_of_range_record_index_clamps_to_chunk_end() {
        let data = write_records(&[b"a", b"b"], WriterOptions::new().chunk_size(1));
        let chunk_a_begin = BLOCK_HEADER_SIZE + CHUNK_HEADER_SIZE; // 64

        let mut reader =
            RecordReader::new(Cursor::new(data), ReaderOptions::new()).expect("reader new ok");
        reader
            .seek(RecordPosition::new(chunk_a_begin, 5))
            .expect("seek past the chunk end must clamp, not error");
        assert_eq!(
            reader.pos(),
            RecordPosition::new(chunk_a_begin, 1),
            "position clamps to the chunk's num_records"
        );
        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(&b"b"[..]),
            "reading continues with the next chunk"
        );
    }

    /// size() promises to preserve the read position even when it FAILS:
    /// the error paths used to restore the bookkeeping but not the chunk
    /// decoder, so a caller that treated the error as informational and
    /// kept reading silently skipped the rest of the current chunk (the
    /// next read fell through to the next chunk).
    #[test]
    fn size_error_path_preserves_the_read_position() {
        // [chunk A: a1 a2 a3][chunk B: b][40 bytes of hash-invalid header]
        let file1 = write_records(&[b"a1", b"a2", b"a3"], WriterOptions::new());
        let file2 = write_records(&[b"b"], WriterOptions::new());
        let mut data = file1.clone();
        data.extend_from_slice(&file2[64..]); // strip block header + signature
        data.extend_from_slice(&[0xABu8; 40]); // corrupt trailing chunk header

        let mut reader =
            RecordReader::new(Cursor::new(data), ReaderOptions::new()).expect("reader new ok");
        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(&b"a1"[..])
        );

        reader
            .size()
            .expect_err("size scan must fail on the corrupt trailing header");

        // The failed size() must not disturb the position: the remaining
        // records of chunk A still come back, then chunk B's.
        for expected in [&b"a2"[..], &b"a3"[..], &b"b"[..]] {
            assert_eq!(
                reader.read_record().expect("read ok").as_deref(),
                Some(expected),
                "records must continue in order after a failed size()"
            );
        }
        assert!(
            reader.read_record().is_err(),
            "the corrupt trailing header still errors on the read path"
        );
    }

    /// size() before the first read must restore the true initial position
    /// (numeric 0) without seeking: the restore special case used to test
    /// the stale chunk_begin==24 model, so it never fired, and the generic
    /// seek path eagerly decoded the first data chunk — firing (and
    /// consuming) a recovery region that the caller never read past.
    #[test]
    fn size_at_initial_position_restores_without_decoding_chunk_data() {
        // [chunk a: CORRUPT data, valid header][chunk b: intact]
        let one = write_records(&[b"a"], WriterOptions::new().chunk_size(1));
        let two = write_records(&[b"a", b"b"], WriterOptions::new().chunk_size(1));
        let mut data = two.clone();
        data[one.len() - 1] ^= 0xFF; // corrupt chunk a's final data byte

        let count: Rc<RefCell<u32>> = Rc::new(RefCell::new(0));
        let rc = Rc::clone(&count);
        let opts = ReaderOptions::new().recovery(move |_r| {
            *rc.borrow_mut() += 1;
            true
        });
        let mut reader = RecordReader::new(Cursor::new(data), opts).expect("reader new ok");

        let pos_before = reader.pos();
        assert_eq!(pos_before.numeric(), 0, "initial position is numeric 0");
        // size() reads only chunk headers (both are hash-valid), so it
        // succeeds and must not touch chunk data or the recovery machinery.
        assert_eq!(reader.size().expect("size ok"), 2);
        assert_eq!(*count.borrow(), 0, "size() must not fire recovery");
        assert_eq!(
            reader.pos(),
            pos_before,
            "size() must preserve the initial read position"
        );

        // The corrupt region is still intact for the actual read: recovery
        // fires now, and the intact sibling is returned.
        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(&b"b"[..])
        );
        assert_eq!(*count.borrow(), 1, "read-path recovery fires normally");
    }

    /// A reader over a file that grows between reads is supported: hitting
    /// a chunk whose claims exceed the current length re-measures the
    /// stream, and a read after the file has grown must succeed. The
    /// re-measure seeks to the end, so the subsequent data read must
    /// position itself explicitly rather than assume the header read left
    /// the reader at the data start.
    #[test]
    fn growing_file_read_resumes_after_remeasure() {
        struct SharedReader(Rc<RefCell<Cursor<Vec<u8>>>>);
        impl std::io::Read for SharedReader {
            fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
                self.0.borrow_mut().read(buf)
            }
        }
        impl std::io::Seek for SharedReader {
            fn seek(&mut self, pos: std::io::SeekFrom) -> std::io::Result<u64> {
                self.0.borrow_mut().seek(pos)
            }
        }

        let one = write_records(&[b"first"], WriterOptions::new().chunk_size(1));
        let full = write_records(&[b"first", b"second"], WriterOptions::new().chunk_size(1));
        assert_eq!(&full[..one.len()], &one[..]);

        // Truncate right after the second chunk's 40-byte header: the header
        // parses, but its claimed data extends past the current end.
        let cut = one.len() + 40;
        assert!(cut < full.len());
        let shared = Rc::new(RefCell::new(Cursor::new(full[..cut].to_vec())));

        let mut reader = RecordReader::new(SharedReader(Rc::clone(&shared)), ReaderOptions::new())
            .expect("reader new ok");
        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(&b"first"[..])
        );

        // Before the file grows, the claim exceeds the stream even after
        // re-measurement: clean error.
        assert!(reader.read_record().is_err(), "truncated read must error");

        // Grow the underlying file to its full contents (position preserved)
        // and retry: the re-measure must accept it AND the data read must
        // land on the chunk, not wherever the re-measure seek ended up.
        {
            let mut c = shared.borrow_mut();
            let pos = c.position();
            *c = Cursor::new(full.clone());
            c.set_position(pos);
        }
        assert_eq!(
            reader
                .read_record()
                .expect("read after growth ok")
                .as_deref(),
            Some(&b"second"[..]),
            "growing-file read must resume correctly after re-measure"
        );
        assert_eq!(reader.read_record().expect("read ok"), None);
    }

    /// A hash-INVALID header's claims are just as hostile as a hash-valid
    /// one's: seek scans peek headers without reporting hash errors inline,
    /// and must not feed unvalidated claims into chunk-end arithmetic (a
    /// u64::MAX claim drives an O(claim) overhead walk — an effective hang).
    #[test]
    fn seek_numeric_rejects_hash_invalid_header_claims() {
        let base = write_records(&[b"only"], WriterOptions::new());
        let mut hostile = hostile_simple_chunk(u64::MAX, 1);
        hostile[0] ^= 0xFF; // corrupt the header hash
        let mut data = base.clone();
        data.extend_from_slice(&hostile);

        let mut reader =
            RecordReader::new(Cursor::new(data), ReaderOptions::new()).expect("reader new ok");
        // Without the peek validity check this call never returns (the
        // overhead walk runs ~2^48 iterations); with it, a clean error.
        let err = reader
            .seek_numeric(u64::MAX / 2)
            .expect_err("corrupt header in scan path must error");
        assert!(
            err.to_string().contains("invalid chunk header hash"),
            "unexpected error: {err}"
        );
    }

    /// Header-claimed sizes beyond the physical stream must produce a clean,
    /// persistent error — no arithmetic overflow (debug panic / release
    /// wrap), no claim-sized allocation, no O(claim) overhead walk. Sweeps
    /// the overflow, mid-range, and barely-past-EOF regimes for both the
    /// data-size and record-count claims.
    #[test]
    fn hostile_header_claims_are_rejected() {
        let base = write_records(&[b"only"], WriterOptions::new());
        for (data_size, num_records) in [
            (u64::MAX, 1u64),   // overflow regime (wraps without saturation)
            (u64::MAX - 40, 1), // offset overflow variant
            (1u64 << 40, 1),    // 1 TiB claim: allocation / walk regime
            (4096, 1),          // modest claim, still past EOF
            (0, u64::MAX >> 8), // maximal record-count claim
            (0, 1u64 << 40),    // mid-range record-count claim
        ] {
            let mut data = base.clone();
            data.extend_from_slice(&hostile_simple_chunk(data_size, num_records));

            let mut reader =
                RecordReader::new(Cursor::new(data), ReaderOptions::new()).expect("reader new ok");
            assert_eq!(
                reader.read_record().expect("read ok").as_deref(),
                Some(&b"only"[..]),
                "data_size={data_size} num_records={num_records}"
            );
            let err = reader
                .read_record()
                .expect_err("hostile claim must be rejected");
            assert!(
                err.to_string().contains("claims"),
                "data_size={data_size} num_records={num_records}: unexpected error: {err}"
            );
            // The rejection must be persistent, like every other read error.
            assert!(
                reader.read_record().is_err(),
                "data_size={data_size} num_records={num_records}: error not persistent"
            );
        }
    }

    /// seek_numeric resolves the target through the block header at the
    /// target's block boundary (C++ SeekToChunkContaining) instead of
    /// walking every chunk from the beginning of the file — so corruption
    /// in an earlier, unrelated region must neither fail the seek nor be
    /// consulted at all.
    #[test]
    fn seek_numeric_is_independent_of_earlier_corruption() {
        // Multi-block file: ~84 KiB across ~40 chunks.
        let records: Vec<Vec<u8>> = (0..80u8).map(|i| vec![i; 1000]).collect();
        let refs: Vec<&[u8]> = records.iter().map(|r| r.as_slice()).collect();
        let data = write_records(&refs, WriterOptions::new().chunk_size(2048));
        assert!(
            data.len() as u64 > BLOCK_SIZE,
            "test file must span more than one block"
        );

        // Collect the last record's position from a clean read.
        let mut clean =
            RecordReader::new(Cursor::new(data.clone()), ReaderOptions::new()).expect("new ok");
        let mut last = None;
        while let Some(rec) = clean.read_record().expect("clean read ok") {
            last = Some((clean.last_pos(), rec));
        }
        let (target_pos, expected) = last.expect("file has records");
        assert!(
            target_pos.chunk_begin > BLOCK_SIZE,
            "target record must live past the first block boundary"
        );

        // Corrupt the FIRST data chunk's header (offset 64..104).
        let mut corrupt = data.clone();
        corrupt[70] ^= 0xFF;

        // Sanity: sequential reading hits the corruption.
        let mut seq =
            RecordReader::new(Cursor::new(corrupt.clone()), ReaderOptions::new()).expect("new ok");
        assert!(
            seq.read_record().is_err(),
            "sequential read must hit the early corruption"
        );

        // Numeric seek to the healthy region succeeds WITHOUT recovery and
        // returns the right record.
        let mut reader =
            RecordReader::new(Cursor::new(corrupt), ReaderOptions::new()).expect("new ok");
        reader
            .seek_numeric(target_pos.numeric())
            .expect("seek_numeric past unrelated corruption must succeed");
        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(expected.as_slice()),
            "seek_numeric must land on the same record as in the clean file"
        );
    }

    /// seek_numeric must not fire the recovery callback for corruption in
    /// regions unrelated to the target (C++ consults the recovery function
    /// only for failures encountered while resolving the target itself).
    #[test]
    fn seek_numeric_does_not_report_unrelated_corruption() {
        let records: Vec<Vec<u8>> = (0..80u8).map(|i| vec![i; 1000]).collect();
        let refs: Vec<&[u8]> = records.iter().map(|r| r.as_slice()).collect();
        let data = write_records(&refs, WriterOptions::new().chunk_size(2048));

        let mut clean =
            RecordReader::new(Cursor::new(data.clone()), ReaderOptions::new()).expect("new ok");
        let mut last = None;
        while let Some(rec) = clean.read_record().expect("clean read ok") {
            last = Some((clean.last_pos(), rec));
        }
        let (target_pos, expected) = last.expect("file has records");
        assert!(target_pos.chunk_begin > BLOCK_SIZE);

        let mut corrupt = data.clone();
        corrupt[70] ^= 0xFF; // first data chunk's header

        let count: Rc<RefCell<u32>> = Rc::new(RefCell::new(0));
        let rc = Rc::clone(&count);
        let opts = ReaderOptions::new().recovery(move |_r| {
            *rc.borrow_mut() += 1;
            true
        });
        let mut reader = RecordReader::new(Cursor::new(corrupt), opts).expect("new ok");
        reader.seek_numeric(target_pos.numeric()).expect("seek ok");
        assert_eq!(
            *count.borrow(),
            0,
            "recovery must not be consulted for regions the seek never visits"
        );
        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(expected.as_slice())
        );
    }

    /// Seeking to a NON-data chunk's address with a nonzero record_index:
    /// C++ reads the chunk AT that address and SetIndex clamps the index
    /// against its zero records — the next read returns record 0 of the
    /// following data chunk. Applying the index to the following chunk
    /// instead silently skipped its first record_index records.
    #[test]
    fn seek_to_padding_chunk_address_clamps_index_against_that_chunk() {
        // [bh][sig][padding @64][chunk "alpha"][chunk "beta"]
        let base = write_records(&[b"alpha", b"beta"], WriterOptions::new().chunk_size(1));
        let mut data = base[..64].to_vec();
        data.extend_from_slice(&special_chunk(ChunkType::Padding, &[0u8; 16], 0, 0));
        data.extend_from_slice(&base[64..]);

        let mut reader =
            RecordReader::new(Cursor::new(data), ReaderOptions::new()).expect("reader new ok");
        reader
            .seek(RecordPosition::new(64, 1))
            .expect("seek to padding address ok");
        assert_eq!(
            reader.pos(),
            RecordPosition::new(64, 0),
            "the index clamps against the addressed (0-record) chunk"
        );
        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(&b"alpha"[..]),
            "no record of the following data chunk may be skipped"
        );
        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(&b"beta"[..])
        );
    }

    /// The unknown-chunk-type rule applies on the seek path exactly as on
    /// the sequential path (C++ ChunkDecoder::Parse backs both): a
    /// hash-valid chunk with an unknown type byte that claims records is an
    /// error — seeking must not silently jump past its records to the next
    /// chunk.
    #[test]
    fn seek_into_unknown_type_chunk_with_records_is_an_error() {
        let one = write_records(&[b"first"], WriterOptions::new().chunk_size(1));
        let two = write_records(&[b"first", b"second"], WriterOptions::new().chunk_size(1));
        assert_eq!(&two[..one.len()], &one[..]);
        let unknown_pos = one.len() as u64;

        let mut data = one.clone();
        data.extend_from_slice(&unknown_type_chunk_with_records(3));
        data.extend_from_slice(&two[one.len()..]);

        let mut reader = RecordReader::new(Cursor::new(data.clone()), ReaderOptions::new())
            .expect("reader new ok");
        let err = reader
            .seek(RecordPosition::new(unknown_pos, 0))
            .expect_err("seek onto an unknown-type chunk carrying records must error");
        assert!(
            err.to_string().contains("unknown chunk type"),
            "unexpected error: {err}"
        );
        // Persistent, like the sequential path.
        assert!(reader.seek(RecordPosition::new(unknown_pos, 0)).is_err());

        // A numeric position inside the unknown chunk's record range
        // resolves to that chunk and fails the same way — not to the next
        // data chunk.
        let mut reader2 =
            RecordReader::new(Cursor::new(data), ReaderOptions::new()).expect("reader new ok");
        let err2 = reader2
            .seek_numeric(unknown_pos + 1)
            .expect_err("numeric seek into an unknown-type chunk's records must error");
        assert!(
            err2.to_string().contains("unknown chunk type"),
            "unexpected error: {err2}"
        );
    }

    /// A shared, growable byte buffer standing in for a file that a
    /// concurrent writer appends to.
    struct SharedBuf {
        buf: Rc<RefCell<Vec<u8>>>,
        pos: u64,
    }

    impl Read for SharedBuf {
        fn read(&mut self, out: &mut [u8]) -> std::io::Result<usize> {
            let buf = self.buf.borrow();
            let pos = self.pos as usize;
            if pos >= buf.len() {
                return Ok(0);
            }
            let n = out.len().min(buf.len() - pos);
            out[..n].copy_from_slice(&buf[pos..pos + n]);
            self.pos += n as u64;
            Ok(n)
        }
    }

    impl Seek for SharedBuf {
        fn seek(&mut self, from: SeekFrom) -> std::io::Result<u64> {
            let len = self.buf.borrow().len() as i64;
            let new = match from {
                SeekFrom::Start(p) => p as i64,
                SeekFrom::End(d) => len + d,
                SeekFrom::Current(d) => self.pos as i64 + d,
            };
            if new < 0 {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::InvalidInput,
                    "seek before start",
                ));
            }
            self.pos = new as u64;
            Ok(self.pos)
        }
    }

    /// End of file is a retriable condition, not a latch: records appended
    /// by a concurrent writer after read_record() returned None must be
    /// returned by later calls (C++ PullChunkHeader re-polls the source and
    /// has an explicit source-has-grown branch — the standard tailing loop
    /// works against a growing file).
    #[test]
    fn eof_is_retriable_records_appended_after_eof_are_returned() {
        let one = write_records(&[b"first"], WriterOptions::new().chunk_size(1));
        let two = write_records(&[b"first", b"second"], WriterOptions::new().chunk_size(1));
        assert_eq!(&two[..one.len()], &one[..], "prefix property");

        let shared: Rc<RefCell<Vec<u8>>> = Rc::new(RefCell::new(one.clone()));
        let mut reader = RecordReader::new(
            SharedBuf {
                buf: Rc::clone(&shared),
                pos: 0,
            },
            ReaderOptions::new(),
        )
        .expect("reader new ok");

        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(&b"first"[..])
        );
        assert_eq!(
            reader.read_record().expect("read ok"),
            None,
            "clean EOF before the writer appends"
        );
        assert_eq!(reader.read_record().expect("read ok"), None);

        // The writer appends a complete chunk.
        *shared.borrow_mut() = two.clone();

        assert_eq!(
            reader
                .read_record()
                .expect("read after growth ok")
                .as_deref(),
            Some(&b"second"[..]),
            "records appended after a clean EOF must be returned"
        );
        assert_eq!(reader.read_record().expect("read ok"), None);
    }

    /// size() honors the recovery contract like read_record() and search()
    /// do: with a callback that continues, corruption is reported and the
    /// scan resumes past the region instead of hard-erroring — and the
    /// count agrees with what read_record() yields under the same policy.
    #[test]
    fn size_consults_the_recovery_callback() {
        // Five single-record chunks; corrupt the third chunk's header.
        let recs: Vec<Vec<u8>> = (0..5u8).map(|i| vec![b'r', b'0' + i]).collect();
        let refs: Vec<&[u8]> = recs.iter().map(|r| r.as_slice()).collect();
        let f2 = write_records(&refs[..2], WriterOptions::new().chunk_size(1));
        let full = write_records(&refs, WriterOptions::new().chunk_size(1));
        assert_eq!(&full[..f2.len()], &f2[..], "prefix property");
        let mut data = full.clone();
        data[f2.len() + 1] ^= 0xFF; // third chunk's header hash

        // Reference: what does read_record() yield under recovery?
        let opts = ReaderOptions::new().recovery(|_r| true);
        let mut seq = RecordReader::new(Cursor::new(data.clone()), opts).expect("new ok");
        let mut readable = 0u64;
        while seq.read_record().expect("read ok").is_some() {
            readable += 1;
        }

        // size() with the same policy must succeed and agree.
        let count: Rc<RefCell<u32>> = Rc::new(RefCell::new(0));
        let rc = Rc::clone(&count);
        let opts = ReaderOptions::new().recovery(move |_r| {
            *rc.borrow_mut() += 1;
            true
        });
        let mut reader = RecordReader::new(Cursor::new(data.clone()), opts).expect("new ok");
        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(recs[0].as_slice())
        );
        let total = reader
            .size()
            .expect("size() must recover from corruption the read path recovers from");
        assert_eq!(total, readable, "size() agrees with the readable count");
        assert!(*count.borrow() >= 1, "the callback saw the region");
        // Position preserved: the next read continues with record 1.
        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(recs[1].as_slice()),
            "size() must not move the read position, recovery included"
        );

        // Without a callback, size() still errors.
        let mut plain = RecordReader::new(Cursor::new(data), ReaderOptions::new()).expect("new ok");
        assert!(plain.size().is_err());
    }

    /// Metadata-chunk DATA corruption (header hash-valid) consults the
    /// recovery callback like the header-peek failure already does — C++
    /// routes ReadChunk and ParseMetadata failures through the recovery
    /// function and returns "no metadata" on recovery.
    #[test]
    fn metadata_data_corruption_recovers_via_callback() {
        let mut buf = Cursor::new(Vec::<u8>::new());
        {
            let mut w = RecordWriter::new(
                &mut buf,
                WriterOptions::new().set_serialized_metadata(b"\x0a\x04test".to_vec()),
            )
            .expect("writer new ok");
            w.write_record(b"rec").expect("write ok");
            w.flush().expect("flush ok");
        }
        let mut data = buf.into_inner();
        data[64 + 40 + 2] ^= 0xFF; // metadata chunk data byte; header stays valid

        // Sanity: without recovery this is a hard error.
        let mut plain =
            RecordReader::new(Cursor::new(data.clone()), ReaderOptions::new()).expect("new ok");
        assert!(
            plain.read_serialized_metadata().is_err(),
            "without recovery the corruption must surface"
        );

        let count: Rc<RefCell<u32>> = Rc::new(RefCell::new(0));
        let rc = Rc::clone(&count);
        let opts = ReaderOptions::new().recovery(move |_r| {
            *rc.borrow_mut() += 1;
            true
        });
        let mut reader = RecordReader::new(Cursor::new(data), opts).expect("new ok");
        assert_eq!(
            reader
                .read_serialized_metadata()
                .expect("recovered metadata read returns Ok"),
            None,
            "the file simply has no readable metadata"
        );
        assert_eq!(*count.borrow(), 1, "the callback saw the region");
        // The record stream is undisturbed (the read path recovers
        // separately when it crosses the same chunk).
        assert_eq!(
            reader.read_record().expect("read ok").as_deref(),
            Some(&b"rec"[..])
        );
    }

    /// A structurally invalid metadata chunk (claiming records) also routes
    /// through the recovery callback — C++ ParseMetadata's num_records
    /// check fails via kRecoverChunkDecoder.
    #[test]
    fn metadata_structural_failure_recovers_via_callback() {
        // [bh][sig][metadata chunk claiming 1 record][data chunk "x"]
        let base = write_records(&[b"x"], WriterOptions::new());
        let mut data = base[..64].to_vec();
        data.extend_from_slice(&special_chunk(ChunkType::FileMetadata, b"\x0a\x00", 1, 2));
        data.extend_from_slice(&base[64..]);

        let count: Rc<RefCell<u32>> = Rc::new(RefCell::new(0));
        let rc = Rc::clone(&count);
        let opts = ReaderOptions::new().recovery(move |_r| {
            *rc.borrow_mut() += 1;
            true
        });
        let mut reader = RecordReader::new(Cursor::new(data), opts).expect("new ok");
        assert_eq!(
            reader
                .read_serialized_metadata()
                .expect("recovered metadata read returns Ok"),
            None
        );
        assert_eq!(*count.borrow(), 1, "the callback saw the region");
    }

    /// Regions reported during forward reads must never overlap, including
    /// across the block-header alias: a resync via a block header whose
    /// next_chunk is 24 ends a region at boundary+24, and a failure right
    /// there canonicalizes back to the boundary — 24 bytes the previous
    /// region already covered.
    #[test]
    fn recovery_regions_never_overlap_at_block_header_alias() {
        use std::cell::RefCell;
        use std::rc::Rc;

        // A chunk whose hash-valid header claims data ending exactly at the
        // 64 KiB boundary has a trusted extent of boundary+24 = 65560. Its
        // data hash mismatches, so region one is [64, 65560). The truncated
        // byte at 65560 then fails at exactly the alias address, whose
        // canonical chunk address is the boundary — 24 bytes back inside
        // region one.
        use crate::chunk_header::{ChunkHeader, ChunkType};
        let mut bytes = Vec::with_capacity(65561);
        bytes.extend_from_slice(&BlockHeader::from_parts(0, 24).to_bytes());
        bytes.extend_from_slice(
            &ChunkHeader::from_parts(&[], ChunkType::FileSignature, 0, 0).to_bytes(),
        );
        // Forged header: hashes computed over ones, file carries zeros.
        let claimed = vec![1u8; 65432]; // data spans [104, 65536) — ends at the boundary
        bytes.extend_from_slice(
            &ChunkHeader::from_parts(&claimed, ChunkType::Simple, 1, 100).to_bytes(),
        );
        bytes.extend_from_slice(&vec![0u8; 65432]);
        assert_eq!(bytes.len(), 65536);
        bytes.extend_from_slice(&BlockHeader::from_parts(65472, 24).to_bytes());
        bytes.push(0xFF);

        let regions: Rc<RefCell<Vec<(u64, u64)>>> = Rc::default();
        let sink = regions.clone();
        let opts = ReaderOptions::new().recovery(move |reg: &crate::SkippedRegion| {
            sink.borrow_mut().push((reg.begin(), reg.end()));
            true
        });
        let mut reader = RecordReader::new(Cursor::new(bytes), opts).expect("recoverable preamble");
        while let Ok(Some(_)) = reader.read_record() {}

        let regs = regions.borrow();
        assert!(regs.len() >= 2, "expected two regions, got {regs:?}");
        let mut last_end = 0u64;
        for &(b, e) in regs.iter() {
            assert!(b < e, "empty region in {regs:?}");
            assert!(b >= last_end, "regions moved backward: {regs:?}");
            last_end = e;
        }
    }
}