armdb 0.6.0

use std::fs::{self, File, OpenOptions};
use std::ops::Range;
use std::os::unix::fs::FileExt;
use std::path::PathBuf;
use std::time::{Duration, Instant};

use crate::error::{DbError, DbResult};
use crate::fixed::bitmap::Bitmap;
use crate::fixed::config::FixedConfig;
use crate::fixed::slot;

// ── Header layout ──────────────────────────────────────────────────
// [0..4]   magic: b"FIXD"
// [4..6]   version: u16 LE
// [6..8]   slot_size: u16 LE
// [8..12]  slot_count: u32 LE
// [12..14] key_len: u16 LE
// [14..16] value_len: u16 LE
// [16]     shard_id: u8
// [17]     clean_shutdown: u8 (0 or 1)
// [18..4096] reserved (zeros)

pub(crate) const HEADER_SIZE: u64 = 4096;
const MAGIC: &[u8; 4] = b"FIXD";
const VERSION: u16 = 2;

/// Offset of the `clean_shutdown` flag in the header.
const CLEAN_SHUTDOWN_OFFSET: u64 = 17;

/// Chunk size for sequential recovery reads. Not configurable.
pub(crate) const RECOVERY_CHUNK_BYTES: usize = 4 << 20; // 4 MiB
/// Two occupied runs whose byte gap is ≤ this are read as one range on the
/// clean path (readahead makes bridging small gaps cheaper than extra syscalls).
pub(crate) const RECOVERY_MERGE_GAP_BYTES: usize = 64 << 10; // 64 KiB

/// Which recovery scan to perform.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum RecoverMode {
    /// Full scan `0..slot_count`; `versions` filled from on-disk meta.
    Dirty,
    /// Sidecar already loaded into `versions`; scan only occupied ranges,
    /// `versions` is the source of truth for status.
    Clean,
}

/// Per-shard file I/O for fixed-slot storage.
///
/// Manages a single data file (`fixed.data`) and an optional versions sidecar
/// (`fixed.versions`) that is written on clean shutdown for fast restart.
pub struct FixedShardInner {
    file: File,
    dir: PathBuf,
    pub(crate) bitmap: Bitmap,
    /// Per-slot `meta` cache (4 bytes × slot_count).  Populated on open;
    /// kept in sync with disk on every write/delete.  Source of truth for
    /// replication modular version comparisons.
    pub(crate) versions: Vec<u32>,
    pub(crate) slot_size: u16,
    pub(crate) slot_count: u32,
    key_len: u16,
    value_len: u16,
    // pub(crate) because Task 9 reads this for the wrap-warning metric label.
    pub(crate) shard_id: u8,
    grow_step: u32,
    // fdatasync batching
    pending_writes: u32,
    sync_batch_size: u32,
    last_sync: Instant,
    sync_interval: Duration,
    enable_fsync: bool,
    #[cfg(feature = "replication")]
    pub(crate) replication_tx:
        Option<rtrb::Producer<crate::fixed_replication::FixedReplicationEvent>>,
}

impl FixedShardInner {
    /// Create or open a fixed-slot shard in the given directory.
    ///
    /// If `fixed.data` already exists the header is validated against the
    /// provided parameters; on mismatch `DbError::FormatMismatch` is returned.
    pub fn open(
        dir: impl Into<PathBuf>,
        shard_id: u8,
        key_len: u16,
        value_len: u16,
        config: &FixedConfig,
    ) -> DbResult<Self> {
        let dir = dir.into();
        fs::create_dir_all(&dir)?;

        // Clean up any stale temp sidecar left by a prior crash mid-shutdown.
        let _ = fs::remove_file(dir.join("fixed.versions.tmp"));

        let data_path = dir.join("fixed.data");
        let raw_slot_size = slot::slot_size(key_len as usize, value_len as usize);
        if raw_slot_size > u16::MAX as usize {
            return Err(DbError::Config(
                "slot_size exceeds u16::MAX (key_len + value_len + header too large)",
            ));
        }
        let slot_size = raw_slot_size as u16;
        let exists = data_path.exists();

        let file = OpenOptions::new()
            .create(true)
            .read(true)
            .write(true)
            .truncate(false)
            .open(&data_path)?;

        if exists {
            // Validate existing header.
            let mut header = [0u8; HEADER_SIZE as usize];
            file.read_exact_at(&mut header, 0)?;

            if &header[0..4] != MAGIC {
                return Err(DbError::FormatMismatch("fixed.data: bad magic".into()));
            }
            let stored_version = u16::from_le_bytes([header[4], header[5]]);
            if stored_version != VERSION {
                return Err(DbError::FormatMismatch(format!(
                    "fixed.data: version mismatch: stored {stored_version}, expected {VERSION}"
                )));
            }
            let stored_slot_size = u16::from_le_bytes([header[6], header[7]]);
            if stored_slot_size != slot_size {
                return Err(DbError::FormatMismatch(format!(
                    "fixed.data: slot_size mismatch: stored {stored_slot_size}, expected {slot_size}"
                )));
            }
            let stored_key_len = u16::from_le_bytes([header[12], header[13]]);
            if stored_key_len != key_len {
                return Err(DbError::FormatMismatch(format!(
                    "fixed.data: key_len mismatch: stored {stored_key_len}, expected {key_len}"
                )));
            }
            let stored_value_len = u16::from_le_bytes([header[14], header[15]]);
            if stored_value_len != value_len {
                return Err(DbError::FormatMismatch(format!(
                    "fixed.data: value_len mismatch: stored {stored_value_len}, expected {value_len}"
                )));
            }
            let stored_shard_id = header[16];
            if stored_shard_id != shard_id {
                return Err(DbError::FormatMismatch(format!(
                    "fixed.data: shard_id mismatch: stored {stored_shard_id}, expected {shard_id}"
                )));
            }

            let stored_slot_count =
                u32::from_le_bytes([header[8], header[9], header[10], header[11]]);

            let expected_file_size = HEADER_SIZE + stored_slot_count as u64 * slot_size as u64;
            let actual_file_size = file.metadata()?.len();
            if actual_file_size < expected_file_size {
                return Err(DbError::FormatMismatch(format!(
                    "fixed.data: file truncated: expected >= {expected_file_size} bytes, got {actual_file_size}"
                )));
            }

            let bitmap = Bitmap::new(stored_slot_count);
            let versions = vec![0u32; stored_slot_count as usize];

            Ok(Self {
                file,
                dir,
                bitmap,
                versions,
                slot_size,
                slot_count: stored_slot_count,
                key_len,
                value_len,
                shard_id,
                grow_step: config.grow_step,
                pending_writes: 0,
                sync_batch_size: config.sync_batch_size,
                last_sync: Instant::now(),
                sync_interval: config.sync_interval,
                enable_fsync: config.enable_fsync,
                #[cfg(feature = "replication")]
                replication_tx: None,
            })
        } else {
            // New file — write header and pre-allocate initial slots.
            let initial_slots = config.grow_step;
            let total_size = HEADER_SIZE + initial_slots as u64 * slot_size as u64;
            file.set_len(total_size)?;

            let mut header = [0u8; HEADER_SIZE as usize];
            header[0..4].copy_from_slice(MAGIC);
            header[4..6].copy_from_slice(&VERSION.to_le_bytes());
            header[6..8].copy_from_slice(&slot_size.to_le_bytes());
            header[8..12].copy_from_slice(&initial_slots.to_le_bytes());
            header[12..14].copy_from_slice(&key_len.to_le_bytes());
            header[14..16].copy_from_slice(&value_len.to_le_bytes());
            header[16] = shard_id;
            // header[17] = 0 — not a clean shutdown yet
            file.write_all_at(&header, 0)?;
            file.sync_data()?;

            let bitmap = Bitmap::new(initial_slots);
            let versions = vec![0u32; initial_slots as usize];

            Ok(Self {
                file,
                dir,
                bitmap,
                versions,
                slot_size,
                slot_count: initial_slots,
                key_len,
                value_len,
                shard_id,
                grow_step: config.grow_step,
                pending_writes: 0,
                sync_batch_size: config.sync_batch_size,
                last_sync: Instant::now(),
                sync_interval: config.sync_interval,
                enable_fsync: config.enable_fsync,
                #[cfg(feature = "replication")]
                replication_tx: None,
            })
        }
    }

    // ── Offset arithmetic ──────────────────────────────────────────

    /// Byte offset of `slot_id` in the data file.
    #[inline]
    fn slot_offset(&self, slot_id: u32) -> u64 {
        HEADER_SIZE + slot_id as u64 * self.slot_size as u64
    }

    // ── Slot I/O ───────────────────────────────────────────────────

    /// Write an OCCUPIED slot: bumps version from `self.versions[slot_id]`,
    /// serializes full slot, pwrite, updates cache. Returns the new `meta`.
    pub fn write_slot(&mut self, slot_id: u32, key: &[u8], value: &[u8]) -> DbResult<u32> {
        debug_assert!(
            slot_id < self.slot_count,
            "slot_id {slot_id} out of range (slot_count {})",
            self.slot_count
        );
        let old_meta = self.versions[slot_id as usize];
        let new_meta = slot::with_status(slot::bump_version(old_meta), slot::STATUS_OCCUPIED);

        let size = self.slot_size as usize;
        let mut buf = vec![0u8; size];
        slot::serialize_slot(&mut buf, new_meta, key, value);

        let offset = self.slot_offset(slot_id);
        self.file.write_all_at(&buf, offset)?;
        self.versions[slot_id as usize] = new_meta;
        self.maybe_warn_wrap(slot_id, slot::version_of(new_meta));

        self.pending_writes += 1;
        if self.enable_fsync {
            self.file.sync_data()?;
        }
        #[cfg(feature = "replication")]
        if let Some(tx) = &mut self.replication_tx
            && tx
                .push(crate::fixed_replication::FixedReplicationEvent::Write {
                    slot_id,
                    payload: buf,
                })
                .is_err()
        {
            metrics::counter!(
                "armdb.fixed.events_dropped",
                "shard" => self.shard_id.to_string()
            )
            .increment(1);
            // Event dropped; follower reconnect catch-up will reconcile via full scan.
        }
        Ok(new_meta)
    }

    /// Mark a slot as DELETED: 4-byte pwrite of the new `meta` only.
    /// Key and value on disk remain intact (will be overwritten on next
    /// OCCUPIED write to this slot_id). Returns the new `meta`.
    ///
    /// The `key` parameter is unused at disk level but carried to the
    /// replication SPSC hook so followers can update their index without
    /// a pread on their own slot.
    #[cfg_attr(not(feature = "replication"), allow(unused_variables))]
    pub fn delete_slot(&mut self, slot_id: u32, key: &[u8]) -> DbResult<u32> {
        debug_assert!(
            slot_id < self.slot_count,
            "slot_id {slot_id} out of range (slot_count {})",
            self.slot_count
        );
        let old_meta = self.versions[slot_id as usize];
        let new_meta = slot::with_status(slot::bump_version(old_meta), slot::STATUS_DELETED);

        let offset = self.slot_offset(slot_id);
        self.file.write_all_at(&new_meta.to_le_bytes(), offset)?;
        self.versions[slot_id as usize] = new_meta;

        self.pending_writes += 1;
        if self.enable_fsync {
            self.file.sync_data()?;
        }
        #[cfg(feature = "replication")]
        if let Some(tx) = &mut self.replication_tx
            && tx
                .push(crate::fixed_replication::FixedReplicationEvent::Delete {
                    slot_id,
                    meta: new_meta,
                    key: key.to_vec(),
                })
                .is_err()
        {
            metrics::counter!(
                "armdb.fixed.events_dropped",
                "shard" => self.shard_id.to_string()
            )
            .increment(1);
            // Event dropped; follower reconnect catch-up will reconcile via full scan.
        }
        Ok(new_meta)
    }

    #[inline]
    fn maybe_warn_wrap(&self, slot_id: u32, version: u32) {
        if version >= slot::VERSION_WARN_THRESHOLD {
            metrics::counter!(
                "armdb.fixed.version_near_wrap",
                "shard" => self.shard_id.to_string()
            )
            .increment(1);
            tracing::warn!(
                shard_id = self.shard_id,
                slot_id,
                version,
                "FixedStore slot version approaching 30-bit wrap"
            );
        }
    }

    /// Apply a slot event from a leader — writes `meta` as-is (no bump).
    /// Used by follower replication apply path.
    pub fn apply_foreign_slot(
        &mut self,
        slot_id: u32,
        meta: u32,
        key: &[u8],
        value: &[u8],
    ) -> DbResult<()> {
        self.apply_foreign_slot_deferred(slot_id, meta, key, value)?;
        if self.enable_fsync {
            self.sync_replication_batch()?;
        }
        Ok(())
    }

    pub fn apply_foreign_slot_deferred(
        &mut self,
        slot_id: u32,
        meta: u32,
        key: &[u8],
        value: &[u8],
    ) -> DbResult<()> {
        debug_assert!(
            slot_id < self.slot_count,
            "slot_id {slot_id} out of range (slot_count {})",
            self.slot_count
        );
        let size = self.slot_size as usize;
        let mut buf = vec![0u8; size];
        slot::serialize_slot(&mut buf, meta, key, value);
        let offset = self.slot_offset(slot_id);
        self.file.write_all_at(&buf, offset)?;
        self.versions[slot_id as usize] = meta;
        self.pending_writes += 1;
        Ok(())
    }

    /// Apply a DELETE event from a leader — 4-byte pwrite of `meta`.
    /// Key and value bytes on disk remain as they were from the prior OCCUPIED state.
    pub fn apply_foreign_delete(&mut self, slot_id: u32, meta: u32) -> DbResult<()> {
        self.apply_foreign_delete_deferred(slot_id, meta)?;
        if self.enable_fsync {
            self.sync_replication_batch()?;
        }
        Ok(())
    }

    pub fn apply_foreign_delete_deferred(&mut self, slot_id: u32, meta: u32) -> DbResult<()> {
        debug_assert!(
            slot_id < self.slot_count,
            "slot_id {slot_id} out of range (slot_count {})",
            self.slot_count
        );
        let offset = self.slot_offset(slot_id);
        self.file.write_all_at(&meta.to_le_bytes(), offset)?;
        self.versions[slot_id as usize] = meta;
        self.pending_writes += 1;
        Ok(())
    }

    pub fn apply_foreign_reset_deferred(&mut self, slot_id: u32, meta: u32) -> DbResult<()> {
        debug_assert!(
            slot_id < self.slot_count,
            "slot_id {slot_id} out of range (slot_count {})",
            self.slot_count
        );
        debug_assert_eq!(slot::status_of(meta), slot::STATUS_FREE);
        let offset = self.slot_offset(slot_id);
        self.file.write_all_at(&meta.to_le_bytes(), offset)?;
        self.versions[slot_id as usize] = meta;
        self.bitmap.clear(slot_id);
        self.pending_writes += 1;
        Ok(())
    }

    pub fn sync_replication_batch(&mut self) -> DbResult<()> {
        if self.pending_writes > 0 {
            if self.enable_fsync {
                self.file.sync_data()?;
                self.last_sync = Instant::now();
            }
            self.pending_writes = 0;
        }
        Ok(())
    }

    /// Read only the first `SLOT_HEADER_SIZE + key_len` bytes of a slot.
    /// Used rarely on follower stale-cleanup path (spec §8).
    pub fn read_slot_header_and_key(&self, slot_id: u32, key_len: usize) -> DbResult<Vec<u8>> {
        debug_assert!(
            slot_id < self.slot_count,
            "slot_id {slot_id} out of range (slot_count {})",
            self.slot_count
        );
        let n = slot::SLOT_HEADER_SIZE + key_len;
        let mut buf = vec![0u8; n];
        self.file
            .read_exact_at(&mut buf, self.slot_offset(slot_id))?;
        Ok(buf)
    }

    /// Read raw slot bytes for `slot_id`.
    pub fn read_slot(&self, slot_id: u32) -> DbResult<Vec<u8>> {
        debug_assert!(
            slot_id < self.slot_count,
            "slot_id {slot_id} out of range (slot_count {})",
            self.slot_count
        );
        let size = self.slot_size as usize;
        let mut buf = vec![0u8; size];
        let offset = self.slot_offset(slot_id);
        self.file.read_exact_at(&mut buf, offset)?;
        Ok(buf)
    }

    /// Rebuild `versions`/`bitmap` from disk with chunked sequential reads,
    /// invoking `on_entry` for each valid OCCUPIED slot. Owns the scan loop so
    /// the callback never needs to borrow `self`.
    #[allow(clippy::single_range_in_vec_init)]
    pub(crate) fn recover_slots(
        &mut self,
        mode: RecoverMode,
        chunk_bytes: usize,
        on_entry: impl FnMut(&[u8], &[u8], u32),
    ) -> DbResult<u32> {
        let ranges = match mode {
            RecoverMode::Dirty => vec![0..self.slot_count],
            RecoverMode::Clean => {
                occupied_ranges(&self.versions, RECOVERY_MERGE_GAP_BYTES, self.slot_size)
            }
        };
        scan_slot_ranges_into(
            &self.file,
            self.slot_size,
            self.key_len as usize,
            self.value_len as usize,
            &ranges,
            chunk_bytes,
            mode,
            &mut self.versions,
            &mut self.bitmap,
            on_entry,
        )
    }

    // ── Growth ─────────────────────────────────────────────────────

    /// Extend the data file by `grow_step` slots.
    pub fn grow(&mut self) -> DbResult<()> {
        let new_count = self
            .slot_count
            .checked_add(self.grow_step)
            .ok_or(DbError::Internal("slot_count overflow on grow"))?;
        let new_size = HEADER_SIZE + new_count as u64 * self.slot_size as u64;

        // Durably commit the new size before updating in-memory state.
        // If set_len or header write fails, in-memory state is unchanged.
        self.file.set_len(new_size)?;
        self.file.write_all_at(&new_count.to_le_bytes(), 8)?;
        self.file.sync_data()?;

        self.bitmap.grow(new_count);
        self.versions.resize(new_count as usize, 0);
        self.slot_count = new_count;
        Ok(())
    }

    /// Allocate a free slot, growing the file if necessary.
    pub fn alloc_slot(&mut self) -> DbResult<u32> {
        match self.bitmap.alloc() {
            Ok(id) => Ok(id),
            Err(DbError::SlotsFull) => {
                self.grow()?;
                self.bitmap.alloc()
            }
            Err(e) => Err(e),
        }
    }

    // ── Sync ───────────────────────────────────────────────────────

    /// Returns `true` when it is time to sync dirty data to disk.
    pub fn should_sync(&self) -> bool {
        self.pending_writes >= self.sync_batch_size
            || self.last_sync.elapsed() >= self.sync_interval
    }

    /// Sync file data to disk and reset counters.
    pub fn sync(&mut self) -> DbResult<()> {
        self.file.sync_data()?;
        self.pending_writes = 0;
        self.last_sync = Instant::now();
        Ok(())
    }

    // ── Clean shutdown ─────────────────────────────────────────────

    const SIDECAR_FILE: &str = "fixed.versions";

    /// Perform a clean shutdown: sync data, write versions sidecar with
    /// CRC trailer, set clean_shutdown flag.
    pub fn clean_shutdown(&mut self) -> DbResult<()> {
        self.file.sync_data()?;
        self.write_versions_sidecar()?;
        self.file.write_all_at(&[1u8], CLEAN_SHUTDOWN_OFFSET)?;
        self.file.sync_data()?;
        Ok(())
    }

    fn write_versions_sidecar(&self) -> DbResult<()> {
        let final_path = self.dir.join(Self::SIDECAR_FILE);
        let tmp_path = self.dir.join("fixed.versions.tmp");

        // Build sidecar buffer: [versions...] [slot_count LE u32] [crc32 LE u32]
        let expected_len = (self.slot_count as usize) * 4 + 8;
        let mut buf: Vec<u8> = Vec::with_capacity(expected_len);
        for &m in &self.versions {
            buf.extend_from_slice(&m.to_le_bytes());
        }
        buf.extend_from_slice(&self.slot_count.to_le_bytes());
        let mut h = crc32fast::Hasher::new();
        h.update(&buf);
        let crc = h.finalize();
        buf.extend_from_slice(&crc.to_le_bytes());

        // Atomic write: temp file → fsync → rename → dir fsync
        {
            let file = std::fs::File::create(&tmp_path)?;
            use std::io::Write;
            (&file).write_all(&buf)?;
            file.sync_all()?;
        }
        std::fs::rename(&tmp_path, &final_path)?;

        // fsync the parent directory to make the rename durable
        let dir_file = std::fs::File::open(&self.dir)?;
        dir_file.sync_all()?;
        Ok(())
    }

    /// Returns `true` when a prior run cleanly shut down AND a versions
    /// sidecar is on disk. Does NOT validate the sidecar — caller must
    /// call `load_versions_sidecar` which does the validation.
    pub fn has_clean_shutdown(&self) -> bool {
        let mut buf = [0u8; 1];
        if self
            .file
            .read_exact_at(&mut buf, CLEAN_SHUTDOWN_OFFSET)
            .is_err()
        {
            return false;
        }
        buf[0] == 1 && self.dir.join(Self::SIDECAR_FILE).exists()
    }

    /// Clear the clean-shutdown flag (called on open before any writes).
    pub fn clear_clean_shutdown(&mut self) -> DbResult<()> {
        self.file.write_all_at(&[0u8], CLEAN_SHUTDOWN_OFFSET)?;
        self.file.sync_data()?;
        Ok(())
    }

    /// Load the versions sidecar, validating its trailer (slot_count + CRC32).
    /// On success, `self.versions` is filled and `self.bitmap` is derived.
    /// On any mismatch, returns `DbError::FormatMismatch` — caller should
    /// remove the sidecar file and fall back to a full slot scan.
    pub fn load_versions_sidecar(&mut self) -> DbResult<()> {
        let path = self.dir.join(Self::SIDECAR_FILE);
        let data = std::fs::read(&path)?;
        let expected_len = (self.slot_count as usize) * 4 + 8;
        if data.len() != expected_len {
            return Err(DbError::FormatMismatch(format!(
                "fixed.versions size mismatch: expected {expected_len}, got {}",
                data.len()
            )));
        }
        let versions_bytes = &data[..self.slot_count as usize * 4];
        let trailer = &data[data.len() - 8..];
        let stored_slot_count = u32::from_le_bytes(trailer[0..4].try_into().expect("4 bytes"));
        let stored_crc = u32::from_le_bytes(trailer[4..8].try_into().expect("4 bytes"));
        if stored_slot_count != self.slot_count {
            return Err(DbError::FormatMismatch(format!(
                "fixed.versions slot_count mismatch: stored {stored_slot_count}, header {}",
                self.slot_count
            )));
        }
        let mut h = crc32fast::Hasher::new();
        h.update(&data[..data.len() - 4]);
        let actual_crc = h.finalize();
        if actual_crc != stored_crc {
            return Err(DbError::FormatMismatch(format!(
                "fixed.versions CRC mismatch: expected {stored_crc:#x}, got {actual_crc:#x}"
            )));
        }
        self.versions.clear();
        self.versions.reserve(self.slot_count as usize);
        for chunk in versions_bytes.chunks_exact(4) {
            self.versions
                .push(u32::from_le_bytes(chunk.try_into().expect("4 bytes")));
        }
        self.bitmap = crate::fixed::bitmap::Bitmap::from_versions(&self.versions);
        Ok(())
    }

    // ── Getters ────────────────────────────────────────────────────

    pub fn slot_count(&self) -> u32 {
        self.slot_count
    }

    pub fn key_len(&self) -> u16 {
        self.key_len
    }

    pub fn value_len(&self) -> u16 {
        self.value_len
    }

    pub fn dir(&self) -> &std::path::Path {
        &self.dir
    }

    /// Read raw bytes from the data file at absolute offset.
    /// Used by the replication server's full scan.
    pub fn read_chunk_at(&self, offset: u64, buf: &mut [u8]) -> DbResult<()> {
        self.file.read_exact_at(buf, offset)?;
        Ok(())
    }
}

/// Collect maximal runs of OCCUPIED slots, then coalesce two adjacent runs
/// when the byte gap between them is ≤ `merge_gap_bytes`. Pure; no I/O.
fn occupied_ranges(versions: &[u32], merge_gap_bytes: usize, slot_size: u16) -> Vec<Range<u32>> {
    let gap_slots = (merge_gap_bytes / (slot_size.max(1) as usize)) as u32;
    let mut ranges: Vec<Range<u32>> = Vec::new();
    let n = versions.len();
    let mut i = 0usize;
    while i < n {
        if slot::status_of(versions[i]) != slot::STATUS_OCCUPIED {
            i += 1;
            continue;
        }
        let start = i as u32;
        while i < n && slot::status_of(versions[i]) == slot::STATUS_OCCUPIED {
            i += 1;
        }
        let end = i as u32; // exclusive
        match ranges.last_mut() {
            Some(last) if start - last.end <= gap_slots => last.end = end,
            _ => ranges.push(start..end),
        }
    }
    ranges
}

/// Read every slot in `ranges` in chunks of up to `chunk_bytes`, reusing one
/// buffer. Updates `versions`/`bitmap` and invokes `on_entry` for each valid
/// OCCUPIED slot. Free function over split field borrows so the caller can
/// hand `&file` + `&mut versions` + `&mut bitmap` disjointly.
#[allow(clippy::too_many_arguments)]
fn scan_slot_ranges_into(
    file: &File,
    slot_size: u16,
    key_len: usize,
    value_len: usize,
    ranges: &[Range<u32>],
    chunk_bytes: usize,
    mode: RecoverMode,
    versions: &mut [u32],
    bitmap: &mut Bitmap,
    mut on_entry: impl FnMut(&[u8], &[u8], u32),
) -> DbResult<u32> {
    // Nothing to scan (e.g. a clean recovery of a mostly-empty collection with
    // no OCCUPIED ranges) — return before allocating any buffer.
    let total_slots: u64 = ranges.iter().map(|r| (r.end - r.start) as u64).sum();
    if total_slots == 0 {
        return Ok(0);
    }

    let ss = (slot_size as usize).max(1);
    // Cap the reusable buffer to the smaller of the chunk budget and the actual
    // slots we will read, so a shard holding little data does not pin a full
    // `chunk_bytes` (≈4 MiB) buffer. Bounds peak recovery memory to the real
    // data size rather than `shard_count * chunk_bytes`.
    let cap = (chunk_bytes / ss).max(1);
    let slots_per_chunk = (total_slots as usize).min(cap);
    let mut buf = vec![0u8; slots_per_chunk * ss];
    let mut recovered = 0u32;

    for range in ranges {
        let mut slot = range.start;
        while slot < range.end {
            let batch = ((range.end - slot) as usize).min(slots_per_chunk);
            let read_len = batch * ss;
            let offset = HEADER_SIZE + slot as u64 * slot_size as u64;
            file.read_exact_at(&mut buf[..read_len], offset)?;

            for j in 0..batch {
                let slot_id = slot + j as u32;
                let sbuf = &buf[j * ss..(j + 1) * ss];

                match mode {
                    RecoverMode::Dirty => {
                        let meta = slot::meta_of(sbuf);
                        versions[slot_id as usize] = meta;
                        if slot::status_of(meta) == slot::STATUS_OCCUPIED {
                            match slot::read_slot(sbuf, key_len, value_len) {
                                Some((_m, k, v)) => {
                                    bitmap.set(slot_id);
                                    on_entry(k, v, slot_id);
                                    recovered += 1;
                                }
                                None => {
                                    // Torn: keep version, clear status to FREE.
                                    versions[slot_id as usize] =
                                        slot::pack_meta(slot::STATUS_FREE, slot::version_of(meta));
                                }
                            }
                        }
                        // DELETED / FREE: versions already set to meta.
                    }
                    RecoverMode::Clean => {
                        // R2: status from `versions` (sidecar), never disk.
                        // Skip non-occupied gap slots before any parse.
                        if slot::status_of(versions[slot_id as usize]) != slot::STATUS_OCCUPIED {
                            continue;
                        }
                        match slot::read_slot(sbuf, key_len, value_len) {
                            Some((_m, k, v)) => {
                                bitmap.set(slot_id);
                                on_entry(k, v, slot_id);
                                recovered += 1;
                            }
                            None => {
                                let m = versions[slot_id as usize];
                                versions[slot_id as usize] =
                                    slot::pack_meta(slot::STATUS_FREE, slot::version_of(m));
                                bitmap.clear(slot_id);
                            }
                        }
                    }
                }
            }
            slot += batch as u32;
        }
    }
    Ok(recovered)
}

#[cfg(test)]
mod tests {
    use super::*;
    use tempfile::tempdir;

    fn test_config() -> FixedConfig {
        FixedConfig {
            grow_step: 64,
            ..FixedConfig::test()
        }
    }

    #[test]
    fn test_create_and_reopen() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = test_config();

        // Create.
        {
            let shard = FixedShardInner::open(&shard_dir, 0, 8, 32, &cfg).unwrap();
            assert_eq!(shard.slot_count(), cfg.grow_step);
            assert_eq!(shard.key_len(), 8);
            assert_eq!(shard.value_len(), 32);
        }

        // Reopen — header validation must pass.
        {
            let shard = FixedShardInner::open(&shard_dir, 0, 8, 32, &cfg).unwrap();
            assert_eq!(shard.slot_count(), cfg.grow_step);
        }
    }

    #[test]
    fn test_open_rejects_oversized_slot() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = test_config();
        let result = FixedShardInner::open(&shard_dir, 0, 32764, 32764, &cfg);
        assert!(result.is_err());
        let msg = result.err().unwrap().to_string();
        assert!(
            msg.contains("slot_size"),
            "expected slot_size error, got: {msg}"
        );
    }

    #[test]
    fn test_reopen_mismatch_detected() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = test_config();

        // Create with key_len=8, value_len=32.
        FixedShardInner::open(&shard_dir, 0, 8, 32, &cfg).unwrap();

        // Reopen with different key_len — triggers slot_size mismatch
        // (since slot_size depends on key_len + value_len).
        let result = FixedShardInner::open(&shard_dir, 0, 16, 32, &cfg);
        assert!(result.is_err());
        let msg = result.err().unwrap().to_string();
        assert!(
            msg.contains("mismatch"),
            "expected a mismatch error, got: {msg}"
        );
    }

    #[test]
    fn test_open_rejects_truncated_data_file() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = test_config();

        {
            let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
            let id = shard.alloc_slot().unwrap();
            shard
                .write_slot(id, b"key_0001", b"value___00000001")
                .unwrap();
        }

        let data_path = shard_dir.join("fixed.data");
        let file = std::fs::OpenOptions::new()
            .write(true)
            .open(&data_path)
            .unwrap();
        file.set_len(HEADER_SIZE).unwrap();

        let result = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg);
        assert!(result.is_err());
        let msg = result.err().unwrap().to_string();
        assert!(
            msg.contains("truncated"),
            "expected truncation error, got: {msg}"
        );
    }

    #[test]
    fn test_write_read_slot() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = test_config();
        let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();

        let slot_id = shard.alloc_slot().unwrap();
        let key = b"key_0001";
        let value = b"value___00000001";

        shard.write_slot(slot_id, key, value).unwrap();

        let buf = shard.read_slot(slot_id).unwrap();
        let (_m, k, v) = slot::read_slot(&buf, key.len(), value.len()).expect("CRC should match");
        assert_eq!(k, key);
        assert_eq!(v, value);
    }

    #[test]
    fn test_delete_slot() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = test_config();
        let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
        let id = shard.alloc_slot().unwrap();
        shard
            .write_slot(id, b"key_0001", b"value___00000001")
            .unwrap();
        shard.delete_slot(id, b"key_0001").unwrap();
        let buf = shard.read_slot(id).unwrap();
        assert_eq!(slot::status_of(slot::meta_of(&buf)), slot::STATUS_DELETED);
        assert!(slot::read_slot(&buf, 8, 16).is_none());
    }

    #[test]
    fn test_grow() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = FixedConfig {
            grow_step: 4,
            ..FixedConfig::test()
        };
        let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 8, &cfg).unwrap();

        assert_eq!(shard.slot_count(), 4);

        // Fill all initial slots.
        for _ in 0..4 {
            let id = shard.alloc_slot().unwrap();
            shard.write_slot(id, b"kkkkkkkk", b"vvvvvvvv").unwrap();
        }

        // Next alloc triggers grow.
        let id = shard.alloc_slot().unwrap();
        assert_eq!(id, 4);
        assert_eq!(shard.slot_count(), 8);

        shard.write_slot(id, b"kkkkkkkk", b"vvvvvvvv").unwrap();
        let buf = shard.read_slot(id).unwrap();
        assert!(slot::read_slot(&buf, 8, 8).is_some());
    }

    #[test]
    fn test_grow_inmemory_matches_header() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = FixedConfig {
            grow_step: 4,
            ..FixedConfig::test()
        };
        let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 8, &cfg).unwrap();
        assert_eq!(shard.slot_count(), 4);

        shard.grow().unwrap();
        assert_eq!(shard.slot_count(), 8);
        assert_eq!(shard.versions.len(), 8);

        // Reopen and verify header slot_count matches what grow() set.
        drop(shard);
        let shard = FixedShardInner::open(&shard_dir, 0, 8, 8, &cfg).unwrap();
        assert_eq!(shard.slot_count(), 8);
    }

    #[test]
    fn test_clean_shutdown_and_reopen() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = test_config();

        {
            let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
            let id = shard.alloc_slot().unwrap();
            shard
                .write_slot(id, b"key_0001", b"value___00000001")
                .unwrap();
            shard.clean_shutdown().unwrap();
            assert!(shard.has_clean_shutdown());
        }

        {
            let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
            assert!(shard.has_clean_shutdown());
            shard.load_versions_sidecar().unwrap();
            assert_eq!(shard.bitmap.occupied(), 1);
            assert!(shard.bitmap.is_set(0));
            shard.clear_clean_shutdown().unwrap();
            assert!(!shard.has_clean_shutdown());
        }
    }

    #[test]
    fn test_versions_sidecar_roundtrip() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = test_config();

        {
            let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
            let id0 = shard.alloc_slot().unwrap();
            let id1 = shard.alloc_slot().unwrap();
            shard
                .write_slot(id0, b"key_0001", b"value___00000001")
                .unwrap();
            shard
                .write_slot(id1, b"key_0002", b"value___00000002")
                .unwrap();
            shard.delete_slot(id1, b"key_0002").unwrap();
            shard.clean_shutdown().unwrap();
        }

        // Reopen: sidecar present, load versions, derive bitmap.
        let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
        assert!(shard.has_clean_shutdown());
        shard.load_versions_sidecar().unwrap();
        assert_eq!(shard.versions.len(), shard.slot_count() as usize);
        assert_eq!(slot::status_of(shard.versions[0]), slot::STATUS_OCCUPIED);
        assert_eq!(slot::status_of(shard.versions[1]), slot::STATUS_DELETED);
        // derived bitmap: only slot 0 is OCCUPIED
        let b = crate::fixed::bitmap::Bitmap::from_versions(&shard.versions);
        assert!(b.is_set(0));
        assert!(!b.is_set(1));
    }

    #[test]
    fn test_versions_sidecar_trailer_validation_fails_on_truncation() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = test_config();

        {
            let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
            let id = shard.alloc_slot().unwrap();
            shard
                .write_slot(id, b"key_0001", b"value___00000001")
                .unwrap();
            shard.clean_shutdown().unwrap();
        }

        // Truncate the sidecar (strip trailer CRC).
        let sidecar = shard_dir.join("fixed.versions");
        let data = std::fs::read(&sidecar).unwrap();
        std::fs::write(&sidecar, &data[..data.len() - 4]).unwrap();

        let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
        let err = shard.load_versions_sidecar().unwrap_err();
        match err {
            DbError::FormatMismatch(msg) => {
                assert!(
                    msg.contains("fixed.versions") || msg.contains("size mismatch"),
                    "got: {msg}"
                );
            }
            other => panic!("expected FormatMismatch, got {other:?}"),
        }
    }

    #[test]
    fn test_versions_sidecar_corrupted_crc_fails() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = test_config();

        {
            let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
            let id = shard.alloc_slot().unwrap();
            shard
                .write_slot(id, b"key_0001", b"value___00000001")
                .unwrap();
            shard.clean_shutdown().unwrap();
        }

        // Flip a bit in the versions data (not the trailer) — CRC now mismatches.
        let sidecar = shard_dir.join("fixed.versions");
        let mut data = std::fs::read(&sidecar).unwrap();
        data[0] ^= 0xFF;
        std::fs::write(&sidecar, &data).unwrap();

        let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
        let err = shard.load_versions_sidecar().unwrap_err();
        match err {
            DbError::FormatMismatch(msg) => {
                assert!(msg.contains("CRC") || msg.contains("crc"), "got: {msg}");
            }
            other => panic!("expected FormatMismatch, got {other:?}"),
        }
    }

    #[test]
    fn test_sidecar_no_tmp_left_after_clean_shutdown() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = test_config();

        let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
        let id = shard.alloc_slot().unwrap();
        shard
            .write_slot(id, b"key_0001", b"value___00000001")
            .unwrap();
        shard.clean_shutdown().unwrap();

        // Temp file must not exist after clean shutdown.
        assert!(!shard_dir.join("fixed.versions.tmp").exists());
        // Final sidecar must exist.
        assert!(shard_dir.join("fixed.versions").exists());
    }

    #[test]
    fn test_should_sync() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = FixedConfig {
            grow_step: 64,
            sync_batch_size: 2,
            sync_interval: Duration::from_secs(60),
            ..FixedConfig::test()
        };
        let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 8, &cfg).unwrap();

        assert!(!shard.should_sync());

        let id0 = shard.alloc_slot().unwrap();
        shard.write_slot(id0, b"kkkkkkkk", b"vvvvvvvv").unwrap();
        assert!(!shard.should_sync());

        let id1 = shard.alloc_slot().unwrap();
        shard.write_slot(id1, b"kkkkkkkk", b"vvvvvvvv").unwrap();
        assert!(shard.should_sync());

        shard.sync().unwrap();
        assert!(!shard.should_sync());
    }

    #[test]
    fn test_write_slot_bumps_version() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = test_config();
        let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
        let id = shard.alloc_slot().unwrap();
        let m1 = shard
            .write_slot(id, b"key_0001", b"value___00000001")
            .unwrap();
        assert_eq!(slot::status_of(m1), slot::STATUS_OCCUPIED);
        assert_eq!(slot::version_of(m1), 1, "first write → version 1");
        let m2 = shard
            .write_slot(id, b"key_0001", b"value___00000002")
            .unwrap();
        assert_eq!(slot::version_of(m2), 2, "second write → version 2");
        assert_eq!(shard.versions[id as usize], m2);
    }

    #[test]
    fn test_delete_slot_4byte_partial_write() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = test_config();
        let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
        let id = shard.alloc_slot().unwrap();
        shard
            .write_slot(id, b"key_0001", b"value___00000001")
            .unwrap();

        let m = shard.delete_slot(id, b"key_0001").unwrap();
        assert_eq!(slot::status_of(m), slot::STATUS_DELETED);
        assert_eq!(slot::version_of(m), 2);

        // key bytes on disk should still be the original 'key_0001' — only
        // first 4 bytes (meta) were overwritten.
        let buf = shard.read_slot(id).unwrap();
        assert_eq!(
            &buf[slot::SLOT_HEADER_SIZE..slot::SLOT_HEADER_SIZE + 8],
            b"key_0001"
        );
        assert_eq!(slot::meta_of(&buf), m);
        // read_slot helper must return None for DELETED.
        assert!(slot::read_slot(&buf, 8, 16).is_none());
    }

    #[test]
    fn test_delete_then_write_continues_version() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = test_config();
        let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
        let id = shard.alloc_slot().unwrap();
        shard
            .write_slot(id, b"key_0001", b"val1_00_0000_000")
            .unwrap();
        shard.delete_slot(id, b"key_0001").unwrap();
        let m3 = shard
            .write_slot(id, b"key_0002", b"val2_00_0000_000")
            .unwrap();
        assert_eq!(slot::version_of(m3), 3, "version continues across DELETE");
    }

    #[test]
    fn test_apply_foreign_slot_overwrites_version() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = test_config();
        let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
        let foreign_meta = slot::pack_meta(slot::STATUS_OCCUPIED, 500);
        shard
            .apply_foreign_slot(0, foreign_meta, b"key_0001", b"value___00000001")
            .unwrap();
        assert_eq!(shard.versions[0], foreign_meta);
        let buf = shard.read_slot(0).unwrap();
        assert_eq!(slot::meta_of(&buf), foreign_meta);
    }

    #[test]
    fn test_apply_foreign_delete_sets_meta() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = test_config();
        let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
        let occ = slot::pack_meta(slot::STATUS_OCCUPIED, 100);
        shard
            .apply_foreign_slot(0, occ, b"keyabcde", b"0123456701234567")
            .unwrap();
        let del = slot::pack_meta(slot::STATUS_DELETED, 101);
        shard.apply_foreign_delete(0, del).unwrap();
        assert_eq!(shard.versions[0], del);
        let buf = shard.read_slot(0).unwrap();
        assert_eq!(slot::meta_of(&buf), del);
        // key bytes preserved (apply_foreign_delete is 4-byte partial write)
        assert_eq!(
            &buf[slot::SLOT_HEADER_SIZE..slot::SLOT_HEADER_SIZE + 8],
            b"keyabcde"
        );
    }

    #[test]
    fn test_read_slot_header_and_key() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = test_config();
        let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
        let id = shard.alloc_slot().unwrap();
        shard
            .write_slot(id, b"key_0001", b"value___00000001")
            .unwrap();
        let buf = shard.read_slot_header_and_key(id, 8).unwrap();
        assert_eq!(buf.len(), slot::SLOT_HEADER_SIZE + 8);
        assert_eq!(&buf[slot::SLOT_HEADER_SIZE..], b"key_0001");
    }

    #[cfg(feature = "replication")]
    #[test]
    fn test_replication_hook_overflow_does_not_block() {
        use crate::fixed_replication::FixedReplicationEvent;
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = test_config();
        let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
        // Tiny channel: only 2 slots.
        let (producer, _consumer) = rtrb::RingBuffer::<FixedReplicationEvent>::new(2);
        shard.replication_tx = Some(producer);

        let id = shard.alloc_slot().unwrap();
        // 3 writes to the same slot. First 2 fill the channel; 3rd overflows silently.
        for _ in 0..3 {
            shard
                .write_slot(id, b"key_0001", b"value___00000001")
                .unwrap();
        }
        // Sanity: versions was still updated on the 3rd write.
        assert!(slot::version_of(shard.versions[id as usize]) >= 3);
    }

    #[test]
    fn test_warn_wrap_threshold_triggers() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = test_config();
        let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
        let id = shard.alloc_slot().unwrap();
        // Force versions[id] to just below wrap threshold so next write triggers.
        shard.versions[id as usize] =
            slot::pack_meta(slot::STATUS_OCCUPIED, slot::VERSION_WARN_THRESHOLD - 1);
        // No panic expected; warning goes to tracing subscriber (test doesn't assert log capture).
        shard
            .write_slot(id, b"key_0001", b"value___00000001")
            .unwrap();
        assert!(slot::version_of(shard.versions[id as usize]) >= slot::VERSION_WARN_THRESHOLD);
    }

    #[cfg(feature = "replication")]
    #[test]
    fn test_replication_hook_pushes_events() {
        use crate::fixed_replication::FixedReplicationEvent;
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = test_config();
        let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
        let (producer, mut consumer) = rtrb::RingBuffer::new(8);
        shard.replication_tx = Some(producer);

        let id = shard.alloc_slot().unwrap();
        shard
            .write_slot(id, b"key_0001", b"value___00000001")
            .unwrap();
        shard.delete_slot(id, b"key_0001").unwrap();

        // Expect two events.
        match consumer.pop().unwrap() {
            FixedReplicationEvent::Write { slot_id, payload } => {
                assert_eq!(slot_id, id);
                assert_eq!(payload.len(), shard.slot_size as usize);
                assert_eq!(
                    slot::status_of(slot::meta_of(&payload)),
                    slot::STATUS_OCCUPIED
                );
            }
            _ => panic!("expected Write"),
        }
        match consumer.pop().unwrap() {
            FixedReplicationEvent::Delete { slot_id, meta, key } => {
                assert_eq!(slot_id, id);
                assert_eq!(slot::status_of(meta), slot::STATUS_DELETED);
                assert_eq!(key, b"key_0001");
            }
            _ => panic!("expected Delete"),
        }
    }

    #[test]
    fn test_reject_old_version() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        std::fs::create_dir_all(&shard_dir).unwrap();

        // Write a valid v1 header manually (VERSION = 1).
        let slot_size = slot::slot_size(8, 16) as u16;
        let mut header = [0u8; 4096];
        header[0..4].copy_from_slice(b"FIXD");
        header[4..6].copy_from_slice(&1u16.to_le_bytes()); // OLD VERSION
        header[6..8].copy_from_slice(&slot_size.to_le_bytes());
        header[8..12].copy_from_slice(&10u32.to_le_bytes());
        header[12..14].copy_from_slice(&8u16.to_le_bytes());
        header[14..16].copy_from_slice(&16u16.to_le_bytes());
        header[16] = 0; // shard_id
        let data_path = shard_dir.join("fixed.data");
        std::fs::write(&data_path, header).unwrap();
        // Extend file to match declared slot_count.
        let f = std::fs::OpenOptions::new()
            .write(true)
            .open(&data_path)
            .unwrap();
        f.set_len(4096 + 10 * slot_size as u64).unwrap();

        let cfg = test_config();
        let err = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg)
            .err()
            .expect("expected an error opening v1 file");
        match err {
            DbError::FormatMismatch(msg) => {
                assert!(msg.contains("version"), "got: {msg}");
            }
            other => panic!("expected FormatMismatch, got {other:?}"),
        }
    }

    fn test_shard(dir: &std::path::Path, shard_id: u8) -> FixedShardInner {
        let cfg = test_config();
        FixedShardInner::open(dir.join(format!("shard_{shard_id}")), shard_id, 8, 16, &cfg).unwrap()
    }

    #[test]
    fn test_apply_foreign_reset_sets_free_meta_and_clears_pending_with_batch_sync() {
        let dir = tempfile::tempdir().unwrap();
        let mut shard = test_shard(dir.path(), 0);

        let occ = slot::pack_meta(slot::STATUS_OCCUPIED, 10);
        shard
            .apply_foreign_slot_deferred(0, occ, b"key_0001", b"value___00000001")
            .unwrap();

        let free = slot::pack_meta(slot::STATUS_FREE, 11);
        shard.apply_foreign_reset_deferred(0, free).unwrap();

        assert_eq!(slot::status_of(shard.versions[0]), slot::STATUS_FREE);
        assert!(!shard.bitmap.is_set(0));
        assert!(shard.pending_writes > 0);

        shard.sync_replication_batch().unwrap();
        assert_eq!(shard.pending_writes, 0);
    }

    #[test]
    fn test_apply_foreign_deferred_does_not_sync_per_event_when_fsync_enabled() {
        let dir = tempfile::tempdir().unwrap();
        let mut cfg = FixedConfig::test();
        cfg.enable_fsync = true;
        let mut shard = FixedShardInner::open(dir.path(), 0, 8, 16, &cfg).unwrap();

        shard
            .apply_foreign_slot_deferred(
                0,
                slot::pack_meta(slot::STATUS_OCCUPIED, 1),
                b"key_0001",
                b"value___00000001",
            )
            .unwrap();
        assert_eq!(shard.pending_writes, 1);

        shard
            .apply_foreign_delete_deferred(0, slot::pack_meta(slot::STATUS_DELETED, 2))
            .unwrap();
        assert_eq!(shard.pending_writes, 2);

        shard.sync_replication_batch().unwrap();
        assert_eq!(shard.pending_writes, 0);
    }

    use crate::fixed::slot::{STATUS_DELETED, STATUS_FREE, STATUS_OCCUPIED, pack_meta};

    fn occ(v: u32) -> u32 {
        pack_meta(STATUS_OCCUPIED, v)
    }

    #[test]
    fn occupied_ranges_empty_and_all_free() {
        assert!(occupied_ranges(&[], 64, 8).is_empty());
        let free = vec![pack_meta(STATUS_FREE, 0); 4];
        assert!(occupied_ranges(&free, 64, 8).is_empty());
    }

    #[test]
    fn occupied_ranges_single_run() {
        let v = vec![
            pack_meta(STATUS_FREE, 0),
            occ(1),
            occ(1),
            pack_meta(STATUS_FREE, 0),
        ];
        assert_eq!(occupied_ranges(&v, 0, 8), vec![1..3]);
    }

    #[test]
    fn occupied_ranges_all_occupied() {
        let v = vec![occ(1); 5];
        assert_eq!(occupied_ranges(&v, 0, 8), vec![0..5]);
    }

    #[test]
    fn occupied_ranges_merges_small_gap() {
        // slot_size 8, gap_bytes 16 => gap_slots 2. Runs [0..1] and [3..4]:
        // gap = 3 - 1 = 2 <= 2 => merge to [0..4].
        let v = vec![
            occ(1),
            pack_meta(STATUS_FREE, 0),
            pack_meta(STATUS_FREE, 0),
            occ(1),
        ];
        assert_eq!(occupied_ranges(&v, 16, 8), vec![0..4]);
    }

    #[test]
    fn occupied_ranges_keeps_large_gap_separate() {
        // slot_size 8, gap_bytes 16 => gap_slots 2. Runs [0..1] and [4..5]:
        // gap = 4 - 1 = 3 > 2 => two ranges.
        let v = vec![
            occ(1),
            pack_meta(STATUS_FREE, 0),
            pack_meta(STATUS_FREE, 0),
            pack_meta(STATUS_FREE, 0),
            occ(1),
        ];
        assert_eq!(occupied_ranges(&v, 16, 8), vec![0..1, 4..5]);
    }

    #[test]
    fn occupied_ranges_deleted_is_gap() {
        // DELETED between two occupied runs is a gap; with gap_bytes 0 it never merges.
        let v = vec![occ(1), pack_meta(STATUS_DELETED, 2), occ(1)];
        assert_eq!(occupied_ranges(&v, 0, 8), vec![0..1, 2..3]);
    }

    #[test]
    fn recover_slots_dirty_multichunk() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = test_config();

        // Write 5 occupied slots (0..5) into a 64-slot file.
        {
            let mut inner = FixedShardInner::open(&shard_dir, 0, 8, 32, &cfg).unwrap();
            for i in 0..5u64 {
                let id = inner.alloc_slot().unwrap();
                assert_eq!(id, i as u32);
                let mut val = [0u8; 32];
                val[..8].copy_from_slice(&(i * 10).to_be_bytes());
                inner.write_slot(id, &i.to_be_bytes(), &val).unwrap();
            }
            inner.sync().unwrap();
        }

        // Reopen: fresh versions=[0;64], empty bitmap.
        let mut inner = FixedShardInner::open(&shard_dir, 0, 8, 32, &cfg).unwrap();
        let chunk = 2 * inner.slot_size as usize; // 2 slots per chunk => multi-chunk
        let mut got: Vec<(Vec<u8>, u32)> = Vec::new();
        let recovered = inner
            .recover_slots(RecoverMode::Dirty, chunk, |k, _v, id| {
                got.push((k.to_vec(), id));
            })
            .unwrap();

        assert_eq!(recovered, 5);
        got.sort_by_key(|(_, id)| *id);
        assert_eq!(got.len(), 5);
        for (i, (k, id)) in got.iter().enumerate() {
            assert_eq!(*id, i as u32);
            assert_eq!(k, &(i as u64).to_be_bytes());
            assert_eq!(status_of_pub(inner.versions[*id as usize]), STATUS_OCCUPIED);
            assert!(inner.bitmap.is_set(*id));
        }
    }

    #[test]
    fn recover_slots_dirty_torn_in_chunk() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = test_config();

        {
            let mut inner = FixedShardInner::open(&shard_dir, 0, 8, 32, &cfg).unwrap();
            for i in 0..3u64 {
                let id = inner.alloc_slot().unwrap();
                inner.write_slot(id, &i.to_be_bytes(), &[7u8; 32]).unwrap();
            }
            inner.sync().unwrap();
        }

        // Corrupt slot 1's value bytes on disk: offset = HEADER + 1*slot_size + 8 (hdr) + 8 (key).
        let data = shard_dir.join("fixed.data");
        let slot_size = crate::fixed::slot::slot_size(8, 32) as u64;
        let off = super::HEADER_SIZE + slot_size + 8 + 8;
        let f = std::fs::OpenOptions::new().write(true).open(&data).unwrap();
        f.write_all_at(&[0xFFu8; 16], off).unwrap();
        f.sync_data().unwrap();
        drop(f);

        let mut inner = FixedShardInner::open(&shard_dir, 0, 8, 32, &cfg).unwrap();
        let chunk = 2 * inner.slot_size as usize;
        let mut ids: Vec<u32> = Vec::new();
        let recovered = inner
            .recover_slots(RecoverMode::Dirty, chunk, |_k, _v, id| ids.push(id))
            .unwrap();

        assert_eq!(recovered, 2, "torn slot 1 skipped");
        ids.sort_unstable();
        assert_eq!(ids, vec![0, 2]);
        assert_eq!(
            status_of_pub(inner.versions[1]),
            STATUS_FREE,
            "torn -> FREE"
        );
        assert!(!inner.bitmap.is_set(1));
    }

    fn status_of_pub(meta: u32) -> u8 {
        crate::fixed::slot::status_of(meta)
    }
}