armdb 0.7.0

use std::fs::{self, File, OpenOptions};
use std::ops::Range;
use std::os::unix::fs::FileExt;
use std::path::PathBuf;
#[cfg(feature = "replication")]
use std::sync::atomic::{AtomicBool, Ordering};
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,
    /// Set by [`Self::clean_shutdown`]; cleared after the first post-shutdown mutation
    /// triggers [`Self::clear_clean_shutdown`]. Not persisted.
    snapshot_taken: bool,
    /// F5: reusable per-shard scratch for serializing a slot before pwrite,
    /// avoiding a `vec![0; slot_size]` alloc on every write. Exclusively owned
    /// under the shard mutex. Used on the local write path when no replication
    /// producer is attached (a producer requires an owned payload buffer) and on
    /// the follower apply path (which never hands the buffer to an event).
    write_buf: Vec<u8>,
    #[cfg(feature = "replication")]
    pub(crate) replication_tx:
        Option<rtrb::Producer<crate::fixed_replication::FixedReplicationEvent>>,
    /// F-02: set when a replication event could not be pushed into the SPSC
    /// ring (overflow). Read-and-cleared by the Phase-2 server via
    /// `FixedEngineAccess::take_dropped` to signal the follower to reconnect
    /// and re-run Phase-1 (full-scan reconcile).
    #[cfg(feature = "replication")]
    pub(crate) replication_dropped: AtomicBool,
}

/// Test-only failpoint: the next `sync()` returns `Err`. Deterministic under
/// nextest (one process per test); under `cargo test` each test sets and
/// consumes the flag itself (`swap`).
#[cfg(test)]
pub(crate) static FAIL_NEXT_SYNC: std::sync::atomic::AtomicBool =
    std::sync::atomic::AtomicBool::new(false);

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)?;

        // F3: a crash during first-time creation can leave `fixed.data` existing
        // but with no durable header — either 0-length (created, nothing written)
        // or set_len'd to full size but all-zero (header write+fsync never
        // landed). The header is always written+fsync'd *before* the first slot,
        // so an all-zero header is a state no committed DB can be in. Treat that
        // as "not yet initialized" and fall through to fresh init, rather than
        // failing every future open with bad-magic / truncated. This can only
        // discard a headerless file, never committed data (a valid magic short-
        // circuits the check below). Any other content (e.g. a partial < header)
        // keeps the existing error path.
        let has_durable_header = if exists {
            let file_len = file.metadata()?.len();
            if file_len == 0 {
                false
            } else if file_len >= HEADER_SIZE {
                let mut probe = [0u8; HEADER_SIZE as usize];
                file.read_exact_at(&mut probe, 0)?;
                probe.iter().any(|&b| b != 0)
            } else {
                // Partial file shorter than the header: not our crash-during-
                // creation shape — let the validation path report it.
                true
            }
        } else {
            false
        };

        if exists && !has_durable_header {
            tracing::warn!(
                shard_id,
                dir = %dir.display(),
                "fixed.data present but header is absent/all-zero (crash during \
                 initial creation); reinitializing shard"
            );
        }

        if has_durable_header {
            // 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,
                snapshot_taken: false,
                write_buf: Vec::new(),
                #[cfg(feature = "replication")]
                replication_tx: None,
                #[cfg(feature = "replication")]
                replication_dropped: AtomicBool::new(false),
            })
        } 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()?;

            // F3: make the new file's directory entry durable, so a crash right
            // after creation cannot leave a `fixed.data` whose header write was
            // lost — the root cause of the headerless-file self-heal above.
            let dir_file = File::open(&dir)?;
            dir_file.sync_all()?;

            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,
                snapshot_taken: false,
                write_buf: Vec::new(),
                #[cfg(feature = "replication")]
                replication_tx: None,
                #[cfg(feature = "replication")]
                replication_dropped: AtomicBool::new(false),
            })
        }
    }

    // ── 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> {
        self.downgrade_clean_shutdown()?;
        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 offset = self.slot_offset(slot_id);

        // F5: serialize into the reusable per-shard scratch for the disk write.
        // Only allocate a fresh owned buffer when a replication producer is
        // attached and will take ownership of the payload. `serialize_slot`
        // zero-fills the full slot, so the reused buffer carries no stale tail.
        #[cfg(feature = "replication")]
        let payload: Option<Vec<u8>> = if self.replication_tx.is_some() {
            let mut buf = vec![0u8; size];
            slot::serialize_slot(&mut buf, new_meta, key, value);
            self.file.write_all_at(&buf, offset)?;
            Some(buf)
        } else {
            self.write_buf.resize(size, 0);
            slot::serialize_slot(self.write_buf.as_mut_slice(), new_meta, key, value);
            self.file.write_all_at(&self.write_buf, offset)?;
            None
        };
        #[cfg(not(feature = "replication"))]
        {
            self.write_buf.resize(size, 0);
            slot::serialize_slot(self.write_buf.as_mut_slice(), new_meta, key, value);
            self.file.write_all_at(&self.write_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()?;
            // F2: the data is now durable — reset the batch bookkeeping exactly
            // like `sync()` so the caller's post-write `should_sync()` is false
            // and no redundant second fdatasync is issued.
            self.pending_writes = 0;
            self.last_sync = Instant::now();
        }
        #[cfg(feature = "replication")]
        if let Some(buf) = payload {
            let dropped = if let Some(tx) = &mut self.replication_tx {
                tx.push(crate::fixed_replication::FixedReplicationEvent::Write {
                    slot_id,
                    payload: buf,
                })
                .is_err()
            } else {
                false
            };
            if dropped {
                // F-02: mark the shard so the Phase-2 server signals a rescan.
                // A metric alone leaves an active follower unaware of the gap.
                self.replication_dropped.store(true, Ordering::Relaxed);
                metrics::counter!(
                    "armdb.fixed.events_dropped",
                    "shard" => self.shard_id.to_string()
                )
                .increment(1);
            }
        }
        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> {
        self.downgrade_clean_shutdown()?;
        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()?;
            // F2: data is durable — reset batch bookkeeping like `sync()` so the
            // caller's post-write `should_sync()` is false (no double fdatasync).
            self.pending_writes = 0;
            self.last_sync = Instant::now();
        }
        #[cfg(feature = "replication")]
        {
            let dropped = 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()
            } else {
                false
            };
            if dropped {
                // F-02: mark the shard so the Phase-2 server signals a rescan.
                self.replication_dropped.store(true, Ordering::Relaxed);
                metrics::counter!(
                    "armdb.fixed.events_dropped",
                    "shard" => self.shard_id.to_string()
                )
                .increment(1);
            }
        }
        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<()> {
        self.downgrade_clean_shutdown()?;
        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 offset = self.slot_offset(slot_id);
        // F5: the follower apply path never hands the buffer to an event, so it
        // can always reuse the per-shard scratch. `serialize_slot` zero-fills the
        // full slot, so no stale tail leaks from a prior write.
        self.write_buf.resize(size, 0);
        slot::serialize_slot(self.write_buf.as_mut_slice(), meta, key, value);
        self.file.write_all_at(&self.write_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<()> {
        self.downgrade_clean_shutdown()?;
        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<()> {
        self.downgrade_clean_shutdown()?;
        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`.
    ///
    /// `on_entry` returns `Some(evicted_slot)` when the entry it just accepted
    /// was a **duplicate key** already present in the index at another slot
    /// (a post-crash artifact — see [`free_evicted_slot`]). The scan loop then
    /// frees that losing slot so exactly one OCCUPIED slot remains per key.
    #[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) -> Option<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.shard_id,
            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<()> {
        #[cfg(test)]
        if FAIL_NEXT_SYNC.swap(false, std::sync::atomic::Ordering::SeqCst) {
            return Err(std::io::Error::other("test: injected sync failure").into());
        }
        self.file.sync_data()?;
        self.pending_writes = 0;
        self.last_sync = Instant::now();
        Ok(())
    }

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

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

    /// Invalidate a prior clean-shutdown snapshot before a mutation.
    ///
    /// After [`clean_shutdown`](Self::clean_shutdown) the on-disk `fixed.versions`
    /// sidecar is a point-in-time snapshot; any subsequent mutation (local write
    /// or replication-apply) makes it stale. Called at the top of every mutator,
    /// **before** the disk write, this clears the on-disk clean flag so a crash
    /// between here and the write leaves the shard in a safely-Dirty state rather
    /// than trusting a stale-but-CRC-valid sidecar on the next open. No-op unless
    /// a snapshot is currently taken.
    #[inline]
    fn downgrade_clean_shutdown(&mut self) -> DbResult<()> {
        if self.snapshot_taken {
            self.clear_clean_shutdown()?;
            self.snapshot_taken = false;
        }
        Ok(())
    }

    /// 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()?;
        self.snapshot_taken = true;
        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(())
    }
}

/// Free a slot evicted during recovery duplicate-key resolution.
///
/// A crash with `enable_fsync=false` can leave the same key durably present in
/// two OCCUPIED, CRC-valid slots (delete rewrites 4 meta bytes in the old slot,
/// the re-put lands in a new slot; the kernel may persist the new page but not
/// the old meta). Recovery scans every valid slot, so the index sees the key
/// twice; the second sighting wins (last-scanned-wins) and this frees the first.
///
/// The version is preserved (status → FREE only) so a future write to this slot
/// bumps monotonically, and the slot is dropped from the bitmap. This heals the
/// leak + index/bitmap divergence; it does **not** guarantee recency — picking
/// the newest of two slots would need a cross-slot ordinal the on-disk format
/// does not carry, so the winner is deterministic but arbitrary (out of scope).
fn free_evicted_slot(versions: &mut [u32], bitmap: &mut Bitmap, evicted: u32) {
    let m = versions[evicted as usize];
    versions[evicted as usize] = slot::pack_meta(slot::STATUS_FREE, slot::version_of(m));
    bitmap.clear(evicted);
    metrics::counter!("armdb.fixed.recovery_duplicate_key").increment(1);
    tracing::warn!(
        evicted_slot = evicted,
        "FixedStore recovery: duplicate key in two slots after crash; freed the \
         losing slot (winner: last-scanned; recency not guaranteed)"
    );
}

/// 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,
    shard_id: u8,
    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) -> Option<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);
                                    recovered += 1;
                                    if let Some(evicted) = on_entry(k, v, slot_id) {
                                        free_evicted_slot(versions, bitmap, evicted);
                                        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);
                                recovered += 1;
                                if let Some(evicted) = on_entry(k, v, slot_id) {
                                    free_evicted_slot(versions, bitmap, evicted);
                                    recovered -= 1;
                                }
                            }
                            None => {
                                // F4: a CRC failure is *not* expected on the clean
                                // path (data was fsync'd and the sidecar says
                                // OCCUPIED), unlike Dirty where torn writes are
                                // normal. Surface it before silently freeing the
                                // slot, so the operator sees the dropped entry.
                                metrics::counter!(
                                    "armdb.fixed.clean_recovery_crc_fail",
                                    "shard" => shard_id.to_string()
                                )
                                .increment(1);
                                tracing::warn!(
                                    slot_id,
                                    shard_id,
                                    "CRC mismatch on clean recovery — dropping slot"
                                );
                                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_clean_shutdown_downgrade_on_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();
        shard.clean_shutdown().unwrap();
        assert!(shard.has_clean_shutdown());

        // Straggler write after clean_shutdown must clear the on-disk flag.
        shard
            .write_slot(id, b"key_0001", b"value___00000002")
            .unwrap();
        assert!(!shard.has_clean_shutdown());
    }

    /// Regression: a foreign (replication-apply) write after `clean_shutdown`
    /// must downgrade the clean flag exactly like a local write. Otherwise a
    /// follower that keeps applying leader events after clean_shutdown leaves a
    /// stale sidecar on disk that passes CRC validation → silent data loss on
    /// the next open. Covers all three deferred apply entry points.
    #[test]
    fn test_clean_shutdown_downgrade_on_foreign_apply() {
        let dir = tempdir().unwrap();
        let cfg = test_config();

        // A slot that already exists on disk, then a clean_shutdown, then one
        // foreign apply of the given kind — the flag must be cleared afterwards.
        let check = |name: &str, apply: &dyn Fn(&mut FixedShardInner, u32)| {
            let shard_dir = dir.path().join(name);
            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(),
                "{name}: flag set by clean_shutdown"
            );

            apply(&mut shard, id);

            assert!(
                !shard.has_clean_shutdown(),
                "{name}: foreign apply after clean_shutdown must clear the flag"
            );
        };

        check("shard_occupied", &|shard, id| {
            let meta = slot::pack_meta(slot::STATUS_OCCUPIED, 7);
            shard
                .apply_foreign_slot_deferred(id, meta, b"key_0002", b"value___00000002")
                .unwrap();
        });
        check("shard_deleted", &|shard, id| {
            let meta = slot::pack_meta(slot::STATUS_DELETED, 8);
            shard.apply_foreign_delete_deferred(id, meta).unwrap();
        });
        check("shard_reset", &|shard, id| {
            let meta = slot::pack_meta(slot::STATUS_FREE, 9);
            shard.apply_foreign_reset_deferred(id, meta).unwrap();
        });
    }

    #[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);
    }

    #[cfg(feature = "replication")]
    #[test]
    fn test_f02_overflow_sets_dropped_flag() {
        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 ring: 2 usable slots.
        let (producer, _consumer) = rtrb::RingBuffer::<FixedReplicationEvent>::new(2);
        shard.replication_tx = Some(producer);

        // No drop yet.
        assert!(!shard.replication_dropped.load(Ordering::Relaxed));

        let id = shard.alloc_slot().unwrap();
        // Overflow the ring: the 3rd write cannot be pushed.
        for _ in 0..3 {
            shard
                .write_slot(id, b"key_0001", b"value___00000001")
                .unwrap();
        }
        // F-02: the drop flag must now be set.
        assert!(
            shard.replication_dropped.load(Ordering::Relaxed),
            "F-02 regression: overflow did not set the dropped flag"
        );

        // delete_slot must also set the flag on overflow.
        shard.replication_dropped.store(false, Ordering::Relaxed);
        shard.delete_slot(id, b"key_0001").unwrap();
        assert!(
            shard.replication_dropped.load(Ordering::Relaxed),
            "F-02 regression: delete overflow did not set the dropped flag"
        );
    }

    #[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));
                None
            })
            .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);
                None
            })
            .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));
    }

    /// F2: with `enable_fsync=true` + `sync_batch_size=1` (the Financial preset
    /// shape), `write_slot`/`delete_slot` fsync the data themselves. They must
    /// also reset the batch bookkeeping (`pending_writes`, `last_sync`) so the
    /// caller's post-write `should_sync()` is false — otherwise `const_map`
    /// issues a redundant second fdatasync per operation.
    #[test]
    fn f2_fsync_write_resets_batch_no_double_sync() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = FixedConfig {
            grow_step: 64,
            sync_batch_size: 1,
            sync_interval: Duration::from_secs(60),
            enable_fsync: true,
            ..FixedConfig::test()
        };
        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();
        assert_eq!(
            shard.pending_writes, 0,
            "write_slot's own fsync must clear pending_writes"
        );
        assert!(
            !shard.should_sync(),
            "no redundant second fdatasync after an fsync'd write"
        );

        shard.delete_slot(id, b"key_0001").unwrap();
        assert_eq!(
            shard.pending_writes, 0,
            "delete_slot's own fsync must clear pending_writes"
        );
        assert!(
            !shard.should_sync(),
            "no redundant second fdatasync after an fsync'd delete"
        );
    }

    /// F3: a crash during first-time shard creation can leave `fixed.data`
    /// existing but with no durable header (0-length). The next open must
    /// self-heal (fresh init), not fail forever with an I/O/format error.
    #[test]
    fn f3_open_self_heals_zero_length_data_file() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        std::fs::create_dir_all(&shard_dir).unwrap();
        std::fs::File::create(shard_dir.join("fixed.data")).unwrap(); // 0-length

        let cfg = test_config();
        let shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg)
            .expect("0-length fixed.data must self-heal, not error forever");
        assert_eq!(shard.slot_count(), cfg.grow_step);
    }

    /// F3: a crash after `set_len` but before the header write+fsync leaves a
    /// full-size file whose header region is all zeros. That state is
    /// unreachable for any committed DB (the header is fsync'd before the first
    /// slot), so the next open self-heals instead of failing with bad-magic.
    #[test]
    fn f3_open_self_heals_all_zero_header() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        std::fs::create_dir_all(&shard_dir).unwrap();

        let cfg = test_config();
        let ss = slot::slot_size(8, 16) as u64;
        let total = HEADER_SIZE + cfg.grow_step as u64 * ss;
        let f = std::fs::File::create(shard_dir.join("fixed.data")).unwrap();
        f.set_len(total).unwrap();
        f.sync_all().unwrap();
        drop(f);

        let shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg)
            .expect("all-zero header must self-heal, not bad-magic FormatMismatch");
        assert_eq!(shard.slot_count(), cfg.grow_step);

        // The healed shard now has a valid, durable header: reopen must validate.
        drop(shard);
        let shard2 = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
        assert_eq!(shard2.slot_count(), cfg.grow_step);
    }

    /// F3 safety: self-heal must NEVER fire on a file with a valid header —
    /// committed data must survive reopen untouched.
    #[test]
    fn f3_open_does_not_self_heal_valid_data() {
        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.sync().unwrap();
        }

        let shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
        let buf = shard.read_slot(0).unwrap();
        let (_m, k, v) = slot::read_slot(&buf, 8, 16).expect("committed row survives reopen");
        assert_eq!(k, b"key_0001");
        assert_eq!(v, b"value___00000001");
    }

    /// F4: on the clean recovery path a CRC-failed OCCUPIED slot (sidecar says
    /// OCCUPIED, data bytes corrupt) is demoted to FREE. Unexpected on the clean
    /// path, so it is now logged + metered; here we pin the freeing behavior at
    /// the scan layer.
    #[test]
    fn f4_clean_recovery_crc_fail_frees_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();
            for i in 0..3u64 {
                let id = shard.alloc_slot().unwrap();
                shard.write_slot(id, &i.to_be_bytes(), &[9u8; 16]).unwrap();
            }
            shard.clean_shutdown().unwrap();
        }

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

        let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
        assert!(shard.has_clean_shutdown());
        shard.load_versions_sidecar().unwrap();
        let mut seen: Vec<Vec<u8>> = Vec::new();
        let recovered = shard
            .recover_slots(RecoverMode::Clean, RECOVERY_CHUNK_BYTES, |k, _v, _id| {
                seen.push(k.to_vec());
                None
            })
            .unwrap();

        assert_eq!(recovered, 2, "CRC-failed slot 1 excluded from recovery");
        assert_eq!(
            slot::status_of(shard.versions[1]),
            slot::STATUS_FREE,
            "CRC-fail slot demoted to FREE on clean path"
        );
        assert!(!shard.bitmap.is_set(1));
    }

    /// F5: writes go through a reusable per-shard scratch buffer (no per-write
    /// heap alloc when no replication producer is attached). Consecutive writes
    /// must not leak bytes from a prior write into the slot — `serialize_slot`
    /// zero-fills the full slot before each write.
    #[test]
    fn f5_reusable_write_buf_is_correct() {
        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"AAAAAAAAAAAAAAAA")
            .unwrap();
        // Rewrite the same slot — reuses the scratch buffer.
        shard
            .write_slot(id, b"key_0002", b"BBBBBBBBBBBBBBBB")
            .unwrap();
        let buf = shard.read_slot(id).unwrap();
        let (_m, k, v) = slot::read_slot(&buf, 8, 16).expect("CRC ok after reuse");
        assert_eq!(k, b"key_0002");
        assert_eq!(v, b"BBBBBBBBBBBBBBBB");

        // A different slot after — buffer reuse must not corrupt neighbors.
        let id2 = shard.alloc_slot().unwrap();
        shard
            .write_slot(id2, b"key_0003", b"CCCCCCCCCCCCCCCC")
            .unwrap();
        let buf2 = shard.read_slot(id2).unwrap();
        let (_m2, k2, v2) = slot::read_slot(&buf2, 8, 16).expect("CRC ok");
        assert_eq!(k2, b"key_0003");
        assert_eq!(v2, b"CCCCCCCCCCCCCCCC");

        // Original slot still reads its latest value.
        let buf3 = shard.read_slot(id).unwrap();
        let (_m3, k3, v3) = slot::read_slot(&buf3, 8, 16).expect("CRC ok");
        assert_eq!(k3, b"key_0002");
        assert_eq!(v3, b"BBBBBBBBBBBBBBBB");
    }

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