armdb 0.7.0

use std::cmp::Reverse;
use std::collections::BinaryHeap;

use zerocopy::FromBytes;

use crate::entry::{EntryHeader, compute_crc32, entry_size};
use crate::error::{DbError, DbResult};

use super::snapshot::{CatchupFile, ShardCatchupSnapshot};

const READ_AHEAD_SIZE: usize = 64 * 1024;
const HEADER_SIZE: usize = size_of::<EntryHeader>();
const PAGE_SIZE: u64 = 4096;
const PROGRESS_STRIDE: usize = 1024;

#[allow(dead_code)]
pub(crate) struct RawEntry {
    pub(crate) data: Vec<u8>,
    pub(crate) gsn: u64,
    pub(crate) file_id: u32,
    pub(crate) file_offset: u64,
    pub(crate) key_len: u16,
}

#[derive(Debug, Clone, Copy)]
struct EntryMeta {
    gsn: u64,
    offset: u64,
    len: u32,
}

struct RunState {
    source: usize,
    file_id: u32,
    run_start: u64,
    run_end: u64,
    next_offset: u64,
    head: EntryMeta,
}

pub(crate) struct ShardLogReader {
    snapshot: ShardCatchupSnapshot,
    key_len: u16,
    runs: Vec<RunState>,
    heap: BinaryHeap<Reverse<(u64, u32, u64, usize)>>,
}

impl std::fmt::Debug for ShardLogReader {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ShardLogReader")
            .field("key_len", &self.key_len)
            .field("run_count", &self.runs.len())
            .finish()
    }
}

impl ShardLogReader {
    pub(crate) fn new(
        snapshot: ShardCatchupSnapshot,
        from_gsn: u64,
        key_len: u16,
        max_runs: usize,
        cancel: &mut dyn FnMut() -> DbResult<()>,
    ) -> DbResult<Self> {
        let mut runs = Vec::new();
        let mut heap = BinaryHeap::new();
        let mut qualifying_runs = 0usize;

        for (source_idx, source) in snapshot.files.iter().enumerate() {
            cancel()?;
            let mut scanner = SourceScanner::new(source);
            let mut offset = 0u64;
            let mut current_start = None;
            let mut current_head = None;
            let mut previous_gsn = None;
            let mut headers = 0usize;

            while let Some((meta, next_offset)) =
                scan_next_meta_with_key(&mut scanner, offset, key_len)?
            {
                if previous_gsn.is_some_and(|previous| meta.gsn < previous) {
                    finish_run(
                        &mut runs,
                        &mut heap,
                        source_idx,
                        source.file_id,
                        current_start.expect("run has a start"),
                        meta.offset,
                        current_head,
                        &mut qualifying_runs,
                        max_runs,
                    )?;
                    current_start = Some(meta.offset);
                    current_head = None;
                } else if current_start.is_none() {
                    current_start = Some(meta.offset);
                }

                if current_head.is_none() && meta.gsn >= from_gsn {
                    current_head = Some(meta);
                }
                previous_gsn = Some(meta.gsn);
                offset = next_offset;
                headers += 1;
                if headers >= PROGRESS_STRIDE {
                    headers = 0;
                    cancel()?;
                }
            }

            if let Some(run_start) = current_start {
                finish_run(
                    &mut runs,
                    &mut heap,
                    source_idx,
                    source.file_id,
                    run_start,
                    source.readable_end,
                    current_head,
                    &mut qualifying_runs,
                    max_runs,
                )?;
            }
        }

        Ok(Self {
            snapshot,
            key_len,
            runs,
            heap,
        })
    }

    pub(crate) fn next_entry(&mut self) -> DbResult<Option<RawEntry>> {
        let Some(Reverse((_, _, _, idx))) = self.heap.pop() else {
            return Ok(None);
        };

        let meta = self.runs[idx].head;
        let source_idx = self.runs[idx].source;
        let file_id = self.runs[idx].file_id;
        let data = self.snapshot.files[source_idx].read_exact_at(meta.offset, meta.len as usize)?;
        let header = EntryHeader::read_from_bytes(&data[..HEADER_SIZE]).map_err(|_| {
            DbError::CorruptedEntry {
                offset: meta.offset,
            }
        })?;
        if header.sequence() != meta.gsn {
            return Err(DbError::CorruptedEntry {
                offset: meta.offset,
            });
        }
        verify_crc(&data, &header, self.key_len, meta.offset)?;

        let next = next_meta_in_run(
            &self.snapshot.files[source_idx],
            self.runs[idx].next_offset,
            self.runs[idx].run_end,
            self.key_len,
        )?;
        if let Some((next_head, next_offset)) = next {
            self.runs[idx].head = next_head;
            self.runs[idx].next_offset = next_offset;
            self.heap.push(Reverse((
                next_head.gsn,
                file_id,
                self.runs[idx].run_start,
                idx,
            )));
        }

        Ok(Some(RawEntry {
            data,
            gsn: meta.gsn,
            file_id,
            file_offset: meta.offset,
            key_len: self.key_len,
        }))
    }

    #[cfg(test)]
    fn buffered_payload_bytes(&self) -> usize {
        0
    }
}

#[allow(clippy::too_many_arguments)]
fn finish_run(
    runs: &mut Vec<RunState>,
    heap: &mut BinaryHeap<Reverse<(u64, u32, u64, usize)>>,
    source: usize,
    file_id: u32,
    run_start: u64,
    run_end: u64,
    head: Option<EntryMeta>,
    qualifying_runs: &mut usize,
    max_runs: usize,
) -> DbResult<()> {
    let Some(head) = head else {
        return Ok(());
    };
    *qualifying_runs += 1;
    if *qualifying_runs > max_runs {
        return Err(DbError::CatchupResourceLimit {
            actual: *qualifying_runs,
            limit: max_runs,
        });
    }

    let idx = runs.len();
    runs.push(RunState {
        source,
        file_id,
        run_start,
        run_end,
        next_offset: head.offset + u64::from(head.len),
        head,
    });
    heap.push(Reverse((head.gsn, file_id, run_start, idx)));
    Ok(())
}

struct SourceScanner<'a> {
    source: &'a CatchupFile,
    buffer: Vec<u8>,
    buffer_offset: u64,
}

impl<'a> SourceScanner<'a> {
    fn new(source: &'a CatchupFile) -> Self {
        Self {
            source,
            buffer: Vec::new(),
            buffer_offset: 0,
        }
    }

    fn read(&mut self, offset: u64, len: usize) -> DbResult<Vec<u8>> {
        let end = offset
            .checked_add(len as u64)
            .ok_or(DbError::CorruptedEntry { offset })?;
        let buffer_end = self.buffer_offset + self.buffer.len() as u64;
        if offset < self.buffer_offset || end > buffer_end {
            let available = self.source.readable_end.saturating_sub(offset) as usize;
            if available < len {
                return Err(DbError::CorruptedEntry { offset });
            }
            let read_len = available.min(READ_AHEAD_SIZE).max(len);
            self.buffer = self.source.read_exact_at(offset, read_len)?;
            self.buffer_offset = offset;
        }
        let start = (offset - self.buffer_offset) as usize;
        Ok(self.buffer[start..start + len].to_vec())
    }
}

fn scan_next_meta_with_key(
    scanner: &mut SourceScanner<'_>,
    mut offset: u64,
    key_len: u16,
) -> DbResult<Option<(EntryMeta, u64)>> {
    loop {
        if offset == scanner.source.readable_end {
            return Ok(None);
        }
        if scanner.source.readable_end.saturating_sub(offset) < HEADER_SIZE as u64 {
            return Err(DbError::CorruptedEntry { offset });
        }
        let header_bytes = scanner.read(offset, HEADER_SIZE)?;
        let header = EntryHeader::read_from_bytes(&header_bytes)
            .map_err(|_| DbError::CorruptedEntry { offset })?;
        if is_zero_header(&header) {
            let page_end = (offset + HEADER_SIZE as u64).div_ceil(PAGE_SIZE) * PAGE_SIZE;
            if page_end > scanner.source.readable_end {
                return Err(DbError::CorruptedEntry { offset });
            }
            let pad = scanner.read(
                offset + HEADER_SIZE as u64,
                (page_end - offset - HEADER_SIZE as u64) as usize,
            )?;
            if !pad.iter().all(|byte| *byte == 0) {
                return Err(DbError::CorruptedEntry { offset });
            }
            offset = page_end;
            continue;
        }
        let total = entry_size(key_len as usize, header.value_len);
        let end = offset
            .checked_add(total)
            .ok_or(DbError::CorruptedEntry { offset })?;
        if end > scanner.source.readable_end {
            return Err(DbError::CorruptedEntry { offset });
        }
        return Ok(Some((
            EntryMeta {
                gsn: header.sequence(),
                offset,
                len: total
                    .try_into()
                    .map_err(|_| DbError::CorruptedEntry { offset })?,
            },
            end,
        )));
    }
}

fn next_meta_in_run(
    source: &CatchupFile,
    offset: u64,
    run_end: u64,
    key_len: u16,
) -> DbResult<Option<(EntryMeta, u64)>> {
    if offset >= run_end {
        return Ok(None);
    }
    let bytes = source.read_exact_at(offset, HEADER_SIZE)?;
    let header =
        EntryHeader::read_from_bytes(&bytes).map_err(|_| DbError::CorruptedEntry { offset })?;
    if is_zero_header(&header) {
        let page_end = (offset + HEADER_SIZE as u64).div_ceil(PAGE_SIZE) * PAGE_SIZE;
        if page_end >= run_end {
            return Ok(None);
        }
        let pad = source.read_exact_at(
            offset + HEADER_SIZE as u64,
            (page_end - offset - HEADER_SIZE as u64) as usize,
        )?;
        if !pad.iter().all(|byte| *byte == 0) {
            return Err(DbError::CorruptedEntry { offset });
        }
        return next_meta_in_run(source, page_end, run_end, key_len);
    }
    let total = entry_size(key_len as usize, header.value_len);
    let end = offset
        .checked_add(total)
        .ok_or(DbError::CorruptedEntry { offset })?;
    if end > run_end {
        return Err(DbError::CorruptedEntry { offset });
    }
    Ok(Some((
        EntryMeta {
            gsn: header.sequence(),
            offset,
            len: total
                .try_into()
                .map_err(|_| DbError::CorruptedEntry { offset })?,
        },
        end,
    )))
}

fn is_zero_header(header: &EntryHeader) -> bool {
    header.gsn == 0 && header.value_len == 0 && header.crc32 == 0
}

fn verify_crc(data: &[u8], header: &EntryHeader, key_len: u16, offset: u64) -> DbResult<()> {
    let key_start = HEADER_SIZE;
    let value_start = key_start + key_len as usize;
    let value_end = value_start
        .checked_add(header.value_len as usize)
        .ok_or(DbError::CorruptedEntry { offset })?;
    if value_end > data.len() {
        return Err(DbError::CorruptedEntry { offset });
    }
    let actual = compute_crc32(
        header.gsn,
        header.value_len,
        &data[key_start..value_start],
        &data[value_start..value_end],
    );
    if actual != header.crc32 {
        return Err(DbError::CrcMismatch {
            expected: header.crc32,
            actual,
        });
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use std::io::Write;
    use std::path::{Path, PathBuf};
    use std::sync::{Arc, atomic::AtomicU64};

    use super::*;
    use crate::entry::serialize_entry;
    use crate::replication::snapshot::ShardCatchupSnapshot;
    use crate::shard::Shard;

    fn snapshot_with_four_monotonic_runs() -> ShardCatchupSnapshot {
        snapshot_from_gsns(&[1, 5, 2, 6, 3, 7, 4, 8], 8)
    }

    fn snapshot_with_single_value(value_len: usize) -> ShardCatchupSnapshot {
        snapshot_from_serialized_entries(vec![serialize_entry(
            1,
            &[1; 8],
            &vec![0xAB; value_len],
            false,
        )])
    }

    fn snapshot_from_gsns(gsns: &[u64], key_len: usize) -> ShardCatchupSnapshot {
        let entries = gsns
            .iter()
            .map(|gsn| serialize_entry(*gsn, &gsn.to_be_bytes()[8 - key_len..], b"v", false))
            .collect();
        snapshot_from_serialized_entries(entries)
    }

    fn snapshot_from_serialized_entries(entries: Vec<Vec<u8>>) -> ShardCatchupSnapshot {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("000001.data");
        let mut file = std::fs::File::create(&path).unwrap();
        for entry in entries {
            file.write_all(&entry).unwrap();
        }
        file.sync_all().unwrap();
        drop(file);

        let shard = Shard::open(
            0,
            dir.path(),
            1 << 20,
            64 * 1024,
            false,
            false,
            false,
            crate::config::IoBackend::default(),
            Arc::new(AtomicU64::new(10_000)),
        )
        .unwrap();
        shard.catchup_snapshot().unwrap()
    }

    #[test]
    fn run_limit_rejects_before_streaming() {
        let snapshot = snapshot_with_four_monotonic_runs();
        let err = ShardLogReader::new(snapshot, 0, 8, 3, &mut || Ok(())).unwrap_err();
        assert!(matches!(
            err,
            DbError::CatchupResourceLimit {
                actual: 4,
                limit: 3
            }
        ));
    }

    #[test]
    fn prescan_buffers_metadata_not_payload() {
        let snapshot = snapshot_with_single_value(256 * 1024);
        let mut reader = ShardLogReader::new(snapshot, 0, 8, 8, &mut || Ok(())).unwrap();
        assert_eq!(reader.buffered_payload_bytes(), 0);
        assert_eq!(
            reader.next_entry().unwrap().unwrap().data.len(),
            entry_size(8, 256 * 1024) as usize
        );
        assert_eq!(reader.buffered_payload_bytes(), 0);
    }

    /// Write a sequence of entries into a file, returning the file path.
    fn write_data_file(dir: &Path, file_id: u32, entries: &[(u64, &[u8], &[u8])]) {
        let path = dir.join(format!("{file_id:06}.data"));
        let mut f = std::fs::File::create(&path).unwrap();
        for &(gsn, key, value) in entries {
            let data = serialize_entry(gsn, key, value, false);
            f.write_all(&data).unwrap();
        }
    }

    fn snapshot_from_dir(
        shard_dir: PathBuf,
        #[cfg(feature = "encryption")] cipher: Option<Arc<crate::crypto::PageCipher>>,
    ) -> DbResult<ShardCatchupSnapshot> {
        #[cfg(feature = "encryption")]
        let shard = Shard::open_encrypted(
            0,
            &shard_dir,
            1 << 30,
            64 * 1024,
            false,
            false,
            false,
            crate::config::IoBackend::default(),
            cipher,
            Arc::new(AtomicU64::new(10_000)),
        )?;
        #[cfg(not(feature = "encryption"))]
        let shard = Shard::open(
            0,
            &shard_dir,
            1 << 30,
            64 * 1024,
            false,
            false,
            false,
            crate::config::IoBackend::default(),
            Arc::new(AtomicU64::new(10_000)),
        )?;
        shard.catchup_snapshot()
    }

    fn reader_from_dir(
        shard_dir: PathBuf,
        from_gsn: u64,
        key_len: u16,
        #[cfg(feature = "encryption")] cipher: Option<Arc<crate::crypto::PageCipher>>,
    ) -> DbResult<ShardLogReader> {
        let snapshot = snapshot_from_dir(
            shard_dir,
            #[cfg(feature = "encryption")]
            cipher,
        )?;
        ShardLogReader::new(snapshot, from_gsn, key_len, usize::MAX, &mut || Ok(()))
    }

    impl ShardLogReader {
        fn new_with_progress(
            shard_dir: PathBuf,
            from_gsn: u64,
            key_len: u16,
            #[cfg(feature = "encryption")] cipher: Option<Arc<crate::crypto::PageCipher>>,
            on_progress: &mut dyn FnMut() -> DbResult<()>,
        ) -> DbResult<Self> {
            let snapshot = snapshot_from_dir(
                shard_dir,
                #[cfg(feature = "encryption")]
                cipher,
            )?;
            ShardLogReader::new(snapshot, from_gsn, key_len, usize::MAX, on_progress)
        }
    }

    /// Regression test for C6: cross-file rewind bug in `skip_until_gsn`.
    ///
    /// The old implementation saved `current_offset` before calling
    /// `next_entry()`.  When the entry at the end of file 0 caused
    /// `next_entry()` to call `advance_file()` — which resets
    /// `current_offset` to 0 for file 1 — the subsequent restore wrote the
    /// stale EOF offset of file 0 back into the new file's position pointer.
    /// The next `next_entry()` call then attempted to read from that stale
    /// offset in file 1, which lies past EOF, and returned `None` instead of
    /// the entry that was actually there.
    ///
    /// The k-way-merge rewrite reads each file with an independent cursor, so
    /// the cross-file advance never corrupts another file's offset.
    #[test]
    fn skip_until_gsn_cross_file_boundary() {
        let dir = tempfile::tempdir().unwrap();
        let shard_dir = dir.path().to_path_buf();

        // 8-byte keys so entry_size(8, 8) = 32 bytes each.
        let key0 = b"key00000";
        let key1 = b"key11111";
        let value = b"vvvvvvvv"; // 8 bytes

        // File 0 (id=0): GSN 1 and GSN 2.
        write_data_file(&shard_dir, 0, &[(1, key0, value), (2, key0, value)]);

        // File 1 (id=1): GSN 3 (the target) and GSN 4.
        write_data_file(&shard_dir, 1, &[(3, key1, value), (4, key1, value)]);

        // Seek to GSN 3 — this must cross the file boundary without corrupting
        // any offset.
        let mut reader = reader_from_dir(
            shard_dir,
            3,
            8,
            #[cfg(feature = "encryption")]
            None,
        )
        .unwrap();

        let entry = reader
            .next_entry()
            .unwrap()
            .expect("should find entry with GSN 3");
        assert_eq!(entry.gsn, 3, "first entry after seek must be GSN 3");
        assert_eq!(entry.file_id, 1, "entry must come from file 1");
        assert_eq!(entry.file_offset, 0, "entry must be at the start of file 1");

        // Confirm the reader is in a sane state: next entry is GSN 4.
        let entry2 = reader
            .next_entry()
            .unwrap()
            .expect("should find entry with GSN 4");
        assert_eq!(entry2.gsn, 4);
    }

    /// Regression: `next_entry` must skip mid-file page padding (zero header +
    /// zero tail to page boundary), not step 16 bytes at a time through the page.
    #[test]
    fn log_reader_skips_mid_file_padding() {
        let dir = tempfile::tempdir().unwrap();
        let shard_dir = dir.path().to_path_buf();

        let key = b"padkey00";
        let value = b"padval00";
        let e1 = serialize_entry(1, key, value, false);
        let e2 = serialize_entry(2, key, value, false);

        // Layout: e1, zero-pad to next page boundary, e2.
        let mut file_bytes = Vec::new();
        file_bytes.extend_from_slice(&e1);
        let pad = 4096 - (file_bytes.len() % 4096);
        file_bytes.extend(std::iter::repeat_n(0u8, pad));
        let page_boundary = file_bytes.len() as u64;
        file_bytes.extend_from_slice(&e2);

        let path = shard_dir.join("000000.data");
        std::fs::write(&path, &file_bytes).unwrap();

        let mut reader = reader_from_dir(
            shard_dir,
            0,
            8,
            #[cfg(feature = "encryption")]
            None,
        )
        .unwrap();

        let entry1 = reader
            .next_entry()
            .unwrap()
            .expect("first entry after padding");
        assert_eq!(entry1.gsn, 1);
        assert_eq!(entry1.file_offset, 0);

        let entry2 = reader
            .next_entry()
            .unwrap()
            .expect("second entry at page boundary");
        assert_eq!(entry2.gsn, 2);
        assert_eq!(
            entry2.file_offset, page_boundary,
            "second entry must start at the page boundary after padding"
        );

        assert!(
            reader.next_entry().unwrap().is_none(),
            "no bogus zero entry"
        );
    }

    /// Regression: `skip_until_gsn` (not just `next_entry`) must skip mid-file
    /// page padding. `log_reader_skips_mid_file_padding` seeks from GSN 0, which
    /// routes through the whole-file run; this seeks to a GSN living in the entry
    /// *after* the padding, forcing the `seek_to_gsn` walk to step over the zero
    /// region.
    #[test]
    fn skip_until_gsn_skips_mid_file_padding() {
        let dir = tempfile::tempdir().unwrap();
        let shard_dir = dir.path().to_path_buf();

        let key = b"padkey00";
        let value = b"padval00";
        let e1 = serialize_entry(1, key, value, false);
        let e2 = serialize_entry(2, key, value, false);

        // Layout: e1 (GSN 1), zero-pad to next page boundary, e2 (GSN 2).
        let mut file_bytes = Vec::new();
        file_bytes.extend_from_slice(&e1);
        let pad = 4096 - (file_bytes.len() % 4096);
        file_bytes.extend(std::iter::repeat_n(0u8, pad));
        let page_boundary = file_bytes.len() as u64;
        file_bytes.extend_from_slice(&e2);

        // No .hint file, so the reader walks with a single whole-file run.
        let path = shard_dir.join("000000.data");
        std::fs::write(&path, &file_bytes).unwrap();

        let mut reader = reader_from_dir(
            shard_dir,
            2,
            8,
            #[cfg(feature = "encryption")]
            None,
        )
        .unwrap();

        let entry = reader
            .next_entry()
            .unwrap()
            .expect("should find GSN 2 after padding");
        assert_eq!(entry.gsn, 2);
        assert_eq!(
            entry.file_offset, page_boundary,
            "GSN 2 must start at the page boundary after padding"
        );

        assert!(reader.next_entry().unwrap().is_none(), "no more entries");
    }

    /// Sanity: seeking to a GSN that lives in the first file still works.
    #[test]
    fn skip_until_gsn_single_file() {
        let dir = tempfile::tempdir().unwrap();
        let shard_dir = dir.path().to_path_buf();

        let key = b"thekey00";
        let value = b"theval00";

        write_data_file(
            &shard_dir,
            0,
            &[(1, key, value), (2, key, value), (3, key, value)],
        );

        let mut reader = reader_from_dir(
            shard_dir,
            2,
            8,
            #[cfg(feature = "encryption")]
            None,
        )
        .unwrap();

        let entry = reader
            .next_entry()
            .unwrap()
            .expect("should find entry with GSN 2");
        assert_eq!(entry.gsn, 2);

        let entry2 = reader
            .next_entry()
            .unwrap()
            .expect("should find entry with GSN 3");
        assert_eq!(entry2.gsn, 3);

        assert!(reader.next_entry().unwrap().is_none(), "no more entries");
    }

    /// F2: a compaction output holds *old* GSNs at a *high* file-id, while a
    /// lower-id file holds *newer* GSNs. Streamed strictly in file-id order the
    /// old GSNs would arrive after the new ones and be dropped by the follower's
    /// watermark. The k-way merge must instead emit a globally GSN-ascending
    /// stream so every entry survives.
    #[test]
    fn kway_merge_orders_across_files_by_gsn() {
        let dir = tempfile::tempdir().unwrap();
        let shard_dir = dir.path().to_path_buf();

        let key = b"mergekey";
        let value = b"mergeval";

        // File 3 (lower id): NEWER GSNs 201..203.
        write_data_file(
            &shard_dir,
            3,
            &[(201, key, value), (202, key, value), (203, key, value)],
        );
        // File 4 (higher id, compaction output): OLDER GSNs 101..103.
        write_data_file(
            &shard_dir,
            4,
            &[(101, key, value), (102, key, value), (103, key, value)],
        );

        let mut reader = reader_from_dir(
            shard_dir,
            0,
            8,
            #[cfg(feature = "encryption")]
            None,
        )
        .unwrap();

        let mut gsns = Vec::new();
        while let Some(e) = reader.next_entry().unwrap() {
            gsns.push(e.gsn);
        }

        assert_eq!(
            gsns,
            vec![101, 102, 103, 201, 202, 203],
            "k-way merge must emit a globally GSN-ascending stream regardless of file-id order"
        );
    }

    /// F2: seeking with `from_gsn` across files where file-id order != GSN order
    /// must still find every entry >= from_gsn and emit them GSN-sorted. This is
    /// the exact watermark-drop scenario: follower is at GSN 150 and must receive
    /// 151..203, including the old-GSN high-id compaction output (151..203 split
    /// across the two files).
    #[test]
    fn kway_merge_from_gsn_across_disordered_files() {
        let dir = tempfile::tempdir().unwrap();
        let shard_dir = dir.path().to_path_buf();

        let key = b"seekkey0";
        let value = b"seekval0";

        // File 2 (lower id): GSNs 100, 160, 170 (a compaction output can hold a
        // gappy set — 100 is below from_gsn and must be skipped).
        write_data_file(
            &shard_dir,
            2,
            &[(100, key, value), (160, key, value), (170, key, value)],
        );
        // File 5 (higher id): GSNs 151, 152, 180.
        write_data_file(
            &shard_dir,
            5,
            &[(151, key, value), (152, key, value), (180, key, value)],
        );

        let mut reader = reader_from_dir(
            shard_dir,
            151,
            8,
            #[cfg(feature = "encryption")]
            None,
        )
        .unwrap();

        let mut gsns = Vec::new();
        while let Some(e) = reader.next_entry().unwrap() {
            gsns.push(e.gsn);
        }

        assert_eq!(
            gsns,
            vec![151, 152, 160, 170, 180],
            "from_gsn seek must skip <151 and emit the rest GSN-sorted across files"
        );
    }

    /// F2: an *internally non-monotonic* file — a compaction output that merged
    /// file-id-vs-GSN-disordered inputs, so a single file holds e.g. GSN 300..302
    /// followed by GSN 50..52. The header prescan must split it into two runs so
    /// the low GSNs are not stranded behind the high ones. Combined with a second
    /// file that holds GSN 100..102, the merged stream must be globally sorted.
    #[test]
    fn kway_merge_splits_internally_unsorted_file_into_runs() {
        let dir = tempfile::tempdir().unwrap();
        let shard_dir = dir.path().to_path_buf();

        let key = b"runkey00";
        let value = b"runval00";

        // File 7: internally non-monotonic — 300,301,302 then 50,51,52.
        write_data_file(
            &shard_dir,
            7,
            &[
                (300, key, value),
                (301, key, value),
                (302, key, value),
                (50, key, value),
                (51, key, value),
                (52, key, value),
            ],
        );
        // File 8: 100,101,102.
        write_data_file(
            &shard_dir,
            8,
            &[(100, key, value), (101, key, value), (102, key, value)],
        );

        let mut reader = reader_from_dir(
            shard_dir,
            0,
            8,
            #[cfg(feature = "encryption")]
            None,
        )
        .unwrap();

        let mut gsns = Vec::new();
        while let Some(e) = reader.next_entry().unwrap() {
            gsns.push(e.gsn);
        }

        assert_eq!(
            gsns,
            vec![50, 51, 52, 100, 101, 102, 300, 301, 302],
            "an internally unsorted file must be split into GSN-monotonic runs and merged"
        );
    }

    /// F2: `from_gsn` seek into the *second run* of an internally-unsorted file.
    /// The low-GSN run (50..52) is entirely below from_gsn=60 and must be
    /// skipped, while the high-GSN run (300..302) is above it.
    #[test]
    fn kway_merge_from_gsn_into_unsorted_file_runs() {
        let dir = tempfile::tempdir().unwrap();
        let shard_dir = dir.path().to_path_buf();

        let key = b"run2key0";
        let value = b"run2val0";

        // File 7: 300,301,302 then 50,51,52 (two runs).
        write_data_file(
            &shard_dir,
            7,
            &[
                (300, key, value),
                (301, key, value),
                (302, key, value),
                (50, key, value),
                (51, key, value),
                (52, key, value),
            ],
        );

        let mut reader = reader_from_dir(
            shard_dir,
            60,
            8,
            #[cfg(feature = "encryption")]
            None,
        )
        .unwrap();

        let mut gsns = Vec::new();
        while let Some(e) = reader.next_entry().unwrap() {
            gsns.push(e.gsn);
        }

        assert_eq!(
            gsns,
            vec![300, 301, 302],
            "from_gsn=60 must skip the entire 50..52 run and emit only 300..302"
        );
    }

    /// Verify that `ShardLogReader::new` with `cipher: None` compiles and
    /// functions identically to the old three-argument form.
    #[test]
    fn plaintext_reader_with_none_cipher_smoke() {
        let dir = tempfile::tempdir().unwrap();
        let shard_dir = dir.path().to_path_buf();

        let key = b"smkeyyyy";
        let value = b"smvalyyy";

        write_data_file(&shard_dir, 0, &[(10, key, value), (11, key, value)]);

        let mut reader = reader_from_dir(
            shard_dir,
            0,
            8,
            #[cfg(feature = "encryption")]
            None,
        )
        .unwrap();

        let e1 = reader.next_entry().unwrap().expect("entry 10");
        assert_eq!(e1.gsn, 10);
        let e2 = reader.next_entry().unwrap().expect("entry 11");
        assert_eq!(e2.gsn, 11);
        assert!(reader.next_entry().unwrap().is_none());
    }

    /// Collect the full GSN-ordered `(gsn, data)` stream from retained snapshot
    /// handles. The final argument is retained to preserve the historical
    /// comparison test's two construction paths.
    fn collect(
        shard_dir: PathBuf,
        from_gsn: u64,
        key_len: u16,
        _legacy_window: usize,
    ) -> Vec<(u64, Vec<u8>)> {
        let snapshot = snapshot_from_dir(
            shard_dir,
            #[cfg(feature = "encryption")]
            None,
        )
        .unwrap();
        let mut r =
            ShardLogReader::new(snapshot, from_gsn, key_len, usize::MAX, &mut || Ok(())).unwrap();
        let mut out = Vec::new();
        while let Some(e) = r.next_entry().unwrap() {
            out.push((e.gsn, e.data));
        }
        out
    }

    /// Preserve the historical bounded-window ordering assertion. A shard with interleaving
    /// single-run files, a disjoint file, and an internally-unsorted (two-run)
    /// file must produce the *identical* GSN-ordered stream whether the merge
    /// is captured through two independent snapshots. Snapshot ownership must
    /// not reorder or drop a single entry.
    #[test]
    fn bounded_window_matches_full_merge() {
        let dir = tempfile::tempdir().unwrap();
        let sd = dir.path().to_path_buf();
        let k = b"eqkey000";
        let v = b"eqval000";

        // Interleaving single-run files (even/odd GSNs in the same range).
        write_data_file(&sd, 0, &[(100, k, v), (102, k, v), (104, k, v)]);
        write_data_file(&sd, 1, &[(101, k, v), (103, k, v), (105, k, v)]);
        // Disjoint higher range.
        write_data_file(&sd, 2, &[(200, k, v), (201, k, v), (202, k, v)]);
        // Internally non-monotonic file: run 300..302 then run 106..108.
        write_data_file(
            &sd,
            3,
            &[
                (300, k, v),
                (301, k, v),
                (302, k, v),
                (106, k, v),
                (107, k, v),
                (108, k, v),
            ],
        );

        let full = collect(sd.clone(), 0, 8, 100_000);
        let bounded = collect(sd.clone(), 0, 8, 1);
        assert_eq!(
            full, bounded,
            "independent retained snapshots must produce the same merged output"
        );

        let gsns: Vec<u64> = full.iter().map(|(g, _)| *g).collect();
        assert_eq!(
            gsns,
            vec![
                100, 101, 102, 103, 104, 105, 106, 107, 108, 200, 201, 202, 300, 301, 302
            ],
            "output must be globally GSN-ascending and complete"
        );

        // Same equivalence under a from_gsn seek that lands mid-run.
        let full2 = collect(sd.clone(), 104, 8, 100_000);
        let bounded2 = collect(sd, 104, 8, 1);
        assert_eq!(full2, bounded2);
        let g2: Vec<u64> = full2.iter().map(|(g, _)| *g).collect();
        assert_eq!(
            g2,
            vec![104, 105, 106, 107, 108, 200, 201, 202, 300, 301, 302],
            "from_gsn seek must skip <104 and stay sorted"
        );
    }

    /// Retained handles, rather than pathnames, own every snapshot source.
    /// Unlinking all captured paths therefore cannot make the merge incomplete.
    #[test]
    fn retained_snapshot_handles_survive_source_unlink() {
        let dir = tempfile::tempdir().unwrap();
        let shard = Shard::open(
            0,
            dir.path(),
            1 << 20,
            64 * 1024,
            false,
            false,
            false,
            crate::config::IoBackend::default(),
            Arc::new(AtomicU64::new(1)),
        )
        .unwrap();
        for gsn in 1u64..=8 {
            let entry = serialize_entry(gsn, &gsn.to_be_bytes(), b"v", false);
            shard.lock().append_raw_entry(0, 8, &entry).unwrap();
            shard.rotate_active_for_test(8).unwrap();
        }
        let snapshot = shard.catchup_snapshot().unwrap();
        for source in &snapshot.files {
            std::fs::remove_file(dir.path().join(format!("{:06}.data", source.file_id))).unwrap();
        }
        let mut r = ShardLogReader::new(snapshot, 0, 8, usize::MAX, &mut || Ok(())).unwrap();

        let mut gsns = Vec::new();
        while let Some(e) = r.next_entry().unwrap() {
            gsns.push(e.gsn);
        }
        assert_eq!(gsns, (1..=8).collect::<Vec<u64>>());
    }

    /// Fix 1 (livelock) — the prescan invokes `on_progress`. A file with more
    /// than `PROGRESS_STRIDE` entries must ping the callback at least once during
    /// the header prescan, which is what lets the server emit a keepalive
    /// heartbeat before the first `EntryBatch` on a large/cold shard. The old
    /// `ShardLogReader::new` had no such hook, so its prescan sent nothing.
    #[test]
    fn prescan_calls_progress_on_large_file() {
        let dir = tempfile::tempdir().unwrap();
        let sd = dir.path().to_path_buf();
        let key = b"prgkey00";
        let value = b"prgval00";
        let entries: Vec<(u64, &[u8], &[u8])> =
            (1..=9000u64).map(|g| (g, &key[..], &value[..])).collect();
        write_data_file(&sd, 0, &entries);

        let mut calls = 0usize;
        {
            let mut on_progress = || -> DbResult<()> {
                calls += 1;
                Ok(())
            };
            let _r = ShardLogReader::new_with_progress(
                sd,
                0,
                8,
                #[cfg(feature = "encryption")]
                None,
                &mut on_progress,
            )
            .unwrap();
        }
        assert!(
            calls >= 1,
            "a >{PROGRESS_STRIDE}-entry prescan must ping on_progress (keepalive hook)"
        );
    }

    /// Fix 1 (livelock) — heartbeats actually reach the writer *during* the
    /// prescan. With an `on_progress` that writes a Heartbeat frame on every call
    /// (a pacer whose interval has elapsed), constructing the reader over a large
    /// file must leave at least one Heartbeat frame in the buffer — proof that a
    /// slow catch-up prep no longer sends the follower nothing. Without the hook
    /// the buffer would be empty, which is exactly the livelock precondition.
    #[test]
    fn new_with_progress_emits_heartbeats_into_writer() {
        use crate::replication::protocol::{
            MessageType, encode_heartbeat, read_frame, write_frame,
        };

        let dir = tempfile::tempdir().unwrap();
        let sd = dir.path().to_path_buf();
        let key = b"hbkey000";
        let value = b"hbval000";
        let entries: Vec<(u64, &[u8], &[u8])> =
            (1..=9000u64).map(|g| (g, &key[..], &value[..])).collect();
        write_data_file(&sd, 0, &entries);

        let mut buf: Vec<u8> = Vec::new();
        {
            let mut on_progress = || -> DbResult<()> {
                write_frame(&mut buf, &encode_heartbeat()).map_err(Into::into)
            };
            let _r = ShardLogReader::new_with_progress(
                sd,
                0,
                8,
                #[cfg(feature = "encryption")]
                None,
                &mut on_progress,
            )
            .unwrap();
        }

        let mut cur = &buf[..];
        let mut heartbeats = 0usize;
        while !cur.is_empty() {
            let f = read_frame(&mut cur).expect("frame");
            if f.msg_type == MessageType::Heartbeat {
                heartbeats += 1;
            }
        }
        assert!(
            heartbeats >= 1,
            "a heartbeat frame must be emitted during a large prescan"
        );
    }

    /// An encrypted snapshot retains both data and tag handles across an unlink,
    /// including an internally non-monotonic source split into multiple runs.
    #[cfg(all(feature = "encryption", feature = "replication"))]
    #[test]
    fn encrypted_snapshot_retains_handles_across_runs() {
        use std::sync::atomic::AtomicU64;

        use crate::crypto::PageCipher;
        use crate::shard::Shard;

        let dir = tempfile::tempdir().unwrap();
        let gsn = Arc::new(AtomicU64::new(0));
        let cipher = Arc::new(PageCipher::new(&[0x33; 32]).expect("create cipher"));

        let shard = Shard::open_encrypted(
            0,
            dir.path(),
            1 << 20,
            64 * 1024,
            false,
            false,
            false,
            crate::config::IoBackend::default(),
            Some(cipher.clone()),
            gsn,
        )
        .expect("open encrypted shard");

        let key_len: u16 = 8;
        let key = b"rekey000";
        // High GSNs first, then low GSNs → one file, two GSN-monotonic runs.
        let order = [10u64, 11, 12, 1, 2, 3];
        {
            let mut inner = shard.lock();
            for &g in &order {
                let mut value = *b"reval000";
                value[0] = g as u8;
                let entry = serialize_entry(g, key, &value, false);
                inner
                    .append_raw_entry(0, key_len, &entry)
                    .expect("append entry");
            }
        }
        shard.flush().expect("flush encrypted shard");

        let snapshot = shard
            .catchup_snapshot()
            .expect("capture encrypted snapshot");
        std::fs::remove_file(dir.path().join("000001.data")).unwrap();
        std::fs::remove_file(dir.path().join("000001.tags")).unwrap();
        let mut reader = ShardLogReader::new(snapshot, 0, key_len, usize::MAX, &mut || Ok(()))
            .expect("open encrypted snapshot reader");

        let kl = key_len as usize;
        let mut got = Vec::new();
        while let Some(e) = reader.next_entry().expect("read entry") {
            assert_eq!(&e.data[16..16 + kl], key, "key must decrypt correctly");
            let val0 = e.data[16 + kl];
            assert_eq!(val0 as u64, e.gsn, "value must decrypt to the right entry");
            got.push(e.gsn);
        }

        assert_eq!(
            got,
            vec![1, 2, 3, 10, 11, 12],
            "encrypted retained-handle merge must emit every entry once, GSN-sorted"
        );
    }

    /// Regression: `ShardLogReader::new` must PROPAGATE a non-`NotFound` open
    /// error (EMFILE/EIO/EACCES) instead of swallowing it. The old
    /// `Err(_) => continue` silently dropped an unopenable file — and with it
    /// every later file — from the merge, so a follower would diverge
    /// permanently under fd exhaustion during a full sync. Here we make one data
    /// file unreadable (`chmod 000` → EACCES) and require construction to fail.
    /// A `NotFound` (a file that vanished to compaction) is still skipped — that
    /// path is exercised by every other test's normal operation and kept by the
    /// explicit `NotFound => continue` arm.
    #[cfg(unix)]
    #[test]
    fn new_propagates_non_notfound_open_error() {
        use std::os::unix::fs::PermissionsExt;

        let dir = tempfile::tempdir().unwrap();
        let shard_dir = dir.path().to_path_buf();

        // A valid file plus an unreadable one (lower id opens first, then the
        // bad one must abort — proving a *later* file's error isn't swallowed).
        write_data_file(&shard_dir, 0, &[(1, b"okkeyyyy", b"okvalyyy")]);
        let bad = shard_dir.join("000001.data");
        std::fs::write(&bad, b"unreadable").unwrap();
        std::fs::set_permissions(&bad, std::fs::Permissions::from_mode(0o000)).unwrap();

        // If the file is still openable (e.g. the test runs as root), this
        // environment can't produce the EACCES we need — skip rather than
        // false-fail.
        if std::fs::File::open(&bad).is_ok() {
            let _ = std::fs::set_permissions(&bad, std::fs::Permissions::from_mode(0o644));
            return;
        }

        let result = reader_from_dir(
            shard_dir,
            0,
            8,
            #[cfg(feature = "encryption")]
            None,
        );

        // Restore perms so tempdir cleanup can remove the file.
        let _ = std::fs::set_permissions(&bad, std::fs::Permissions::from_mode(0o644));

        assert!(
            result.is_err(),
            "a non-NotFound open error (EACCES) must propagate, not be swallowed"
        );
    }

    /// Verify that ShardLogReader correctly decrypts entries written by an
    /// encrypted Shard, producing the same key+value bytes as were written.
    #[cfg(all(feature = "encryption", feature = "replication"))]
    #[test]
    fn encrypted_log_reader_decrypts_entries() {
        use std::sync::{Arc, atomic::AtomicU64};

        use crate::crypto::PageCipher;
        use crate::shard::Shard;

        let dir = tempfile::tempdir().unwrap();
        let gsn = Arc::new(AtomicU64::new(0));
        let raw_cipher = PageCipher::new(&[0xAB; 32]).expect("create cipher");
        let cipher = Arc::new(raw_cipher);

        // Open an encrypted shard. write_buf_size must be >= 4096 so at least
        // one complete page can be flushed.
        let shard = Shard::open_encrypted(
            0,
            dir.path(),
            1 << 20,
            64 * 1024,
            false,
            false,
            false,
            crate::config::IoBackend::default(),
            Some(cipher.clone()),
            gsn,
        )
        .expect("open encrypted shard");

        // Build a handful of serialized entries and append them via
        // `append_raw_entry` (the replication ingestion path).
        let key_len: u16 = 8;
        let key = b"testkey0";
        let value1 = b"value001";
        let value2 = b"value002";
        let value3 = b"value003";

        let entry1 = serialize_entry(1, key, value1, false);
        let entry2 = serialize_entry(2, key, value2, false);
        let entry3 = serialize_entry(3, key, value3, false);

        {
            let mut inner = shard.lock();
            inner
                .append_raw_entry(0, key_len, &entry1)
                .expect("append entry 1");
            inner
                .append_raw_entry(0, key_len, &entry2)
                .expect("append entry 2");
            inner
                .append_raw_entry(0, key_len, &entry3)
                .expect("append entry 3");
        }

        // flush_buf uses flush_write_buf which only flushes complete 4096-byte
        // pages. Use flush() (flush_write_buf_final + fsync) to ensure all
        // entries land on disk even if the buffer hasn't reached a full page.
        shard.flush().expect("flush shard to disk");

        // Now construct a ShardLogReader with the same cipher.
        let shard_dir = dir.path().to_path_buf();
        let mut reader =
            reader_from_dir(shard_dir, 0, key_len, Some(cipher)).expect("create log reader");

        // Collect all entries and verify.
        let mut entries = Vec::new();
        while let Some(e) = reader.next_entry().expect("read entry") {
            entries.push(e);
        }

        assert_eq!(
            entries.len(),
            3,
            "expected 3 entries, got {}",
            entries.len()
        );

        // Verify GSNs are in order.
        assert_eq!(entries[0].gsn, 1);
        assert_eq!(entries[1].gsn, 2);
        assert_eq!(entries[2].gsn, 3);

        // Verify key bytes in each entry (bytes 16..16+key_len).
        let kl = key_len as usize;
        for e in &entries {
            assert_eq!(&e.data[16..16 + kl], key, "key mismatch in entry {}", e.gsn);
        }

        // Verify value bytes in each entry.
        for (i, (e, expected_val)) in entries.iter().zip([value1, value2, value3]).enumerate() {
            let val_start = 16 + kl;
            let val_end = val_start + expected_val.len();
            assert_eq!(
                &e.data[val_start..val_end],
                expected_val,
                "value mismatch in entry {}",
                i + 1
            );
        }
    }

    /// Encrypted variant of `log_reader_skips_mid_file_padding`: a page-aligned
    /// encrypted shard pads the partial page on `flush()`, so appending more
    /// entries afterwards leaves mid-file padding whose on-disk bytes are the
    /// *ciphertext* of zeros. The reader must decrypt through
    /// `fill_read_buf_encrypted`, recognise the zero padding, and resume at the
    /// next real entry rather than emitting a bogus zero entry.
    #[cfg(all(feature = "encryption", feature = "replication"))]
    #[test]
    fn encrypted_log_reader_skips_mid_file_padding() {
        use std::sync::{Arc, atomic::AtomicU64};

        use crate::crypto::PageCipher;
        use crate::shard::Shard;

        let dir = tempfile::tempdir().unwrap();
        let gsn = Arc::new(AtomicU64::new(0));
        let cipher = Arc::new(PageCipher::new(&[0x5C; 32]).expect("create cipher"));

        let shard = Shard::open_encrypted(
            0,
            dir.path(),
            1 << 20,
            64 * 1024,
            false,
            false,
            false,
            crate::config::IoBackend::default(),
            Some(cipher.clone()),
            gsn,
        )
        .expect("open encrypted shard");

        let key_len: u16 = 8;
        let key1 = b"padkey01";
        let key2 = b"padkey02";
        let entry1 = serialize_entry(1, key1, b"padval01", false);
        let entry2 = serialize_entry(2, key2, b"padval02", false);

        // Append e1, force-flush (pads page 0 to the boundary), then append e2
        // and flush again → e2 lands at the next page boundary with mid-file
        // padding between the two entries.
        {
            let mut inner = shard.lock();
            inner
                .append_raw_entry(0, key_len, &entry1)
                .expect("append e1");
        }
        shard.flush().expect("flush after e1");
        {
            let mut inner = shard.lock();
            inner
                .append_raw_entry(0, key_len, &entry2)
                .expect("append e2");
        }
        shard.flush().expect("flush after e2");

        let mut reader = reader_from_dir(dir.path().to_path_buf(), 0, key_len, Some(cipher))
            .expect("create encrypted log reader");

        let kl = key_len as usize;
        let e1 = reader.next_entry().unwrap().expect("entry gsn 1");
        assert_eq!(e1.gsn, 1);
        assert_eq!(e1.file_offset, 0);
        assert_eq!(&e1.data[16..16 + kl], key1);

        let e2 = reader
            .next_entry()
            .unwrap()
            .expect("entry gsn 2 after padding");
        assert_eq!(e2.gsn, 2);
        assert!(
            e2.file_offset > 0 && e2.file_offset.is_multiple_of(4096),
            "gsn 2 must start at a page boundary after padding, got {}",
            e2.file_offset
        );
        assert_eq!(&e2.data[16..16 + kl], key2);

        assert!(
            reader.next_entry().unwrap().is_none(),
            "no bogus zero entry from padding"
        );
    }

    /// Regression test for the `peek_bytes_from` alignment bug in the
    /// encrypted refill path.
    ///
    /// `fill_read_buf_encrypted` sets `read_buf_offset` to the 4096-byte page
    /// boundary at-or-below the requested offset. The old post-fill fallback
    /// in `peek_bytes_from` returned `read_buf[..len]` — i.e. data starting
    /// at `read_buf_offset`, not at the requested `offset`. This silently
    /// returned wrong bytes for any entry whose file offset was not
    /// page-aligned and which fell into a refill after the first one.
    ///
    /// The fix is to re-check the cache-hit slice with
    /// `start = offset - read_buf_offset`. This test writes enough entries
    /// to exceed READ_AHEAD_SIZE (64 KiB), forcing a second refill at a
    /// non-page-aligned offset. Before the fix, entries past the 64 KiB
    /// boundary either fail CRC or return wrong GSN bytes.
    #[cfg(all(feature = "encryption", feature = "replication"))]
    #[test]
    fn encrypted_log_reader_handles_non_page_aligned_refill() {
        use std::sync::{Arc, atomic::AtomicU64};

        use crate::crypto::PageCipher;
        use crate::shard::Shard;

        let dir = tempfile::tempdir().unwrap();
        let gsn = Arc::new(AtomicU64::new(0));
        let cipher = Arc::new(PageCipher::new(&[0xCD; 32]).expect("create cipher"));

        // max_file_size >> 64 KiB so all entries land in a single data file
        // and the reader has to refill at least twice from it.
        let shard = Shard::open_encrypted(
            0,
            dir.path(),
            1 << 24,    // 16 MiB
            256 * 1024, // 256 KiB write buffer
            false,
            false,
            false,
            crate::config::IoBackend::default(),
            Some(cipher.clone()),
            gsn,
        )
        .expect("open encrypted shard");

        let key_len: u16 = 8;
        let key: &[u8; 8] = b"regrkey0";
        // value_len = 8 → entry_size(8, 8) = (16 + 8 + 8) rounded up to 8 = 32 bytes
        // 32 bytes does divide 4096, so let's use a non-divisor value length.
        // entry_size(8, 16) = (16 + 8 + 16) = 40, padded to 40 — also a divisor of 8.
        // We want the entry size to NOT divide 4096 so successive entry offsets
        // drift away from page boundaries.  Use value_len = 21 →
        // entry_size(8, 21) = (16 + 8 + 21) = 45 rounded up to 48.
        // 4096 / 48 = 85.33 — not integral, good.
        let value_template: [u8; 21] = [0x77; 21];

        // Need total on-disk bytes > READ_AHEAD_SIZE (64 KiB = 65536).
        // 65536 / 48 ≈ 1366.  Use 2000 entries to be comfortably past the
        // first refill window.
        let num_entries: u64 = 2000;

        {
            let mut inner = shard.lock();
            for i in 1..=num_entries {
                let mut value = value_template;
                // Encode the GSN into the first 8 bytes of the value so we can
                // detect mis-aligned reads by content (the bug would surface
                // as a value bytestring shifted by the page-alignment delta).
                value[0..8].copy_from_slice(&i.to_le_bytes());
                let entry = serialize_entry(i, key, &value, false);
                inner
                    .append_raw_entry(0, key_len, &entry)
                    .expect("append entry");
            }
        }

        shard.flush().expect("flush shard to disk");

        let shard_dir = dir.path().to_path_buf();
        let mut reader = reader_from_dir(shard_dir, 0, key_len, Some(cipher))
            .expect("create encrypted log reader");

        let kl = key_len as usize;
        let mut count: u64 = 0;
        while let Some(e) = reader.next_entry().expect("read entry") {
            count += 1;
            // Expect entries in the order they were written, GSN 1..=num_entries.
            assert_eq!(e.gsn, count, "entry #{count} has wrong gsn {}", e.gsn);

            // Verify key matches.
            assert_eq!(&e.data[16..16 + kl], key, "key mismatch in entry {count}");

            // Verify value's first 8 bytes encode the GSN (i.e. the bytes are
            // not shifted by a 4096-alignment delta).
            let val_start = 16 + kl;
            let recovered_gsn = u64::from_le_bytes(
                e.data[val_start..val_start + 8]
                    .try_into()
                    .expect("8 bytes"),
            );
            assert_eq!(
                recovered_gsn, count,
                "value at entry #{count} carries wrong embedded gsn {recovered_gsn}"
            );
        }

        assert_eq!(
            count, num_entries,
            "expected {num_entries} entries, decoded {count}"
        );
    }
}