armdb 0.7.0

use std::fs;
use std::hash::Hash;
use std::path::Path;

use zerocopy::FromBytes;

use crate::Key;

use crate::const_map::MapEntry;
use crate::disk_loc::DiskLoc;
use crate::entry::{EntryHeader, compute_crc32, entry_size};
use crate::error::{DbError, DbResult};
use crate::hint;
use crate::io::direct;
use crate::map_index::ShardIndex;
use crate::skiplist::SkipList;
#[cfg(feature = "var-collections")]
use crate::skiplist::node::VarNode;
use crate::skiplist::node::{ConstNode, SkipNode, random_height};
use crate::sync::{self, Mutex};

#[cfg(feature = "typed-tree")]
use crate::codec::Codec;
#[cfg(feature = "typed-tree")]
use crate::skiplist::node::{TypedData, TypedNode};

#[cfg(feature = "encryption")]
use std::sync::Arc;

#[cfg(feature = "encryption")]
use crate::crypto::PageCipher;

#[cfg(feature = "encryption")]
use crate::io::tags::{self, TagFile};

// ---------------------------------------------------------------------------
// Reader helper — dispatches to plain or encrypted pread_value
// ---------------------------------------------------------------------------

type ReadFn = dyn Fn(&std::fs::File, u64, usize) -> DbResult<Vec<u8>>;

#[cfg(test)]
static FULL_SCAN_PROBE: std::sync::Mutex<
    Option<(
        std::path::PathBuf,
        std::sync::Arc<std::sync::atomic::AtomicUsize>,
    )>,
> = std::sync::Mutex::new(None);

#[cfg(test)]
struct FullScanProbeGuard {
    path: std::path::PathBuf,
}

#[cfg(test)]
impl Drop for FullScanProbeGuard {
    fn drop(&mut self) {
        let mut probe = FULL_SCAN_PROBE
            .lock()
            .expect("full scan probe mutex poisoned");
        if probe.as_ref().is_some_and(|(path, _)| path == &self.path) {
            *probe = None;
        }
    }
}

#[cfg(test)]
fn observe_full_scans_for(
    path: std::path::PathBuf,
    counter: std::sync::Arc<std::sync::atomic::AtomicUsize>,
) -> FullScanProbeGuard {
    let mut probe = FULL_SCAN_PROBE
        .lock()
        .expect("full scan probe mutex poisoned");
    assert!(probe.is_none(), "a full scan probe is already registered");
    *probe = Some((path.clone(), counter));
    FullScanProbeGuard { path }
}

fn plain_reader() -> Box<ReadFn> {
    Box::new(direct::pread_value)
}

#[cfg(feature = "encryption")]
fn encrypted_reader(
    cipher: &Arc<PageCipher>,
    tag_file: &Arc<TagFile>,
    file_id: u32,
) -> Box<ReadFn> {
    let cipher = cipher.clone();
    let tag_file = tag_file.clone();
    Box::new(move |file, offset, len| {
        direct::pread_value_encrypted(file, &tag_file, &cipher, file_id, offset, len)
    })
}

#[cfg(feature = "encryption")]
fn make_reader(
    cipher: &Option<Arc<PageCipher>>,
    dir: &Path,
    file_id: u32,
) -> DbResult<Box<ReadFn>> {
    if let Some(cipher) = cipher {
        let tp = tags::tags_path_for_data(&dir.join(format!("{file_id:06}.data")));
        if tp.exists() {
            let tag_file = Arc::new(TagFile::open_read(&tp)?);
            return Ok(encrypted_reader(cipher, &tag_file, file_id));
        }
        // Tag file missing for an encrypted database — data cannot be decrypted.
        // Return an error instead of silently reading encrypted bytes as plaintext.
        return Err(crate::error::DbError::EncryptionError(format!(
            "tag file missing for encrypted data file {file_id:06}.data"
        )));
    }
    Ok(plain_reader())
}

#[cfg(not(feature = "encryption"))]
fn make_reader(_dir: &Path, _file_id: u32) -> DbResult<Box<ReadFn>> {
    Ok(plain_reader())
}

// ---------------------------------------------------------------------------
// RecoveryOutcome types
// ---------------------------------------------------------------------------

/// Outcome of a recovery pass over all shards.
pub(crate) struct RecoveryOutcome {
    /// Maximum GSN seen across all shards.
    pub max_gsn: u64,
    /// One entry per shard whose active file needs tail adjustment.
    pub active_tails: Vec<ActiveTail>,
    /// Per-shard dead_bytes maps: (shard_idx, HashMap<file_id, dead_bytes>).
    pub shard_dead_bytes: Vec<(usize, DeadBytesMap)>,
    /// Per-shard maximum **durable** GSN recovered from disk: `(shard_idx,
    /// max_gsn)`. The replication follower clamps its reconnect `from_gsn` to
    /// this so a cursor fsynced ahead of the shard's flushed data (power loss
    /// between cursor save and shard flush) refetches the lost tail (F3).
    #[cfg(feature = "replication")]
    pub shard_max_gsns: Vec<(usize, u64)>,
}

/// Describes the tail of the active (last) data file in one shard.
pub(crate) struct ActiveTail {
    /// Index into the `shards` slice.
    pub shard_idx: usize,
    /// File ID of the active data file.
    pub file_id: u32,
    /// Last offset at which a valid (or zero-padding) boundary was found.
    pub last_valid_offset: u64,
}

// ---------------------------------------------------------------------------
// Per-key sequence tracker helpers
// ---------------------------------------------------------------------------

type RecoveryTracker = std::collections::HashMap<Vec<u8>, (u64, u32, u64)>;

/// Per-shard dead-bytes map: file_id -> dead bytes.
type DeadBytesMap = std::collections::HashMap<u32, u64>;

/// Return type for per-shard recovery threads.
type ShardRecoveryResult = DbResult<(u64, Option<ActiveTail>, usize, DeadBytesMap)>;

/// Update the per-key tracker and dead_bytes accounting.
///
/// Returns `true` if `entry` is newer and should be applied to the index.
/// Returns `false` if `entry` is stale (already superseded); the caller
/// should skip it.
fn tracker_update(
    tracker: &mut RecoveryTracker,
    dead_bytes: &mut DeadBytesMap,
    key_bytes: &[u8],
    seq: u64,
    file_id: u32,
    entry_total_size: u64,
) -> bool {
    if let Some(val) = tracker.get_mut(key_bytes) {
        let (prev_seq, prev_file_id, prev_entry_size) = *val;
        if seq <= prev_seq {
            *dead_bytes.entry(file_id).or_insert(0) += entry_total_size;
            return false;
        }
        *dead_bytes.entry(prev_file_id).or_insert(0) += prev_entry_size;
        *val = (seq, file_id, entry_total_size);
        return true;
    }
    tracker.insert(key_bytes.to_vec(), (seq, file_id, entry_total_size));
    true
}

// ---------------------------------------------------------------------------
// Shared recovery core
//
// A single per-shard orchestrator drives every collection type. The parts that
// vary between collections are provided as closures:
//   * `apply`         — how a validated entry mutates that collection's index;
//   * `make_read_fn`  — how to build the (plain or decrypting) value reader.
// The parts that are identical — file discovery, the hint fast-path with its
// validation (R2/R3/R4), the full-scan fallback, active-tail capture, and the
// durable removal of the active file's replayed hint (R1) — live here once.
// ---------------------------------------------------------------------------

/// A validated recovery entry ready to be applied to an index.
struct Recovered<'a, K: Key> {
    key: K,
    /// Value bytes; empty for a tombstone or a value-less (Var) collection.
    value: &'a [u8],
    disk: DiskLoc,
    tombstone: bool,
}

/// Whether the hint fast-path applied the file or the caller must full-scan it.
enum HintReplay {
    Applied,
    FullScan,
}

/// Fold the per-shard thread results into a single [`RecoveryOutcome`].
fn fold_shard_results(results: Vec<ShardRecoveryResult>) -> DbResult<RecoveryOutcome> {
    let mut max_gsn: u64 = 0;
    let mut active_tails: Vec<ActiveTail> = Vec::new();
    let mut shard_dead_bytes: Vec<(usize, DeadBytesMap)> = Vec::new();
    #[cfg(feature = "replication")]
    let mut shard_max_gsns: Vec<(usize, u64)> = Vec::new();
    for result in results {
        let (shard_gsn, tail, shard_idx, dead) = result?;
        if shard_gsn > max_gsn {
            max_gsn = shard_gsn;
        }
        #[cfg(feature = "replication")]
        shard_max_gsns.push((shard_idx, shard_gsn));
        if let Some(t) = tail {
            active_tails.push(t);
        }
        if !dead.is_empty() {
            shard_dead_bytes.push((shard_idx, dead));
        }
    }
    Ok(RecoveryOutcome {
        max_gsn,
        active_tails,
        shard_dead_bytes,
        #[cfg(feature = "replication")]
        shard_max_gsns,
    })
}

/// Durably remove the active (last) file's hint after recovery replayed it.
///
/// The active file's hint is consumed on open: it is replayed and then must be
/// unlinked so a later crash cannot resurrect a stale hint that omits entries
/// appended after this open (R1). POSIX does not make an unlink durable until
/// the containing directory is fsynced, so the shard directory is fsynced after
/// removal. A non-`NotFound` removal error is propagated rather than ignored.
fn remove_active_hint(dir: &Path, file_ids: &[u32]) -> DbResult<()> {
    let Some(&last_id) = file_ids.last() else {
        return Ok(());
    };
    let path = dir.join(format!("{last_id:06}.hint"));
    match fs::remove_file(&path) {
        Ok(()) => direct::sync_dir(dir),
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
        Err(e) => Err(DbError::Io(e)),
    }
}

/// Drive recovery of every data file in one shard directory.
///
/// For each `.data` file: replay its hint via [`hint_replay`] when present and
/// valid, otherwise [`full_scan_walk`] the file. Each surviving entry is handed
/// to `apply`. Returns `(max_gsn, active_tail, dead_bytes, used_hints)`.
#[allow(clippy::too_many_arguments)]
fn recover_one_shard_files<K, MkRead, F>(
    shard_idx: usize,
    shard_id: u8,
    dir: &Path,
    hints: bool,
    reads_values: bool,
    make_read_fn: MkRead,
    apply: &mut F,
) -> DbResult<(u64, Option<ActiveTail>, DeadBytesMap, bool)>
where
    K: Key,
    MkRead: Fn(u32) -> DbResult<Box<ReadFn>>,
    F: FnMut(Recovered<'_, K>) -> DbResult<()>,
{
    let mut local_max_gsn: u64 = 0;
    let mut used_hints = false;
    let mut active_tail: Option<ActiveTail> = None;
    let file_ids = scan_data_files(dir)?;
    let hint_ids = if hints {
        hint::scan_hint_files(dir)?
    } else {
        Vec::new()
    };
    let mut tracker: RecoveryTracker = std::collections::HashMap::new();
    let mut dead_bytes: DeadBytesMap = std::collections::HashMap::new();

    for file_id in &file_ids {
        let data_path = dir.join(format!("{file_id:06}.data"));
        let is_active = Some(file_id) == file_ids.last();
        let read_fn = make_read_fn(*file_id)?;

        let mut applied_via_hint = false;
        if hints && hint_ids.contains(file_id) {
            let hint_path = dir.join(format!("{file_id:06}.hint"));
            if let Some(hint_data) = hint::read_hint_file(&hint_path)? {
                let data_file = direct::open_bulk_read(&data_path)?;
                let file_len = data_file.metadata()?.len();
                match hint_replay::<K, F>(
                    *file_id,
                    &hint_data,
                    &data_file,
                    file_len,
                    reads_values,
                    &mut local_max_gsn,
                    &read_fn,
                    &mut tracker,
                    &mut dead_bytes,
                    apply,
                )? {
                    HintReplay::Applied => {
                        direct::fadvise_dontneed(&data_file, 0, file_len);
                        used_hints = true;
                        applied_via_hint = true;
                    }
                    HintReplay::FullScan => {}
                }
            }
        }

        if !applied_via_hint {
            let last_valid = full_scan_walk::<K, F>(
                shard_id,
                *file_id,
                &data_path,
                &mut local_max_gsn,
                &read_fn,
                &mut tracker,
                &mut dead_bytes,
                apply,
            )?;

            if hints && !is_active {
                let data_file = direct::open_bulk_read(&data_path)?;
                let hint_data = hint::generate_hint_data_with_reader(
                    &data_file,
                    last_valid,
                    size_of::<K>(),
                    read_fn.as_ref(),
                )?;
                let hint_path = hint::hint_path_for_data(&data_path);
                hint::write_hint_file(&hint_path, &hint_data)?;
                direct::fadvise_dontneed(&data_file, 0, last_valid);
            }

            if is_active {
                active_tail = Some(ActiveTail {
                    shard_idx,
                    file_id: *file_id,
                    last_valid_offset: last_valid,
                });
            }
        }
    }

    remove_active_hint(dir, &file_ids)?;

    Ok((local_max_gsn, active_tail, dead_bytes, used_hints))
}

// ---------------------------------------------------------------------------
// Hint replay (fast path) — two-phase validate-then-apply
// ---------------------------------------------------------------------------

/// Replay a hint file, validating it fully before mutating any recovery state.
///
/// Hints are regenerable metadata: a corrupt, truncated, or out-of-bounds hint
/// must degrade to a full scan, never fail `open()`. All validation therefore
/// happens in phase 1 **before** phase 2 touches `tracker`/`dead_bytes`/`index`/
/// `max_gsn`; on any problem the function returns [`HintReplay::FullScan`] with
/// no partial state applied. Applying partially and then re-scanning would let
/// `tracker_update` re-see the applied entries on the `seq <= prev_seq` branch
/// and double-count them as dead bytes — pre-validation avoids that.
///
/// Validation covers:
///   * R3 — reject an old-format/corrupt hint (no v3 magic), a wrong-sized one,
///     or one whose header `logical_end` is inconsistent with its entries;
///   * P1-1 — using the v3 header's `logical_end`, prove the region past the last
///     hint entry is genuine page padding. This single check catches BOTH a
///     stale/lagging hint (real entries appended past `logical_end`) AND a
///     truncated hint (real entries in `[covered_end, logical_end)` that the hint
///     dropped): either leaves non-padding bytes in `[covered_end, file_len)`, so
///     the file falls back to a full scan instead of silently reverting them;
///   * R4 / P1-2 — reject any entry whose value region falls past the data file
///     end, or whose offset cannot fit a `u32` `DiskLoc` (all range math is
///     `checked_add`);
///   * R2 — for value-bearing (const/typed) collections, read each value back
///     and verify its CRC (`reads_values`); tombstone CRCs are always verified
///     against an empty value, while live Var entries remain bounds-only (P2-1).
///
/// P1-3 (bounded RAM): phase 1 reads and CRC-checks every value but retains only
/// entry *metadata* (never the value bytes); phase 2 re-reads each applied value
/// from disk. Peak RAM is therefore O(1) values per shard, not O(file), while the
/// "fully validate before mutating any recovery state" guarantee is preserved.
#[allow(clippy::too_many_arguments)]
fn hint_replay<K: Key, F: FnMut(Recovered<'_, K>) -> DbResult<()>>(
    file_id: u32,
    hint_data: &[u8],
    data_file: &std::fs::File,
    file_len: u64,
    reads_values: bool,
    max_gsn: &mut u64,
    read_fn: &ReadFn,
    tracker: &mut RecoveryTracker,
    dead_bytes: &mut DeadBytesMap,
    apply: &mut F,
) -> DbResult<HintReplay> {
    // R3: fully validate the v3 header and payload metadata. A missing/unknown
    // magic or corrupt hint degrades to a full scan and is regenerated.
    let Some((header, _entries_bytes)) = hint::validate_hint_file(hint_data, size_of::<K>()) else {
        tracing::warn!(
            file_id,
            "hint file missing/invalid v3 metadata, falling back to full scan"
        );
        return Ok(HintReplay::FullScan);
    };

    let entries: Vec<hint::HintEntry<K>> = hint::parse_hint_entries::<K>(hint_data).collect();

    // `covered_end` (C) = the byte just past the last hint entry's value, or 0
    // when the hint has no entries. All range math is checked (P1-2): a corrupt
    // offset/len must never wrap a u64.
    let covered_end: u64 = match entries.last() {
        None => 0,
        Some(last) => match last.value_offset.checked_add(last.value_len as u64) {
            Some(c) => c,
            None => {
                tracing::warn!(
                    file_id,
                    "hint entry range overflow, falling back to full scan"
                );
                return Ok(HintReplay::FullScan);
            }
        },
    };

    // P1-1: the header must satisfy `C <= logical_end <= file_len`. A hint that
    // claims a logical end below its own coverage, or beyond the physical file,
    // is internally inconsistent (stale, truncated, or corrupt) → full scan.
    if !(covered_end <= header.logical_end && header.logical_end <= file_len) {
        tracing::warn!(
            file_id,
            covered_end,
            logical_end = header.logical_end,
            file_len,
            "hint header inconsistent (need C <= logical_end <= file_len), falling back to full scan"
        );
        return Ok(HintReplay::FullScan);
    }

    // P1-1: everything in `[covered_end, file_len)` must be genuine page padding.
    // A stale/lagging hint leaves real appended entries past `logical_end`; a
    // truncated hint leaves real entries in `[covered_end, logical_end)`. Both
    // show up here as non-zero (or, encrypted, non-zero plaintext) bytes and force
    // a full scan, so their newest values/tombstones are never silently dropped.
    // Only genuine trailing page padding (all zeros) passes. The walk is
    // page-bounded and short-circuits on the first non-padding byte (P1-3).
    if !tail_is_all_padding(read_fn, data_file, covered_end, file_len)? {
        tracing::warn!(
            file_id,
            covered_end,
            logical_end = header.logical_end,
            file_len,
            "hint tail is not page padding (stale or truncated hint), falling back to full scan"
        );
        return Ok(HintReplay::FullScan);
    }

    // Phase 1 — validate every entry with no mutation of recovery state. Values
    // are read and CRC-checked but NOT retained (P1-3): only entry metadata is
    // kept, bounding peak RAM to O(entries) small structs rather than O(file).
    let mut validated: Vec<hint::HintEntry<K>> = Vec::with_capacity(entries.len());
    for entry in &entries {
        // P1-2: the value offset must fit a u32 `DiskLoc` field. A valid data file
        // is <= 4 GiB, so this rejects only a corrupt hint.
        if entry.value_offset > u32::MAX as u64 {
            tracing::warn!(
                file_id,
                value_offset = entry.value_offset,
                "hint entry offset exceeds u32 DiskLoc range, falling back to full scan"
            );
            return Ok(HintReplay::FullScan);
        }
        // R4: the value region must lie within the data file (checked add).
        let end = match entry.value_offset.checked_add(entry.value_len as u64) {
            Some(e) => e,
            None => {
                tracing::warn!(
                    file_id,
                    "hint entry range overflow, falling back to full scan"
                );
                return Ok(HintReplay::FullScan);
            }
        };
        if end > file_len {
            tracing::warn!(
                file_id,
                value_offset = entry.value_offset,
                value_len = entry.value_len,
                file_len,
                "hint entry past end of data file, falling back to full scan"
            );
            return Ok(HintReplay::FullScan);
        }

        if entry.is_tombstone() && entry.value_len != 0 {
            tracing::warn!(
                file_id,
                value_len = entry.value_len,
                "hint tombstone has a non-empty value, falling back to full scan"
            );
            return Ok(HintReplay::FullScan);
        }

        // Every zero-length entry has all CRC inputs available without reading
        // the data file. Validate the raw GSN as well as the key so flipping the
        // tombstone bit in either direction cannot bypass the Var fast path.
        if entry.value_len == 0 {
            let expected_crc = compute_crc32(entry.gsn, 0, entry.key.as_bytes(), &[]);
            if expected_crc != entry.crc32 {
                tracing::warn!(
                    file_id,
                    expected = expected_crc,
                    actual = entry.crc32,
                    "hint zero-length entry CRC mismatch, falling back to full scan"
                );
                return Ok(HintReplay::FullScan);
            }
            validated.push(*entry);
            continue;
        }

        // P2-1: a Var collection (`reads_values == false`) stores only a `DiskLoc`
        // and never reads a live value, so it cannot recompute that value's CRC.
        // Tombstones are safe because their empty-value CRC is checked above.
        if !reads_values {
            validated.push(*entry);
            continue;
        }

        // R2: read the value back and verify its CRC, then DISCARD it (P1-3). A
        // corrupt hint offset/len or CRC surfaces here and degrades to a full scan.
        let value_bytes = match read_fn(data_file, entry.value_offset, entry.value_len as usize) {
            Ok(b) => b,
            Err(e) => {
                tracing::warn!(
                    file_id,
                    err = ?e,
                    "hint value read failed, falling back to full scan"
                );
                return Ok(HintReplay::FullScan);
            }
        };
        let expected_crc = compute_crc32(
            entry.gsn,
            entry.value_len,
            entry.key.as_bytes(),
            &value_bytes,
        );
        if expected_crc != entry.crc32 {
            tracing::warn!(
                file_id,
                expected = expected_crc,
                actual = entry.crc32,
                "hint value CRC mismatch, falling back to full scan"
            );
            return Ok(HintReplay::FullScan);
        }
        validated.push(*entry);
    }

    // Phase 2 — apply. Everything validated, so mutating state is now safe. Each
    // surviving value is re-read from disk just before it is applied (P1-3);
    // phase 1 already proved the read succeeds and the CRC matches, so a failure
    // here is a genuine I/O error and propagates (matching the full-scan path).
    for entry in &validated {
        let seq = entry.sequence();
        if seq > *max_gsn {
            *max_gsn = seq;
        }
        let total = entry_size(size_of::<K>(), entry.value_len);
        if !tracker_update(
            tracker,
            dead_bytes,
            entry.key.as_bytes(),
            seq,
            file_id,
            total,
        ) {
            continue;
        }
        let value_bytes: Vec<u8> = if entry.is_tombstone() || !reads_values {
            Vec::new()
        } else {
            // Phase 1 already read this exact range and matched its CRC, so a
            // failure here is a genuine (persistent) I/O error, not a bad hint:
            // propagate it and fail `open()`, matching `full_scan_walk`. A
            // transient re-read fault failing open (rather than degrading to a
            // full scan) is an accepted, extremely narrow trade-off of P1-3's
            // bounded two-pass replay — the file was just read successfully.
            read_fn(data_file, entry.value_offset, entry.value_len as usize)?
        };
        let disk = DiskLoc::new(file_id, entry.value_offset as u32, entry.value_len);
        apply(Recovered {
            key: entry.key,
            value: &value_bytes,
            disk,
            tombstone: entry.is_tombstone(),
        })?;
    }
    Ok(HintReplay::Applied)
}

/// Verify that every byte of `[start, file_len)` is page padding — zero in plain
/// mode, or decrypting to zero in encrypted mode (the reader handles both).
///
/// The scan is page-bounded: it reads at most one page at a time and
/// short-circuits on the first non-zero (or short-read) byte, so a hint whose
/// tail hides real entries returns `false` after a single page read rather than
/// buffering the whole tail (P1-3). A page-aligned `start == file_len` (empty
/// range) is vacuously padding.
fn tail_is_all_padding(
    read_fn: &ReadFn,
    file: &std::fs::File,
    start: u64,
    file_len: u64,
) -> DbResult<bool> {
    const PAGE_SIZE: u64 = 4096;
    let mut pos = start;
    while pos < file_len {
        // Read up to the next page boundary (never crossing it in one read) so a
        // multi-page tail is checked one bounded page at a time.
        let next_boundary = (pos / PAGE_SIZE + 1) * PAGE_SIZE;
        let end = next_boundary.min(file_len);
        let len = (end - pos) as usize;
        match read_fn(file, pos, len) {
            Ok(bytes) => {
                if bytes.len() < len || bytes.iter().any(|&b| b != 0) {
                    // A short read (can't confirm padding) or any real byte means
                    // the tail is not clean padding.
                    return Ok(false);
                }
            }
            // An unreadable tail cannot be confirmed as padding → full scan.
            Err(_) => return Ok(false),
        }
        pos = end;
    }
    Ok(true)
}

// ---------------------------------------------------------------------------
// Full data scan (fallback path)
// ---------------------------------------------------------------------------

/// Scan every entry in a data file, validating headers/CRC and handing each
/// surviving (newest) entry to `apply`. Returns the last valid offset (used to
/// set the active file's write tail).
#[allow(clippy::too_many_arguments)]
fn full_scan_walk<K: Key, F: FnMut(Recovered<'_, K>) -> DbResult<()>>(
    shard_id: u8,
    file_id: u32,
    data_path: &Path,
    max_gsn: &mut u64,
    read_fn: &ReadFn,
    tracker: &mut RecoveryTracker,
    dead_bytes: &mut DeadBytesMap,
    apply: &mut F,
) -> DbResult<u64> {
    #[cfg(test)]
    {
        use std::sync::atomic::Ordering;

        let probe = FULL_SCAN_PROBE
            .lock()
            .expect("full scan probe mutex poisoned");
        if let Some((target, counter)) = probe.as_ref()
            && target == data_path
        {
            counter.fetch_add(1, Ordering::Relaxed);
        }
    }

    let file = direct::open_bulk_read(data_path)?;
    let file_len = file.metadata()?.len();
    let mut offset: u64 = 0;

    while offset + size_of::<EntryHeader>() as u64 <= file_len {
        let header_bytes = match read_fn(&file, offset, size_of::<EntryHeader>()) {
            Ok(b) => b,
            Err(e) => {
                if matches!(&e, DbError::Io(io) if io.kind() == std::io::ErrorKind::UnexpectedEof) {
                    break;
                }
                return Err(e);
            }
        };
        let header = match EntryHeader::read_from_bytes(&header_bytes) {
            Ok(h) => h,
            Err(_) => break,
        };

        // Page padding sentinel: force-flush pads the partial page with zeros.
        // If we see an all-zero header AND the rest of the current page is also
        // zero, that's clean padding — skip to the next page boundary.
        if header.gsn == 0
            && header.value_len == 0
            && header.crc32 == 0
            && crate::entry::check_page_padding_at(
                &file,
                offset + size_of::<EntryHeader>() as u64,
                read_fn,
            )?
        {
            const PAGE_SIZE: u64 = 4096;
            let next_page =
                (offset + size_of::<EntryHeader>() as u64).div_ceil(PAGE_SIZE) * PAGE_SIZE;
            if next_page >= file_len {
                break;
            }
            offset = next_page;
            continue;
        }

        let total = entry_size(size_of::<K>(), header.value_len);
        if offset + total > file_len {
            tracing::warn!(
                shard_id,
                file_id,
                offset,
                "truncating partial entry at end of file"
            );
            break;
        }

        let key_value_len = size_of::<K>() + header.value_len as usize;
        let key_value_bytes = read_fn(
            &file,
            offset + size_of::<EntryHeader>() as u64,
            key_value_len,
        )?;

        let key_bytes = &key_value_bytes[..size_of::<K>()];
        let value_bytes = &key_value_bytes[size_of::<K>()..];

        let expected_crc = compute_crc32(header.gsn, header.value_len, key_bytes, value_bytes);
        if expected_crc != header.crc32 {
            if offset + total > file_len {
                tracing::warn!(shard_id, file_id, offset, "CRC mismatch at end of file");
                break;
            }
            return Err(DbError::CrcMismatch {
                expected: expected_crc,
                actual: header.crc32,
            });
        }

        let seq = header.sequence();
        if seq > *max_gsn {
            *max_gsn = seq;
        }

        if !tracker_update(tracker, dead_bytes, key_bytes, seq, file_id, total) {
            offset += total;
            continue;
        }

        let key: K = K::from_bytes(key_bytes);
        let value_offset = offset + size_of::<EntryHeader>() as u64 + size_of::<K>() as u64;
        let disk = DiskLoc::new(file_id, value_offset as u32, header.value_len);

        apply(Recovered {
            key,
            value: value_bytes,
            disk,
            tombstone: header.is_tombstone(),
        })?;

        offset += total;
    }
    direct::fadvise_dontneed(&file, 0, file_len);
    Ok(offset)
}

// ---------------------------------------------------------------------------
// Public entry points — thin per-collection adapters over the shared core
// ---------------------------------------------------------------------------

/// Recover a ConstTree index by scanning all data files in each shard.
/// Uses hint files when available, falls back to full data scan otherwise.
/// Returns a [`RecoveryOutcome`] with the maximum GSN and per-shard active tails.
pub fn recover_const_tree<K: Key, const V: usize>(
    shard_dirs: &[&Path],
    shard_ids: &[u8],
    index: &SkipList<ConstNode<K, V>>,
    hints: bool,
    #[cfg(feature = "encryption")] ciphers: Option<Arc<Vec<Arc<PageCipher>>>>,
) -> DbResult<RecoveryOutcome> {
    let results: Vec<ShardRecoveryResult> = std::thread::scope(|s| {
        let handles: Vec<_> = shard_dirs
            .iter()
            .enumerate()
            .map(|(i, &dir)| {
                let shard_id = shard_ids[i];
                #[cfg(feature = "encryption")]
                let cipher = ciphers.as_ref().map(|c| c[i].clone());
                s.spawn(move || -> ShardRecoveryResult {
                    let shard_start = std::time::Instant::now();
                    let guard = index.collector().enter();
                    let mut apply = |r: Recovered<'_, K>| -> DbResult<()> {
                        if r.tombstone {
                            index.remove(r.key.as_bytes(), &guard);
                        } else {
                            let value: [u8; V] = r.value.try_into().map_err(|_| {
                                DbError::CorruptedEntry {
                                    offset: r.disk.offset as u64,
                                }
                            })?;
                            let node_ptr = ConstNode::alloc(r.key, value, r.disk, random_height());
                            match index.insert(node_ptr, &guard) {
                                crate::skiplist::InsertResult::Inserted => {}
                                crate::skiplist::InsertResult::Exists(existing) => {
                                    existing.write_data(r.disk, &value);
                                    unsafe {
                                        ConstNode::<K, V>::dealloc_node(node_ptr);
                                    }
                                }
                            }
                        }
                        Ok(())
                    };
                    #[cfg(feature = "encryption")]
                    let make_read_fn = |fid: u32| make_reader(&cipher, dir, fid);
                    #[cfg(not(feature = "encryption"))]
                    let make_read_fn = |fid: u32| make_reader(dir, fid);
                    let (max_gsn, tail, dead, used_hints) = recover_one_shard_files::<K, _, _>(
                        i, shard_id, dir, hints, true, make_read_fn, &mut apply,
                    )?;
                    let elapsed = shard_start.elapsed().as_secs_f64();
                    metrics::histogram!("armdb.recovery.duration_seconds", "path" => if used_hints { "hint" } else { "full_scan" })
                        .record(elapsed);
                    Ok((max_gsn, tail, i, dead))
                })
            })
            .collect();

        handles
            .into_iter()
            .map(|h| match h.join() {
                Ok(r) => r,
                Err(_) => Err(DbError::RecoveryPanic),
            })
            .collect()
    });

    fold_shard_results(results)
}

#[cfg(feature = "var-collections")]
/// Recover a VarTree index by scanning all data files in each shard.
pub fn recover_var_tree<K: Key>(
    shard_dirs: &[&Path],
    shard_ids: &[u8],
    index: &SkipList<VarNode<K>>,
    hints: bool,
    #[cfg(feature = "encryption")] ciphers: Option<Arc<Vec<Arc<PageCipher>>>>,
) -> DbResult<RecoveryOutcome> {
    let results: Vec<ShardRecoveryResult> = std::thread::scope(|s| {
        let handles: Vec<_> = shard_dirs
            .iter()
            .enumerate()
            .map(|(i, &dir)| {
                let shard_id = shard_ids[i];
                #[cfg(feature = "encryption")]
                let cipher = ciphers.as_ref().map(|c| c[i].clone());
                s.spawn(move || -> ShardRecoveryResult {
                    let shard_start = std::time::Instant::now();
                    let guard = index.collector().enter();
                    let mut apply = |r: Recovered<'_, K>| -> DbResult<()> {
                        if r.tombstone {
                            index.remove(r.key.as_bytes(), &guard);
                        } else {
                            let node_ptr = VarNode::alloc(r.key, r.disk, random_height());
                            match index.insert(node_ptr, &guard) {
                                crate::skiplist::InsertResult::Inserted => {}
                                crate::skiplist::InsertResult::Exists(existing) => {
                                    // Recovery is single-threaded and runs before the index is
                                    // published to concurrent readers, so write_loc here is the
                                    // only writer (no shard mutex).
                                    existing.write_loc(r.disk);
                                    unsafe {
                                        VarNode::<K>::dealloc_node(node_ptr);
                                    }
                                }
                            }
                        }
                        Ok(())
                    };
                    #[cfg(feature = "encryption")]
                    let make_read_fn = |fid: u32| make_reader(&cipher, dir, fid);
                    #[cfg(not(feature = "encryption"))]
                    let make_read_fn = |fid: u32| make_reader(dir, fid);
                    let (max_gsn, tail, dead, used_hints) = recover_one_shard_files::<K, _, _>(
                        i, shard_id, dir, hints, false, make_read_fn, &mut apply,
                    )?;
                    let elapsed = shard_start.elapsed().as_secs_f64();
                    metrics::histogram!("armdb.recovery.duration_seconds", "path" => if used_hints { "hint" } else { "full_scan" })
                        .record(elapsed);
                    Ok((max_gsn, tail, i, dead))
                })
            })
            .collect();

        handles
            .into_iter()
            .map(|h| match h.join() {
                Ok(r) => r,
                Err(_) => Err(DbError::RecoveryPanic),
            })
            .collect()
    });

    fold_shard_results(results)
}

/// Recover a TypedTree index by scanning all data files in each shard.
/// Uses hint files when available, falls back to full data scan otherwise.
/// Returns a [`RecoveryOutcome`] with the maximum GSN and per-shard active tails.
#[cfg(feature = "typed-tree")]
pub fn recover_typed_tree<K: Key, T: Send + Sync, C: Codec<T> + Sync>(
    shard_dirs: &[&Path],
    shard_ids: &[u8],
    index: &SkipList<TypedNode<K, T>>,
    codec: &C,
    hints: bool,
    #[cfg(feature = "encryption")] ciphers: Option<Arc<Vec<Arc<PageCipher>>>>,
) -> DbResult<RecoveryOutcome> {
    let results: Vec<ShardRecoveryResult> = std::thread::scope(|s| {
        let handles: Vec<_> = shard_dirs
            .iter()
            .enumerate()
            .map(|(i, &dir)| {
                let shard_id = shard_ids[i];
                #[cfg(feature = "encryption")]
                let cipher = ciphers.as_ref().map(|c| c[i].clone());
                s.spawn(move || -> ShardRecoveryResult {
                    let shard_start = std::time::Instant::now();
                    let guard = index.collector().enter();
                    let mut apply = |r: Recovered<'_, K>| -> DbResult<()> {
                        if r.tombstone {
                            index.remove(r.key.as_bytes(), &guard);
                        } else {
                            let value: T = codec.decode_from(r.value).map_err(|_| {
                                DbError::CorruptedEntry {
                                    offset: r.disk.offset as u64,
                                }
                            })?;
                            let node_ptr = TypedNode::alloc(r.key, value, r.disk, random_height());
                            match index.insert(node_ptr, &guard) {
                                crate::skiplist::InsertResult::Inserted => {}
                                crate::skiplist::InsertResult::Exists(existing) => unsafe {
                                    let new_data = (*node_ptr).data.swap(
                                        std::ptr::null_mut(),
                                        std::sync::atomic::Ordering::Relaxed,
                                    );
                                    let old_data = existing.swap_data(new_data);
                                    drop(Box::from_raw(old_data));
                                    TypedNode::<K, T>::dealloc_node(node_ptr);
                                },
                            }
                        }
                        Ok(())
                    };
                    #[cfg(feature = "encryption")]
                    let make_read_fn = |fid: u32| make_reader(&cipher, dir, fid);
                    #[cfg(not(feature = "encryption"))]
                    let make_read_fn = |fid: u32| make_reader(dir, fid);
                    let (max_gsn, tail, dead, used_hints) = recover_one_shard_files::<K, _, _>(
                        i, shard_id, dir, hints, true, make_read_fn, &mut apply,
                    )?;
                    let elapsed = shard_start.elapsed().as_secs_f64();
                    metrics::histogram!("armdb.recovery.duration_seconds", "path" => if used_hints { "hint" } else { "full_scan" })
                        .record(elapsed);
                    Ok((max_gsn, tail, i, dead))
                })
            })
            .collect();

        handles
            .into_iter()
            .map(|h| match h.join() {
                Ok(r) => r,
                Err(_) => Err(DbError::RecoveryPanic),
            })
            .collect()
    });

    fold_shard_results(results)
}

// ---------------------------------------------------------------------------
// ConstMap / VarMap / TypedMap recovery
// ---------------------------------------------------------------------------

pub fn recover_const_map<K: Key + Send + Sync + Hash + Eq, const V: usize>(
    shard_dirs: &[&Path],
    shard_ids: &[u8],
    indexes: &[Mutex<ShardIndex<K, MapEntry<V, DiskLoc>>>],
    hints: bool,
    #[cfg(feature = "encryption")] ciphers: Option<Arc<Vec<Arc<PageCipher>>>>,
) -> DbResult<RecoveryOutcome> {
    let results: Vec<ShardRecoveryResult> = std::thread::scope(|s| {
        let handles: Vec<_> = shard_dirs
            .iter()
            .enumerate()
            .map(|(i, &dir)| {
                let shard_id = shard_ids[i];
                let index = &indexes[i];
                #[cfg(feature = "encryption")]
                let cipher = ciphers.as_ref().map(|c| c[i].clone());
                s.spawn(move || -> ShardRecoveryResult {
                    let shard_start = std::time::Instant::now();
                    let mut map = sync::lock(index);
                    let mut apply = |r: Recovered<'_, K>| -> DbResult<()> {
                        if r.tombstone {
                            map.remove(&r.key);
                        } else {
                            let value: [u8; V] = r.value.try_into().map_err(|_| {
                                DbError::CorruptedEntry {
                                    offset: r.disk.offset as u64,
                                }
                            })?;
                            map.upsert(r.key, MapEntry { loc: r.disk, value });
                        }
                        Ok(())
                    };
                    #[cfg(feature = "encryption")]
                    let make_read_fn = |fid: u32| make_reader(&cipher, dir, fid);
                    #[cfg(not(feature = "encryption"))]
                    let make_read_fn = |fid: u32| make_reader(dir, fid);
                    let (max_gsn, tail, dead, used_hints) = recover_one_shard_files::<K, _, _>(
                        i, shard_id, dir, hints, true, make_read_fn, &mut apply,
                    )?;
                    let elapsed = shard_start.elapsed().as_secs_f64();
                    metrics::histogram!("armdb.recovery.duration_seconds", "path" => if used_hints { "hint" } else { "full_scan" })
                        .record(elapsed);
                    Ok((max_gsn, tail, i, dead))
                })
            })
            .collect();

        handles
            .into_iter()
            .map(|h| match h.join() {
                Ok(r) => r,
                Err(_) => Err(DbError::RecoveryPanic),
            })
            .collect()
    });

    fold_shard_results(results)
}

#[cfg(feature = "var-collections")]
pub fn recover_var_map<K: Key + Send + Sync + Hash + Eq>(
    shard_dirs: &[&Path],
    shard_ids: &[u8],
    indexes: &[Mutex<ShardIndex<K, DiskLoc>>],
    hints: bool,
    #[cfg(feature = "encryption")] ciphers: Option<Arc<Vec<Arc<PageCipher>>>>,
) -> DbResult<RecoveryOutcome> {
    let results: Vec<ShardRecoveryResult> = std::thread::scope(|s| {
        let handles: Vec<_> = shard_dirs
            .iter()
            .enumerate()
            .map(|(i, &dir)| {
                let shard_id = shard_ids[i];
                let index = &indexes[i];
                #[cfg(feature = "encryption")]
                let cipher = ciphers.as_ref().map(|c| c[i].clone());
                s.spawn(move || -> ShardRecoveryResult {
                    let shard_start = std::time::Instant::now();
                    let mut map = sync::lock(index);
                    let mut apply = |r: Recovered<'_, K>| -> DbResult<()> {
                        if r.tombstone {
                            map.remove(&r.key);
                        } else {
                            map.upsert(r.key, r.disk);
                        }
                        Ok(())
                    };
                    #[cfg(feature = "encryption")]
                    let make_read_fn = |fid: u32| make_reader(&cipher, dir, fid);
                    #[cfg(not(feature = "encryption"))]
                    let make_read_fn = |fid: u32| make_reader(dir, fid);
                    let (max_gsn, tail, dead, used_hints) = recover_one_shard_files::<K, _, _>(
                        i, shard_id, dir, hints, false, make_read_fn, &mut apply,
                    )?;
                    let elapsed = shard_start.elapsed().as_secs_f64();
                    metrics::histogram!("armdb.recovery.duration_seconds", "path" => if used_hints { "hint" } else { "full_scan" })
                        .record(elapsed);
                    Ok((max_gsn, tail, i, dead))
                })
            })
            .collect();

        handles
            .into_iter()
            .map(|h| match h.join() {
                Ok(r) => r,
                Err(_) => Err(DbError::RecoveryPanic),
            })
            .collect()
    });

    fold_shard_results(results)
}

#[cfg(feature = "typed-tree")]
pub fn recover_typed_map<K: Key + Send + Sync + Hash + Eq, T: Send + Sync, C: Codec<T> + Sync>(
    shard_dirs: &[&Path],
    shard_ids: &[u8],
    indexes: &[Mutex<ShardIndex<K, crate::typed_map::TypedMapEntry<T>>>],
    codec: &C,
    hints: bool,
    #[cfg(feature = "encryption")] ciphers: Option<Arc<Vec<Arc<PageCipher>>>>,
) -> DbResult<RecoveryOutcome> {
    let results: Vec<ShardRecoveryResult> = std::thread::scope(|s| {
        let handles: Vec<_> = shard_dirs
            .iter()
            .enumerate()
            .map(|(i, &dir)| {
                let shard_id = shard_ids[i];
                let index = &indexes[i];
                #[cfg(feature = "encryption")]
                let cipher = ciphers.as_ref().map(|c| c[i].clone());
                s.spawn(move || -> ShardRecoveryResult {
                    let shard_start = std::time::Instant::now();
                    let mut map = sync::lock(index);
                    let mut apply = |r: Recovered<'_, K>| -> DbResult<()> {
                        if r.tombstone {
                            if let Some(old) = map.remove(&r.key) {
                                unsafe {
                                    drop(Box::from_raw(old.ptr));
                                }
                            }
                        } else {
                            let value = codec.decode_from(r.value).map_err(|_| {
                                DbError::CorruptedEntry {
                                    offset: r.disk.offset as u64,
                                }
                            })?;
                            let ptr = Box::into_raw(Box::new(TypedData {
                                disk: r.disk,
                                value,
                            }));
                            if let Some(old) =
                                map.upsert(r.key, crate::typed_map::TypedMapEntry { ptr })
                            {
                                unsafe {
                                    drop(Box::from_raw(old.ptr));
                                }
                            }
                        }
                        Ok(())
                    };
                    #[cfg(feature = "encryption")]
                    let make_read_fn = |fid: u32| make_reader(&cipher, dir, fid);
                    #[cfg(not(feature = "encryption"))]
                    let make_read_fn = |fid: u32| make_reader(dir, fid);
                    let (max_gsn, tail, dead, used_hints) = recover_one_shard_files::<K, _, _>(
                        i, shard_id, dir, hints, true, make_read_fn, &mut apply,
                    )?;
                    let elapsed = shard_start.elapsed().as_secs_f64();
                    metrics::histogram!("armdb.recovery.duration_seconds", "path" => if used_hints { "hint" } else { "full_scan" })
                        .record(elapsed);
                    Ok((max_gsn, tail, i, dead))
                })
            })
            .collect();

        handles
            .into_iter()
            .map(|h| match h.join() {
                Ok(r) => r,
                Err(_) => Err(DbError::RecoveryPanic),
            })
            .collect()
    });

    fold_shard_results(results)
}

// ---------------------------------------------------------------------------
// Utilities
// ---------------------------------------------------------------------------

/// Scan a directory for `.data` files and return sorted file IDs.
fn scan_data_files(dir: &Path) -> DbResult<Vec<u32>> {
    let mut ids = Vec::new();
    if !dir.exists() {
        return Ok(ids);
    }
    for entry in fs::read_dir(dir)? {
        let entry = entry?;
        let name = entry.file_name();
        let name = name.to_string_lossy();
        if name.ends_with(".data")
            && let Ok(id) = name.trim_end_matches(".data").parse::<u32>()
        {
            ids.push(id);
        }
    }
    ids.sort();
    Ok(ids)
}

#[cfg(test)]
mod tests {
    use crate::entry::check_page_padding_at;

    #[test]
    fn page_padding_detected_for_zero_tail() {
        let dir = std::env::temp_dir().join(format!("armdb_pad_{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        let path = dir.join("p.data");

        // One page: 100 bytes of 0xAB, rest zero (padding).
        let mut page = vec![0u8; 4096];
        for b in page[..100].iter_mut() {
            *b = 0xAB;
        }
        let f = crate::io::direct::open_bulk_write(&path).unwrap();
        crate::io::direct::pwrite_at(&f, &page, 0).unwrap();
        crate::io::direct::fsync(&f).unwrap();
        drop(f);

        let r = crate::io::direct::open_bulk_read(&path).unwrap();
        let read = |file: &std::fs::File, off: u64, len: usize| {
            crate::io::direct::pread_value(file, off, len)
        };
        // From offset 100 to page end is all zero → padding.
        assert!(check_page_padding_at(&r, 100, &read).unwrap());
        // From offset 50 it includes 0xAB → not padding.
        assert!(!check_page_padding_at(&r, 50, &read).unwrap());

        std::fs::remove_dir_all(&dir).ok();
    }
}

/// v3 hint-format regressions for the four hint-path findings (P1-1/P1-2/P1-3).
/// These are white-box: they open a real collection to lay down data + a v3 hint,
/// then poke the on-disk bytes and drive recovery through a reopen.
#[cfg(test)]
mod hint_v3_regression {
    use crate::config::Config;
    use crate::const_tree::ConstTree;
    use crate::entry::{TOMBSTONE_BIT, compute_crc32, serialize_entry};
    use crate::error::DbError;
    use crate::hint::{
        HINT_HEADER_SIZE, fail_hint_write_for, hint_entry_size, hint_file_bytes, read_hint_header,
        validate_hint_file,
    };
    #[cfg(feature = "var-collections")]
    use crate::var_tree::VarTree;
    use std::fs;
    use std::path::{Path, PathBuf};
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use tempfile::tempdir;

    fn one_shard() -> Config {
        let mut c = Config::test();
        c.shard_count = 1;
        c
    }

    fn shard0(root: &Path) -> PathBuf {
        root.join("shard_000")
    }

    /// Append one hint entry `GSN(8) | Key(8) | Offset(8) | Len(4) | CRC(4)` to a
    /// raw (headerless) entries buffer.
    fn push_entry(buf: &mut Vec<u8>, gsn: u64, key: &[u8; 8], off: u64, len: u32, crc: u32) {
        buf.extend_from_slice(&gsn.to_ne_bytes());
        buf.extend_from_slice(key);
        buf.extend_from_slice(&off.to_ne_bytes());
        buf.extend_from_slice(&len.to_ne_bytes());
        buf.extend_from_slice(&crc.to_ne_bytes());
    }

    #[cfg(feature = "var-collections")]
    fn rotated_var_config() -> Config {
        let mut config = one_shard();
        config.max_file_size = 8192;
        config.write_buffer_size = 8192;
        config
    }

    #[test]
    #[cfg(feature = "var-collections")]
    fn hint_rewrite_forged_low_max_is_repaired_and_second_reopen_fast_paths() {
        let dir = tempdir().unwrap();
        let config = rotated_var_config();

        {
            let tree = VarTree::<[u8; 8]>::open(dir.path(), config.clone()).unwrap();
            for key in 0u64..300 {
                tree.put(&key.to_be_bytes(), &[0x5A; 64]).unwrap();
            }
            tree.close().unwrap();
        }

        let data_path = shard0(dir.path()).join("000001.data");
        let hint_path = shard0(dir.path()).join("000001.hint");
        let mut forged = fs::read(&hint_path).unwrap();
        let actual_max = read_hint_header(&hint_path).unwrap().unwrap().max_gsn;
        assert!(
            actual_max > 1,
            "test requires a hint with a max GSN above one"
        );
        forged[16..24].copy_from_slice(&1u64.to_le_bytes());
        let header_crc = crc32fast::hash(&forged[..32]);
        forged[32..36].copy_from_slice(&header_crc.to_le_bytes());
        fs::write(&hint_path, forged).unwrap();
        assert_eq!(read_hint_header(&hint_path).unwrap().unwrap().max_gsn, 1);

        let scans = Arc::new(AtomicUsize::new(0));
        let _probe = super::observe_full_scans_for(data_path, scans.clone());
        {
            let tree = VarTree::<[u8; 8]>::open(dir.path(), config.clone()).unwrap();
            assert_eq!(tree.len(), 300);
            let repaired = fs::read(&hint_path).unwrap();
            let (header, _) = validate_hint_file(&repaired, 8)
                .expect("the rewritten hint must pass full validation");
            assert_eq!(header.max_gsn, actual_max);
            assert_eq!(
                read_hint_header(&hint_path).unwrap().unwrap().max_gsn,
                actual_max
            );
            tree.close().unwrap();
        }
        assert_eq!(scans.load(Ordering::Relaxed), 1);

        scans.store(0, Ordering::Relaxed);
        {
            let tree = VarTree::<[u8; 8]>::open(dir.path(), config).unwrap();
            assert_eq!(tree.len(), 300);
            tree.close().unwrap();
        }
        assert_eq!(
            scans.load(Ordering::Relaxed),
            0,
            "the repaired immutable hint must avoid a second full scan"
        );
    }

    #[test]
    #[cfg(feature = "var-collections")]
    fn hint_rewrite_failure_is_fatal() {
        let dir = tempdir().unwrap();
        let config = rotated_var_config();

        {
            let tree = VarTree::<[u8; 8]>::open(dir.path(), config.clone()).unwrap();
            for key in 0u64..300 {
                tree.put(&key.to_be_bytes(), &[0x5A; 64]).unwrap();
            }
            tree.close().unwrap();
        }

        let hint_path = shard0(dir.path()).join("000001.hint");
        let mut hint = fs::read(&hint_path).unwrap();
        hint[..4].copy_from_slice(b"AHF2");
        fs::write(&hint_path, hint).unwrap();

        let _fault = fail_hint_write_for(hint_path.clone(), std::io::Error::other("test: ENOSPC"));
        match VarTree::<[u8; 8]>::open(dir.path(), config) {
            Err(DbError::Io(error)) => assert!(error.to_string().contains("test: ENOSPC")),
            Err(error) => panic!("expected injected I/O error, got {error}"),
            Ok(_) => panic!("hint rewrite failure must fail open"),
        }
        assert_eq!(&fs::read(hint_path).unwrap()[..4], b"AHF2");
    }

    // -----------------------------------------------------------------------
    // P1-1 — a stale/lagging active-file hint outrun by a *sub-page* durable
    // append must not silently drop the appended entry. The gap is one 32-byte
    // entry (< 4096), which the old `file_len - covered_end < PAGE_SIZE`
    // heuristic waved through as "page padding". The v3 tail-padding check sees
    // the real appended bytes and forces a full scan.
    //
    // FAILS WITHOUT FIX: the old heuristic applies the stale hint and the
    // appended key is missing after reopen.
    // -----------------------------------------------------------------------
    #[test]
    fn p1_1_stale_hint_subpage_append_is_recovered() {
        let dir = tempdir().unwrap();
        let config = one_shard();

        // Write 10 entries and close → an active hint covering exactly them.
        {
            let tree = ConstTree::<[u8; 8], 8>::open(dir.path(), config.clone()).unwrap();
            for i in 0u64..10 {
                tree.put(&i.to_be_bytes(), &(i * 10).to_be_bytes()).unwrap();
            }
            tree.close().unwrap();
        }

        // Durably append ONE more entry (< 4096 B) past the hint's coverage.
        let data_path = shard0(dir.path()).join("000001.data");
        let appended = serialize_entry(100, &100u64.to_be_bytes(), &4242u64.to_be_bytes(), false);
        let mut bytes = fs::read(&data_path).unwrap();
        bytes.extend_from_slice(&appended);
        fs::write(&data_path, &bytes).unwrap();

        // Reopen: the appended entry must survive (tail is not padding → full scan).
        let tree = ConstTree::<[u8; 8], 8>::open(dir.path(), config).unwrap();
        assert_eq!(
            tree.len(),
            11,
            "stale sub-page hint dropped the appended entry"
        );
        assert_eq!(
            tree.get(&100u64.to_be_bytes()),
            Some(4242u64.to_be_bytes()),
            "the durably-appended key was silently dropped (P1-1)"
        );
        for i in 0u64..10 {
            assert_eq!(tree.get(&i.to_be_bytes()), Some((i * 10).to_be_bytes()));
        }
    }

    // -----------------------------------------------------------------------
    // P1-1 — a hint truncated by exactly ONE entry (a sub-page gap) over an
    // intact data file must fall back to a full scan, not revert the dropped
    // entry. This is the discriminating case the old heuristic missed: the gap
    // is 32 bytes (< PAGE_SIZE), so the old code treated it as padding.
    //
    // FAILS WITHOUT FIX: the last key is missing after reopen.
    // -----------------------------------------------------------------------
    #[test]
    fn p1_1_truncated_hint_single_entry_is_recovered() {
        let dir = tempdir().unwrap();
        let config = one_shard();

        {
            let tree = ConstTree::<[u8; 8], 8>::open(dir.path(), config.clone()).unwrap();
            for i in 0u64..10 {
                tree.put(&i.to_be_bytes(), &(i * 10).to_be_bytes()).unwrap();
            }
            tree.close().unwrap();
        }

        // Drop the last hint entry (32 bytes), keeping the header and the data
        // file intact.
        let hint_path = shard0(dir.path()).join("000001.hint");
        let hint = fs::read(&hint_path).unwrap();
        let entry_sz = hint_entry_size(8);
        assert_eq!(hint.len(), HINT_HEADER_SIZE + 10 * entry_sz);
        fs::write(&hint_path, &hint[..hint.len() - entry_sz]).unwrap();

        let tree = ConstTree::<[u8; 8], 8>::open(dir.path(), config).unwrap();
        assert_eq!(tree.len(), 10, "truncated hint dropped an entry");
        for i in 0u64..10 {
            assert_eq!(
                tree.get(&i.to_be_bytes()),
                Some((i * 10).to_be_bytes()),
                "key {i} lost after single-entry hint truncation (P1-1)"
            );
        }
    }

    // -----------------------------------------------------------------------
    // P1-1 — a clean shutdown whose active file ends with multi-page zero
    // padding in `[logical_end, file_len)` must STILL apply via the hint fast
    // path, not spuriously full-scan. We prove the fast path was taken by
    // planting a data entry whose *header* CRC is corrupt (a full scan would
    // error on it) while the hint carries the correct CRC — so a successful
    // open can only have come from the hint.
    //
    // FAILS WITHOUT FIX: the old heuristic sees a >= PAGE_SIZE tail gap, full
    // scans, hits the corrupt header CRC, and `open()` errors.
    // -----------------------------------------------------------------------
    #[test]
    fn p1_1_clean_multipage_padding_still_fast_paths() {
        let dir = tempdir().unwrap();
        let config = one_shard();

        // Create the db structure (db.meta, salt, shard_000/000001.data).
        {
            let tree = ConstTree::<[u8; 8], 8>::open(dir.path(), config.clone()).unwrap();
            drop(tree);
        }

        let key_a = 1u64.to_be_bytes();
        let key_b = 2u64.to_be_bytes();
        let val_a = [0xA1u8; 8];
        let val_b = [0xB2u8; 8];
        let ea = serialize_entry(1, &key_a, &val_a, false); // 32B @0
        let mut eb = serialize_entry(2, &key_b, &val_b, false); // 32B @32
        // Corrupt B's header CRC (bytes 8..12 of the entry) — a full scan errors
        // here; the hint path never consults the header CRC.
        eb[8] ^= 0xFF;

        let mut file_bytes = Vec::new();
        file_bytes.extend_from_slice(&ea);
        file_bytes.extend_from_slice(&eb);
        let logical_end = file_bytes.len() as u64; // 64
        file_bytes.resize(8192, 0); // two clean pages of trailing padding
        let data_path = shard0(dir.path()).join("000001.data");
        fs::write(&data_path, &file_bytes).unwrap();

        // Hint lists A and B with correct CRCs over their real values.
        let mut entries = Vec::new();
        push_entry(
            &mut entries,
            1,
            &key_a,
            24,
            8,
            compute_crc32(1, 8, &key_a, &val_a),
        );
        push_entry(
            &mut entries,
            2,
            &key_b,
            56,
            8,
            compute_crc32(2, 8, &key_b, &val_b),
        );
        let hint = hint_file_bytes(&entries, logical_end, 8).unwrap();
        fs::write(shard0(dir.path()).join("000001.hint"), &hint).unwrap();

        // Reopen: must apply via the hint (clean padding recognised), NOT error.
        let tree = ConstTree::<[u8; 8], 8>::open(dir.path(), config)
            .expect("clean multi-page padding must fast-path via the hint, not full-scan");
        assert_eq!(tree.len(), 2);
        assert_eq!(tree.get(&key_a), Some(val_a));
        assert_eq!(tree.get(&key_b), Some(val_b));
    }

    // -----------------------------------------------------------------------
    // P1-3 — the bounded two-pass hint replay must recover byte-identically to
    // a full scan over the same data, including superseded (overwritten) and
    // deleted keys. Behaviour-preserving guard for the phase1-discard /
    // phase2-reread change.
    // -----------------------------------------------------------------------
    #[test]
    fn p1_3_two_pass_matches_full_scan() {
        let dir = tempdir().unwrap();
        let config = one_shard();

        {
            let tree = ConstTree::<[u8; 8], 8>::open(dir.path(), config.clone()).unwrap();
            for i in 0u64..60 {
                tree.put(&i.to_be_bytes(), &(i * 10).to_be_bytes()).unwrap();
            }
            for i in 0u64..30 {
                tree.put(&i.to_be_bytes(), &(i * 100).to_be_bytes())
                    .unwrap();
            }
            tree.delete(&45u64.to_be_bytes()).unwrap();
            tree.close().unwrap();
        }

        let expect = |i: u64| -> Option<[u8; 8]> {
            if i == 45 {
                None
            } else if i < 30 {
                Some((i * 100).to_be_bytes())
            } else {
                Some((i * 10).to_be_bytes())
            }
        };

        // Recover via the HINT fast path (hint present) and snapshot.
        let via_hint: Vec<Option<[u8; 8]>> = {
            let tree = ConstTree::<[u8; 8], 8>::open(dir.path(), config.clone()).unwrap();
            let snap = (0u64..60).map(|i| tree.get(&i.to_be_bytes())).collect();
            tree.close().unwrap();
            snap
        };

        // Force a full scan by removing every hint, then snapshot.
        for e in fs::read_dir(shard0(dir.path())).unwrap().flatten() {
            if e.path().extension().and_then(|x| x.to_str()) == Some("hint") {
                fs::remove_file(e.path()).unwrap();
            }
        }
        let via_scan: Vec<Option<[u8; 8]>> = {
            let tree = ConstTree::<[u8; 8], 8>::open(dir.path(), config).unwrap();
            (0u64..60).map(|i| tree.get(&i.to_be_bytes())).collect()
        };

        for i in 0u64..60 {
            assert_eq!(via_hint[i as usize], expect(i), "hint path wrong at {i}");
            assert_eq!(
                via_hint[i as usize], via_scan[i as usize],
                "hint replay diverged from full scan at key {i} (P1-3)"
            );
        }
    }

    #[test]
    fn hint_v3_live_entry_tombstone_flag_corruption_full_scans() {
        let dir = tempdir().unwrap();
        let config = one_shard();
        let key = 7u64.to_be_bytes();
        let value = 70u64.to_be_bytes();

        {
            let tree = ConstTree::<[u8; 8], 8>::open(dir.path(), config.clone()).unwrap();
            tree.put(&key, &value).unwrap();
            tree.close().unwrap();
        }

        let hint_path = shard0(dir.path()).join("000001.hint");
        let mut hint = fs::read(&hint_path).unwrap();
        let gsn_offset = HINT_HEADER_SIZE;
        let gsn = u64::from_ne_bytes(hint[gsn_offset..gsn_offset + 8].try_into().unwrap());
        hint[gsn_offset..gsn_offset + 8].copy_from_slice(&(gsn | TOMBSTONE_BIT).to_ne_bytes());
        fs::write(&hint_path, hint).unwrap();

        let tree = ConstTree::<[u8; 8], 8>::open(dir.path(), config).unwrap();
        assert_eq!(
            tree.get(&key),
            Some(value),
            "a live entry with a corrupt tombstone flag must full-scan, not delete the key"
        );
    }

    #[test]
    #[cfg(feature = "var-collections")]
    fn hint_v3_genuine_tombstone_key_corruption_full_scans() {
        let dir = tempdir().unwrap();
        let config = one_shard();
        let key = 9u64.to_be_bytes();

        {
            let tree = VarTree::<[u8; 8]>::open(dir.path(), config.clone()).unwrap();
            tree.put(&key, b"value").unwrap();
            tree.delete(&key).unwrap();
            tree.close().unwrap();
        }

        let hint_path = shard0(dir.path()).join("000001.hint");
        let mut hint = fs::read(&hint_path).unwrap();
        let entry_sz = hint_entry_size(8);
        assert_eq!(hint.len(), HINT_HEADER_SIZE + 2 * entry_sz);
        let tombstone_key_offset = HINT_HEADER_SIZE + entry_sz + 8;
        hint[tombstone_key_offset] ^= 0xff;
        fs::write(&hint_path, hint).unwrap();

        let tree = VarTree::<[u8; 8]>::open(dir.path(), config).unwrap();
        assert_eq!(
            tree.get(&key),
            None,
            "a tombstone with a corrupt key must full-scan, not preserve the deleted key"
        );
    }

    #[test]
    #[cfg(feature = "var-collections")]
    fn hint_v3_cleared_tombstone_bit_corruption_full_scans() {
        let dir = tempdir().unwrap();
        let config = one_shard();
        let key = 10u64.to_be_bytes();

        {
            let tree = VarTree::<[u8; 8]>::open(dir.path(), config.clone()).unwrap();
            tree.put(&key, b"value").unwrap();
            tree.delete(&key).unwrap();
            tree.close().unwrap();
        }

        let hint_path = shard0(dir.path()).join("000001.hint");
        let mut hint = fs::read(&hint_path).unwrap();
        let entry_sz = hint_entry_size(8);
        assert_eq!(hint.len(), HINT_HEADER_SIZE + 2 * entry_sz);
        let tombstone_gsn_offset = HINT_HEADER_SIZE + entry_sz;
        let tombstone_gsn = u64::from_ne_bytes(
            hint[tombstone_gsn_offset..tombstone_gsn_offset + 8]
                .try_into()
                .unwrap(),
        );
        assert_ne!(tombstone_gsn & TOMBSTONE_BIT, 0);
        hint[tombstone_gsn_offset..tombstone_gsn_offset + 8]
            .copy_from_slice(&(tombstone_gsn & !TOMBSTONE_BIT).to_ne_bytes());
        fs::write(&hint_path, hint).unwrap();

        let tree = VarTree::<[u8; 8]>::open(dir.path(), config).unwrap();
        assert!(
            tree.get(&key).is_none(),
            "clearing a tombstone bit must full-scan, not resurrect an empty live value"
        );
    }

    #[test]
    #[cfg(feature = "var-collections")]
    fn hint_v3_zero_length_live_tombstone_bit_corruption_full_scans() {
        let dir = tempdir().unwrap();
        let config = one_shard();
        let key = 11u64.to_be_bytes();

        {
            let tree = VarTree::<[u8; 8]>::open(dir.path(), config.clone()).unwrap();
            tree.put(&key, b"").unwrap();
            tree.close().unwrap();
        }

        let hint_path = shard0(dir.path()).join("000001.hint");
        let mut hint = fs::read(&hint_path).unwrap();
        let gsn_offset = HINT_HEADER_SIZE;
        let gsn = u64::from_ne_bytes(hint[gsn_offset..gsn_offset + 8].try_into().unwrap());
        assert_eq!(gsn & TOMBSTONE_BIT, 0);
        hint[gsn_offset..gsn_offset + 8].copy_from_slice(&(gsn | TOMBSTONE_BIT).to_ne_bytes());
        fs::write(&hint_path, hint).unwrap();

        let tree = VarTree::<[u8; 8]>::open(dir.path(), config).unwrap();
        let value = tree
            .get(&key)
            .expect("zero-length live value must survive a corrupt tombstone bit");
        assert!(value.as_ref().is_empty());
    }
}

/// P1-2 — a corrupt hint entry with an out-of-range value offset must never
/// panic or truncate into a bad `DiskLoc`; it degrades to a full scan. Uses a
/// Var collection (`reads_values == false`), the path the old unchecked add
/// could bypass.
#[cfg(all(test, feature = "var-collections"))]
mod hint_v3_regression_var {
    use crate::config::Config;
    use crate::hint::{HINT_HEADER_SIZE, hint_entry_size};
    use crate::var_tree::VarTree;
    use std::fs;
    use tempfile::tempdir;

    // FAILS WITHOUT FIX: the old `covered_end = value_offset + value_len` add
    // overflows u64 and panics in debug builds (or, in release, truncates the
    // offset into a bad `DiskLoc`).
    #[test]
    fn p1_2_huge_hint_offset_no_panic_full_scan() {
        let dir = tempdir().unwrap();
        let mut config = Config::test();
        config.shard_count = 1;

        {
            let tree = VarTree::<[u8; 8]>::open(dir.path(), config.clone()).unwrap();
            for i in 0u64..4 {
                tree.put(&i.to_be_bytes(), format!("value_{i}").as_bytes())
                    .unwrap();
            }
            tree.close().unwrap();
        }

        // Set the last hint entry's value offset to u64::MAX so `offset + len`
        // overflows. Entry layout: GSN(8) | Key(8) | Offset(8) | Len(4) | CRC(4).
        let hint_path = dir.path().join("shard_000").join("000001.hint");
        let mut hint = fs::read(&hint_path).unwrap();
        let entry_sz = hint_entry_size(8);
        assert!(hint.len() >= HINT_HEADER_SIZE + entry_sz);
        let off_field = hint.len() - entry_sz + 16;
        hint[off_field..off_field + 8].copy_from_slice(&u64::MAX.to_ne_bytes());
        fs::write(&hint_path, &hint).unwrap();

        // Reopen must not panic and must recover every entry via a full scan.
        let tree = VarTree::<[u8; 8]>::open(dir.path(), config).unwrap();
        assert_eq!(tree.len(), 4, "overflow-offset hint must full-scan cleanly");
        for i in 0u64..4 {
            let v = tree
                .get(&i.to_be_bytes())
                .unwrap_or_else(|| panic!("key {i} unreadable after P1-2 recovery"));
            assert_eq!(v.as_ref(), format!("value_{i}").as_bytes());
        }
    }

    /// P1-2: an offset that is `> u32::MAX` but does NOT overflow `u64` must be
    /// rejected by the per-entry `value_offset > u32::MAX` guard specifically (a
    /// `DiskLoc` offset is u32; `offset as u32` would silently truncate to an
    /// out-of-bounds location). Corrupt a NON-last entry so the last-entry
    /// `covered_end <= logical_end` consistency check still passes — that routes
    /// the corrupt offset all the way to the per-entry guard, which is the one
    /// preventing the truncated `DiskLoc`, rather than tripping the header check.
    #[test]
    fn p1_2_offset_above_u32_full_scan_no_bad_diskloc() {
        let dir = tempdir().unwrap();
        let mut config = Config::test();
        config.shard_count = 1;

        {
            let tree = VarTree::<[u8; 8]>::open(dir.path(), config.clone()).unwrap();
            for i in 0u64..4 {
                tree.put(&i.to_be_bytes(), format!("value_{i}").as_bytes())
                    .unwrap();
            }
            tree.close().unwrap();
        }

        // Set the FIRST hint entry's value offset to 5_000_000_000 (> u32::MAX,
        // but offset + len stays well within u64). Entry layout:
        // GSN(8) | Key(8) | Offset(8) | Len(4) | CRC(4).
        let hint_path = dir.path().join("shard_000").join("000001.hint");
        let mut hint = fs::read(&hint_path).unwrap();
        let entry_sz = hint_entry_size(8);
        assert!(hint.len() >= HINT_HEADER_SIZE + 2 * entry_sz);
        let off_field = HINT_HEADER_SIZE + 16; // entry 0's Offset field
        hint[off_field..off_field + 8].copy_from_slice(&5_000_000_000u64.to_ne_bytes());
        fs::write(&hint_path, &hint).unwrap();

        // Reopen must not panic, must NOT install a truncated DiskLoc, and must
        // recover every entry via a full scan.
        let tree = VarTree::<[u8; 8]>::open(dir.path(), config).unwrap();
        assert_eq!(tree.len(), 4, ">u32 offset hint must full-scan cleanly");
        for i in 0u64..4 {
            let v = tree
                .get(&i.to_be_bytes())
                .unwrap_or_else(|| panic!("key {i} unreadable after P1-2 recovery"));
            assert_eq!(v.as_ref(), format!("value_{i}").as_bytes());
        }
    }
}