armdb 0.14.1

//! Durability trait — abstracts Bitcask (append-only log) and FixedStore
//! (fixed-slot pwrite) backends behind a common interface.
//!
//! [`DurabilityInner`] covers per-shard write operations (one locked shard).
//! [`Durability`] covers engine-level operations (shard routing, open/close).
//!
//! Concrete types:
//! - [`Bitcask`] / [`ShardInner`](crate::shard::ShardInner) — append-only log
//! - [`Fixed`] / [`FixedShardInner`](crate::fixed::shard::FixedShardInner) — fixed-slot pwrite

use std::path::Path;
use std::sync::Arc;

use crate::disk_loc::DiskLoc;
use crate::entry::entry_size;
use crate::error::{DbError, DbResult};
use crate::fixed::config::FixedConfig;
use crate::fixed::shard::{RECOVERY_CHUNK_BYTES, RecoverMode, SyncSite};
use crate::key::Location;
use crate::sync::{self, MutexGuard};

// ── Trait: per-shard write operations ────────────────────────────────────────

/// Per-shard write operations abstracted over storage backend.
///
/// Implementations operate on a single shard (already locked by the caller
/// via [`Durability::lock_shard`]).
pub trait DurabilityInner: Send {
    type Loc: Location;

    /// Write a new entry (key not previously present). Returns location.
    fn write_new(&mut self, shard_id: u8, key: &[u8], value: &[u8]) -> DbResult<Self::Loc>;

    /// Update an existing entry. `old_loc` is used for dead-bytes tracking
    /// (Bitcask) or in-place overwrite (FixedStore).
    /// Returns the new location (Bitcask: new DiskLoc; FixedStore: same slot_id).
    fn write_update(
        &mut self,
        shard_id: u8,
        old_loc: Self::Loc,
        key: &[u8],
        value: &[u8],
    ) -> DbResult<Self::Loc>;

    /// Delete an entry. Writes a tombstone (Bitcask) or marks the slot deleted
    /// (FixedStore).
    fn write_tombstone(&mut self, shard_id: u8, old_loc: Self::Loc, key: &[u8]) -> DbResult<()>;

    /// Discard a written location (race-condition cleanup).
    /// Bitcask: no-op (compaction handles stale entries).
    /// FixedStore: free the allocated slot.
    fn write_discard(&mut self, loc: Self::Loc, key: &[u8]) -> DbResult<()>;

    /// Whether sync is needed after recent writes.
    fn should_sync(&self) -> bool;

    /// Perform sync (fdatasync).
    fn sync(&mut self) -> DbResult<()>;
}

// ── Trait: engine-level operations ───────────────────────────────────────────

/// Engine-level durability operations.
///
/// One `Durability` instance owns the full set of shards for a single
/// database directory. Collections (ConstTree, ConstMap, etc.) are
/// generic over `D: Durability` and delegate all disk I/O through this
/// trait.
pub trait Durability: Send + Sync + Sized {
    type Loc: Location;
    type Inner: DurabilityInner<Loc = Self::Loc>;

    fn shard_count(&self) -> usize;
    fn lock_shard(&self, shard_id: usize) -> MutexGuard<'_, Self::Inner>;
    fn sync_shard(&self, shard_id: usize) -> DbResult<()>;
    fn shard_prefix_bits(&self) -> usize;
    /// Shards holding writes no barrier has made durable yet.
    fn dirty_shards(&self) -> usize;
    fn flush(&self) -> DbResult<()>;
    fn close(&self) -> DbResult<()>;
    fn max_projected_rows(&self) -> usize;
    fn max_projected_bytes(&self) -> usize;
}

// ══════════════════════════════════════════════════════════════════════════════
// Bitcask backend
// ══════════════════════════════════════════════════════════════════════════════

/// Engine-level wrapper for the Bitcask (append-only log) backend.
#[allow(dead_code)]
pub struct Bitcask {
    pub(crate) engine: crate::engine::Engine,
    pub(crate) compaction_threshold: f64,
}

// ── DurabilityInner for ShardInner (Bitcask) ─────────────────────────────────

impl DurabilityInner for crate::shard::ShardInner {
    type Loc = DiskLoc;

    fn write_new(&mut self, shard_id: u8, key: &[u8], value: &[u8]) -> DbResult<DiskLoc> {
        let (loc, _gsn) = self.append_entry(shard_id, key, value, false)?;
        Ok(loc)
    }

    fn write_update(
        &mut self,
        shard_id: u8,
        old_loc: DiskLoc,
        key: &[u8],
        value: &[u8],
    ) -> DbResult<DiskLoc> {
        let (new_loc, _gsn) = self.append_entry(shard_id, key, value, false)?;
        self.add_dead_bytes(old_loc.file_id, entry_size(key.len(), old_loc.len));
        Ok(new_loc)
    }

    fn write_tombstone(&mut self, shard_id: u8, old_loc: DiskLoc, key: &[u8]) -> DbResult<()> {
        let (_loc, _gsn) = self.append_entry(shard_id, key, &[], true)?;
        self.add_dead_bytes(old_loc.file_id, entry_size(key.len(), old_loc.len));
        Ok(())
    }

    fn write_discard(&mut self, _loc: DiskLoc, _key: &[u8]) -> DbResult<()> {
        // No-op: compaction handles stale entries in Bitcask.
        Ok(())
    }

    fn should_sync(&self) -> bool {
        crate::shard::ShardInner::should_sync(self)
    }

    fn sync(&mut self) -> DbResult<()> {
        Ok(())
    }
}

// ── Durability for Bitcask ───────────────────────────────────────────────────

impl Durability for Bitcask {
    type Loc = DiskLoc;
    type Inner = crate::shard::ShardInner;

    fn shard_count(&self) -> usize {
        self.engine.shards().len()
    }

    fn lock_shard(&self, shard_id: usize) -> MutexGuard<'_, crate::shard::ShardInner> {
        self.engine.shards()[shard_id].lock()
    }

    fn sync_shard(&self, shard_id: usize) -> DbResult<()> {
        #[cfg(feature = "ctx-gates")]
        crate::ctx_gates_sync::record_sync_call();
        self.engine.shards()[shard_id].flush()
    }

    fn shard_prefix_bits(&self) -> usize {
        self.engine.config().shard_prefix_bits
    }

    fn dirty_shards(&self) -> usize {
        self.engine.dirty_shards()
    }

    fn flush(&self) -> DbResult<()> {
        self.engine.flush()
    }

    fn close(&self) -> DbResult<()> {
        self.engine.close()
    }

    fn max_projected_rows(&self) -> usize {
        self.engine.config().max_projected_rows
    }

    fn max_projected_bytes(&self) -> usize {
        self.engine.config().max_projected_bytes
    }
}

// ══════════════════════════════════════════════════════════════════════════════
// Fixed backend
// ══════════════════════════════════════════════════════════════════════════════

/// Engine-level wrapper for the FixedStore (fixed-slot pwrite) backend.
#[allow(dead_code)]
pub struct Fixed {
    pub(crate) engine: Arc<crate::fixed::engine::FixedEngine>,
}

impl Fixed {
    /// Open or create a fixed-slot database at the given path.
    ///
    #[doc = include_str!("../docs/exclusive-access.md")]
    pub fn open(
        path: impl AsRef<Path>,
        config: FixedConfig,
        key_len: usize,
        value_len: usize,
    ) -> DbResult<Self> {
        if key_len > u16::MAX as usize {
            return Err(DbError::Config("key_len exceeds u16::MAX"));
        }
        if value_len > u16::MAX as usize {
            return Err(DbError::Config("value_len exceeds u16::MAX"));
        }
        let engine = crate::fixed::engine::FixedEngine::open(
            path,
            config,
            key_len as u16,
            value_len as u16,
        )?;
        Ok(Self {
            engine: Arc::new(engine),
        })
    }

    /// Recover entries from all shards in parallel (one thread per shard).
    ///
    /// `make_visitor` is called **inside** each shard thread to build that
    /// shard's visitor with hoisted state; it therefore need not be `Send`.
    /// Populates `versions`/`bitmap` for every shard (clean or dirty path) and
    /// invokes the shard's visitor for each valid OCCUPIED slot. Returns the
    /// total recovered count.
    ///
    /// Crate-internal: this is a low-level recovery primitive used only by the
    /// collection constructors (`ConstTree`/`ConstMap` and their Zero wrappers).
    /// It is deliberately not part of the public API — the visitor-factory shape
    /// is an implementation detail of parallel per-shard recovery.
    pub(crate) fn recover_entries<F, V>(&self, make_visitor: F) -> DbResult<u32>
    where
        F: Fn(usize) -> V + Sync,
        V: FnMut(&[u8], &[u8], u32) -> Option<u32>,
    {
        let shards = self.engine.shards();
        let make_visitor = &make_visitor;

        // Cap concurrency so a database with many shards (up to 255) does not
        // spawn a thread — and its per-shard scan buffer — for every shard at
        // once. For the common case (shard_count <= cores) this is a single
        // batch, identical to recovering all shards in parallel.
        let max_parallel = crate::config::default_shard_count().max(1);
        let mut results: Vec<DbResult<u32>> = Vec::with_capacity(shards.len());

        for batch_start in (0..shards.len()).step_by(max_parallel) {
            let batch_end = (batch_start + max_parallel).min(shards.len());
            let batch: Vec<DbResult<u32>> = std::thread::scope(|s| {
                let handles: Vec<_> = (batch_start..batch_end)
                    .map(|shard_idx| {
                        let shard = &shards[shard_idx];
                        // R3: `make_visitor` is invoked inside the shard thread
                        // (in `recover_one_shard`), so the visitor never crosses
                        // the thread boundary — only `&make_visitor` (Sync) does.
                        s.spawn(move || Self::recover_one_shard(shard, shard_idx, make_visitor))
                    })
                    .collect();

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

        // If several shards fail at once (e.g. an Io error and a panic), the
        // first failure by shard index wins — `results` is ordered by shard, so
        // `?` short-circuits on the lowest-index error and later ones are dropped.
        let mut total = 0u32;
        for r in results {
            total += r?;
        }
        Ok(total)
    }

    /// Recover a single shard (runs on its own thread). Builds the shard's
    /// visitor via `make_visitor(shard_idx)` in-thread, picks the clean-sidecar
    /// or dirty full-scan path, populates `versions`/`bitmap`, and records the
    /// recovery-duration metric labelled with the path actually taken.
    fn recover_one_shard<F, V>(
        shard: &crate::fixed::engine::FixedShard,
        shard_idx: usize,
        make_visitor: &F,
    ) -> DbResult<u32>
    where
        F: Fn(usize) -> V,
        V: FnMut(&[u8], &[u8], u32) -> Option<u32>,
    {
        let start = std::time::Instant::now();
        let mut inner = sync::lock(&shard.inner);

        // Clean sidecar or dirty full scan?
        let use_sidecar = if inner.has_clean_shutdown() {
            match inner.load_versions_sidecar() {
                Ok(()) => true,
                Err(e) => {
                    tracing::warn!(
                        shard = shard_idx, error = %e,
                        "fixed.versions sidecar invalid; falling back to full scan"
                    );
                    let _ = std::fs::remove_file(inner.dir().join("fixed.versions"));
                    false
                }
            }
        } else {
            false
        };

        let mode = if use_sidecar {
            RecoverMode::Clean
        } else {
            RecoverMode::Dirty
        };

        // R3: build the visitor inside the thread.
        let mut visitor = make_visitor(shard_idx);
        let recovered = inner.recover_slots(mode, RECOVERY_CHUNK_BYTES, |k, v, slot_id| {
            visitor(k, v, slot_id)
        })?;

        inner.clear_clean_shutdown()?;

        let elapsed = start.elapsed().as_secs_f64();
        metrics::histogram!(
            "armdb.recovery.duration_seconds",
            "path" => if use_sidecar { "fixed_sidecar" } else { "fixed_scan" }
        )
        .record(elapsed);

        Ok(recovered)
    }
}

// ── DurabilityInner for FixedShardInner ──────────────────────────────────────

impl DurabilityInner for crate::fixed::shard::FixedShardInner {
    type Loc = u32;

    fn write_new(&mut self, _shard_id: u8, key: &[u8], value: &[u8]) -> DbResult<u32> {
        if let Some(armed) = self.take_armed_slot() {
            // Replay after a crash finds the slot free: the allocation lived
            // only in the dead process's memory, so re-mark it here or the next
            // `alloc_slot` would hand the same slot to another key.
            self.bitmap.set(armed.slot_id);
            self.write_slot_with_meta(armed.slot_id, armed.meta, key, value)?;
            return Ok(armed.slot_id);
        }
        let slot = self.alloc_slot()?;
        let _ = self.write_slot(slot, key, value)?;
        Ok(slot)
    }

    fn write_update(
        &mut self,
        _shard_id: u8,
        old_loc: u32,
        key: &[u8],
        value: &[u8],
    ) -> DbResult<u32> {
        if let Some(armed) = self.take_armed_slot() {
            if armed.slot_id != old_loc {
                return Err(DbError::Internal(
                    "armed FixedStore slot does not match the key's current slot",
                ));
            }
            self.write_slot_with_meta(old_loc, armed.meta, key, value)?;
            return Ok(old_loc);
        }
        let _ = self.write_slot(old_loc, key, value)?;
        Ok(old_loc)
    }

    fn write_tombstone(&mut self, _shard_id: u8, old_loc: u32, key: &[u8]) -> DbResult<()> {
        if let Some(armed) = self.take_armed_slot() {
            if armed.slot_id != old_loc {
                return Err(DbError::Internal(
                    "armed FixedStore slot does not match the key's current slot",
                ));
            }
            self.delete_slot_with_meta(old_loc, armed.meta, key)?;
            self.bitmap.clear(old_loc);
            return Ok(());
        }
        self.delete_slot(old_loc, key)?;
        self.bitmap.clear(old_loc);
        Ok(())
    }

    fn write_discard(&mut self, loc: u32, key: &[u8]) -> DbResult<()> {
        self.delete_slot(loc, key)?;
        self.bitmap.clear(loc);
        Ok(())
    }

    fn should_sync(&self) -> bool {
        self.should_sync()
    }

    fn sync(&mut self) -> DbResult<()> {
        crate::fixed::shard::FixedShardInner::sync(self, SyncSite::Write)
    }
}

// ── Durability for Fixed ─────────────────────────────────────────────────────

impl Durability for Fixed {
    type Loc = u32;
    type Inner = crate::fixed::shard::FixedShardInner;

    fn shard_count(&self) -> usize {
        self.engine.shards().len()
    }

    fn lock_shard(&self, shard_id: usize) -> MutexGuard<'_, crate::fixed::shard::FixedShardInner> {
        sync::lock(&self.engine.shards()[shard_id].inner)
    }

    fn sync_shard(&self, shard_id: usize) -> DbResult<()> {
        self.lock_shard(shard_id).sync(SyncSite::Write)
    }

    fn shard_prefix_bits(&self) -> usize {
        self.engine.config().shard_prefix_bits
    }

    fn dirty_shards(&self) -> usize {
        self.engine.dirty_shards()
    }

    fn flush(&self) -> DbResult<()> {
        self.engine.flush()
    }

    fn close(&self) -> DbResult<()> {
        self.engine.close()
    }

    fn max_projected_rows(&self) -> usize {
        self.engine.config().max_projected_rows
    }

    fn max_projected_bytes(&self) -> usize {
        self.engine.config().max_projected_bytes
    }
}

#[cfg(test)]
thread_local! {
    static TTL_DELETE_FAIL_AFTER: std::cell::Cell<usize> = const { std::cell::Cell::new(usize::MAX) };
}

/// Scoped test-only fault for a TTL delete after an applied prefix.
#[cfg(test)]
pub(crate) struct TtlDeleteFailAfter(usize);

#[cfg(test)]
impl TtlDeleteFailAfter {
    pub(crate) fn after(applied: usize) -> Self {
        let previous = TTL_DELETE_FAIL_AFTER.with(|remaining| {
            let previous = remaining.get();
            remaining.set(applied);
            previous
        });
        Self(previous)
    }
}

#[cfg(test)]
impl Drop for TtlDeleteFailAfter {
    fn drop(&mut self) {
        TTL_DELETE_FAIL_AFTER.with(|remaining| remaining.set(self.0));
    }
}

#[cfg(test)]
pub(crate) fn ttl_delete_should_fail() -> bool {
    TTL_DELETE_FAIL_AFTER.with(|remaining| match remaining.get() {
        0 => true,
        usize::MAX => false,
        value => {
            remaining.set(value - 1);
            false
        }
    })
}

#[cfg(test)]
#[cfg(all(feature = "typed-tree", feature = "var-collections"))]
mod write_sync_family_tests {
    use crate::codec::Codec;
    use crate::hook::TypedWriteHook;
    use crate::{
        Config, ConstTree, DbError, DbResult, TypedMap, TypedTree, Upsert, VarTypedMap,
        VarTypedTree, ZeroTree,
    };

    use crate::flush_test::{data_successes, reset};

    #[derive(Clone, Copy)]
    struct U64Codec;

    impl Codec<u64> for U64Codec {
        fn encode_to(&self, value: &u64, buf: &mut Vec<u8>) -> DbResult<()> {
            buf.clear();
            buf.extend_from_slice(&value.to_be_bytes());
            Ok(())
        }

        fn decode_from(&self, bytes: &[u8]) -> DbResult<u64> {
            let value: [u8; 8] = bytes
                .try_into()
                .map_err(|_| DbError::CorruptedEntry { offset: 0 })?;
            Ok(u64::from_be_bytes(value))
        }
    }

    #[derive(Clone, Copy)]
    struct Expiring;

    impl TypedWriteHook<[u8; 8], u64> for Expiring {
        const EXPIRES: bool = true;

        fn deadline(&self, _key: &[u8; 8], _value: &u64) -> Option<u64> {
            Some(1)
        }

        fn on_write(&self, _key: &[u8; 8], _old: Option<&u64>, _new: Option<&u64>) {}
    }

    fn write_sync_config() -> Config {
        let mut config = Config::test();
        config.shard_count = 1;
        config.sync_batch_size = 4;
        config
    }

    fn write_sync_error_config() -> Config {
        let mut config = write_sync_config();
        config.sync_batch_size = 3;
        config
    }

    fn assert_ttl_error_barriers_prior_writes(result: DbResult<crate::ttl::SweepOutcome>) {
        assert!(
            matches!(
                result,
                Err(DbError::Internal("test: TTL delete injected failure"))
            ),
            "the fourth TTL delete must return the injected write error"
        );
        assert_eq!(
            data_successes(),
            1,
            "three applied TTL tombstones must be barriered before the later write error returns"
        );
    }

    #[test]
    fn write_sync_family_routes_typed_map_after_threshold() {
        reset();
        let dir = tempfile::tempdir().unwrap();
        let map =
            TypedMap::<[u8; 8], u64, U64Codec>::open(dir.path(), write_sync_config(), U64Codec)
                .expect("open typed map");

        for value in 0..4_u64 {
            map.put(&value.to_be_bytes(), value).expect("put");
        }

        assert_eq!(
            data_successes(),
            1,
            "four TypedMap writes must complete one Bitcask barrier"
        );
    }

    #[test]
    fn write_sync_family_routes_zero_tree_after_threshold() {
        reset();
        let dir = tempfile::tempdir().unwrap();
        let tree =
            ZeroTree::<[u8; 8], 8>::open(dir.path(), write_sync_config()).expect("open zero tree");

        for value in 0..4_u64 {
            let bytes = value.to_be_bytes();
            tree.put(&bytes, &bytes).expect("put");
        }

        assert_eq!(
            data_successes(),
            1,
            "four ZeroTree writes must complete one Bitcask barrier"
        );
    }

    #[test]
    fn write_sync_family_routes_var_typed_tree_after_threshold() {
        reset();
        let dir = tempfile::tempdir().unwrap();
        let tree =
            VarTypedTree::<[u8; 8], u64, U64Codec>::open(dir.path(), write_sync_config(), U64Codec)
                .expect("open var typed tree");

        for value in 0..4_u64 {
            tree.put(&value.to_be_bytes(), &value).expect("put");
        }

        assert_eq!(
            data_successes(),
            1,
            "four VarTypedTree writes must complete one Bitcask barrier"
        );
    }

    #[test]
    fn write_sync_atomic_routes_typed_tree_after_threshold() {
        reset();
        let dir = tempfile::tempdir().unwrap();
        let tree =
            TypedTree::<[u8; 8], u64, U64Codec>::open(dir.path(), write_sync_config(), U64Codec)
                .expect("open typed tree");
        let anchor = 0_u64.to_be_bytes();

        tree.atomic(&anchor, |shard| {
            for value in 0..4_u64 {
                shard.put(&value.to_be_bytes(), value)?;
            }
            Ok(())
        })
        .expect("atomic writes");

        assert_eq!(
            data_successes(),
            1,
            "four TypedTree atomic writes must complete one Bitcask barrier"
        );
    }

    #[test]
    fn write_sync_atomic_routes_var_typed_tree_after_threshold() {
        reset();
        let dir = tempfile::tempdir().unwrap();
        let tree =
            VarTypedTree::<[u8; 8], u64, U64Codec>::open(dir.path(), write_sync_config(), U64Codec)
                .expect("open var typed tree");
        let anchor = 0_u64.to_be_bytes();

        tree.atomic(&anchor, |shard| {
            for value in 0..4_u64 {
                shard.put(&value.to_be_bytes(), &value)?;
            }
            Ok(())
        })
        .expect("atomic writes");

        assert_eq!(
            data_successes(),
            1,
            "four VarTypedTree atomic writes must complete one Bitcask barrier"
        );
    }

    #[test]
    fn write_sync_update_many_routes_const_tree_after_threshold() {
        reset();
        let dir = tempfile::tempdir().unwrap();
        let tree = ConstTree::<[u8; 8], 8>::open(dir.path(), write_sync_config())
            .expect("open const tree");
        let rows = (0..4_u64)
            .map(|value| (value.to_be_bytes(), value))
            .collect();

        tree.update_many(rows, |_key, _old, value| Upsert::Put(value.to_be_bytes()))
            .expect("update many");

        assert_eq!(
            data_successes(),
            1,
            "four ConstTree bulk writes must complete one Bitcask barrier"
        );
    }

    #[test]
    fn write_sync_ttl_routes_typed_map_after_threshold() {
        reset();
        let dir = tempfile::tempdir().unwrap();
        let map = TypedMap::<[u8; 8], u64, U64Codec, Expiring>::open_hooked(
            dir.path(),
            write_sync_config(),
            U64Codec,
            Expiring,
        )
        .expect("open expiring typed map");
        map.build_ttl_index().expect("build ttl index");
        for value in 0..4_u64 {
            map.put(&value.to_be_bytes(), value).expect("put");
        }

        reset();
        let outcome = map.sweep_expired(1, 4).expect("sweep expired");
        assert_eq!(outcome.deleted, 4, "all four rows must expire");

        assert_eq!(
            data_successes(),
            1,
            "four TypedMap TTL deletes must complete one Bitcask barrier"
        );
    }

    #[test]
    fn write_sync_ttl_routes_var_typed_map_after_threshold() {
        reset();
        let dir = tempfile::tempdir().unwrap();
        let map = VarTypedMap::<[u8; 8], u64, U64Codec, Expiring>::open_hooked(
            dir.path(),
            write_sync_config(),
            U64Codec,
            Expiring,
        )
        .expect("open expiring var typed map");
        map.build_ttl_index().expect("build ttl index");
        for value in 0..4_u64 {
            map.put(&value.to_be_bytes(), &value).expect("put");
        }

        reset();
        let outcome = map.sweep_expired(1, 4).expect("sweep expired");
        assert_eq!(outcome.deleted, 4, "all four rows must expire");

        assert_eq!(
            data_successes(),
            1,
            "four VarTypedMap TTL deletes must complete one Bitcask barrier"
        );
    }

    #[test]
    fn write_sync_ttl_error_routes_typed_tree_partial_success() {
        reset();
        let dir = tempfile::tempdir().unwrap();
        let tree = TypedTree::<[u8; 8], u64, U64Codec, Expiring>::open_hooked(
            dir.path(),
            write_sync_error_config(),
            U64Codec,
            Expiring,
        )
        .expect("open expiring typed tree");
        tree.build_ttl_index().expect("build ttl index");
        for value in 0..4_u64 {
            tree.put(&value.to_be_bytes(), value).expect("put");
        }

        reset();
        let _failure = super::TtlDeleteFailAfter::after(3);
        assert_ttl_error_barriers_prior_writes(tree.sweep_expired(1, 4));
    }

    #[test]
    fn write_sync_ttl_error_routes_typed_map_partial_success() {
        reset();
        let dir = tempfile::tempdir().unwrap();
        let map = TypedMap::<[u8; 8], u64, U64Codec, Expiring>::open_hooked(
            dir.path(),
            write_sync_error_config(),
            U64Codec,
            Expiring,
        )
        .expect("open expiring typed map");
        map.build_ttl_index().expect("build ttl index");
        for value in 0..4_u64 {
            map.put(&value.to_be_bytes(), value).expect("put");
        }

        reset();
        let _failure = super::TtlDeleteFailAfter::after(3);
        assert_ttl_error_barriers_prior_writes(map.sweep_expired(1, 4));
    }

    #[test]
    fn write_sync_ttl_error_routes_var_typed_tree_partial_success() {
        reset();
        let dir = tempfile::tempdir().unwrap();
        let tree = VarTypedTree::<[u8; 8], u64, U64Codec, Expiring>::open_hooked(
            dir.path(),
            write_sync_error_config(),
            U64Codec,
            Expiring,
        )
        .expect("open expiring var typed tree");
        tree.build_ttl_index().expect("build ttl index");
        for value in 0..4_u64 {
            tree.put(&value.to_be_bytes(), &value).expect("put");
        }

        reset();
        let _failure = super::TtlDeleteFailAfter::after(3);
        assert_ttl_error_barriers_prior_writes(tree.sweep_expired(1, 4));
    }

    #[test]
    fn write_sync_ttl_error_routes_var_typed_map_partial_success() {
        reset();
        let dir = tempfile::tempdir().unwrap();
        let map = VarTypedMap::<[u8; 8], u64, U64Codec, Expiring>::open_hooked(
            dir.path(),
            write_sync_error_config(),
            U64Codec,
            Expiring,
        )
        .expect("open expiring var typed map");
        map.build_ttl_index().expect("build ttl index");
        for value in 0..4_u64 {
            map.put(&value.to_be_bytes(), &value).expect("put");
        }

        reset();
        let _failure = super::TtlDeleteFailAfter::after(3);
        assert_ttl_error_barriers_prior_writes(map.sweep_expired(1, 4));
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::fixed::config::FixedConfig;
    use crate::fixed::slot::{STATUS_DELETED, STATUS_FREE, STATUS_OCCUPIED, status_of, version_of};
    use tempfile::tempdir;

    fn test_fixed_config() -> FixedConfig {
        FixedConfig {
            shard_count: 1,
            grow_step: 16,
            ..FixedConfig::test()
        }
    }

    #[test]
    fn test_recover_populates_versions_dirty() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("db");

        // First run: write entries, drop without clean shutdown (simulate crash).
        {
            let fixed = Fixed::open(&path, test_fixed_config(), 8, 16).unwrap();
            {
                let mut inner = sync::lock(&fixed.engine.shards()[0].inner);
                let id0 = inner.alloc_slot().unwrap();
                inner
                    .write_slot(id0, b"keyaaaaa", b"val_0000_0000_00")
                    .unwrap();
                let id1 = inner.alloc_slot().unwrap();
                inner
                    .write_slot(id1, b"keybbbbb", b"val_0001_0000_00")
                    .unwrap();
                inner.delete_slot(id1, b"keybbbbb").unwrap();
                inner.sync(SyncSite::Write).unwrap();
            }
            drop(fixed);
        }

        // Second run: reopen, recover populates versions.
        let fixed = Fixed::open(&path, test_fixed_config(), 8, 16).unwrap();
        let entries = std::sync::Mutex::new(Vec::new());
        let entries_ref = &entries;
        fixed
            .recover_entries(move |_shard_idx| {
                move |k: &[u8], v: &[u8], slot_id: u32| {
                    entries_ref
                        .lock()
                        .unwrap()
                        .push((k.to_vec(), v.to_vec(), slot_id));
                    None
                }
            })
            .unwrap();
        let entries = entries.into_inner().unwrap();
        assert_eq!(entries.len(), 1, "only one OCCUPIED entry");
        assert_eq!(entries[0].0, b"keyaaaaa");
        let inner = sync::lock(&fixed.engine.shards()[0].inner);
        assert_eq!(status_of(inner.versions[0]), STATUS_OCCUPIED);
        assert_eq!(status_of(inner.versions[1]), STATUS_DELETED);
    }

    #[test]
    fn test_recover_populates_versions_clean() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("db");
        {
            let fixed = Fixed::open(&path, test_fixed_config(), 8, 16).unwrap();
            {
                let mut inner = sync::lock(&fixed.engine.shards()[0].inner);
                let id = inner.alloc_slot().unwrap();
                inner
                    .write_slot(id, b"keyaaaaa", b"val_0000_0000_00")
                    .unwrap();
            }
            fixed.engine.close().unwrap();
        }
        let fixed = Fixed::open(&path, test_fixed_config(), 8, 16).unwrap();
        let entries = std::sync::Mutex::new(Vec::new());
        let entries_ref = &entries;
        fixed
            .recover_entries(move |_shard_idx| {
                move |k: &[u8], _v: &[u8], _id: u32| {
                    entries_ref.lock().unwrap().push(k.to_vec());
                    None
                }
            })
            .unwrap();
        let entries = entries.into_inner().unwrap();
        assert_eq!(entries.len(), 1);
        let inner = sync::lock(&fixed.engine.shards()[0].inner);
        assert_eq!(status_of(inner.versions[0]), STATUS_OCCUPIED);
        assert_eq!(version_of(inner.versions[0]), 1);
    }

    #[test]
    fn test_torn_slot_preserves_version() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("db");

        {
            let fixed = Fixed::open(&path, test_fixed_config(), 8, 16).unwrap();
            {
                let mut inner = sync::lock(&fixed.engine.shards()[0].inner);
                let id = inner.alloc_slot().unwrap();
                inner
                    .write_slot(id, b"keyaaaaa", b"val_0000_0000_00")
                    .unwrap();
                inner.sync(SyncSite::Write).unwrap();
            }
            drop(fixed);
        }

        // Corrupt the value bytes on disk at offset = header(4096) + 0*slot_size + slot_header(8) + key_len(8) = 4112
        let shard_data = path.join("shard_000").join("fixed.data");
        use std::os::unix::fs::FileExt;
        let f = std::fs::OpenOptions::new()
            .write(true)
            .open(&shard_data)
            .unwrap();
        f.write_all_at(&[0xFFu8; 16], 4112).unwrap();
        f.sync_data().unwrap();
        drop(f);

        // Reopen dirty → recover → CRC mismatch on slot 0.
        let fixed = Fixed::open(&path, test_fixed_config(), 8, 16).unwrap();
        let entries = std::sync::Mutex::new(Vec::new());
        let entries_ref = &entries;
        fixed
            .recover_entries(move |_shard_idx| {
                move |k: &[u8], _v: &[u8], _id: u32| {
                    entries_ref.lock().unwrap().push(k.to_vec());
                    None
                }
            })
            .unwrap();
        let entries = entries.into_inner().unwrap();
        assert!(entries.is_empty(), "torn slot must be skipped");

        // versions[0] must have STATUS_FREE but preserve the original version (1).
        let inner = sync::lock(&fixed.engine.shards()[0].inner);
        assert_eq!(status_of(inner.versions[0]), STATUS_FREE);
        assert_eq!(
            version_of(inner.versions[0]),
            1,
            "version must be preserved so next bump continues monotonically"
        );
    }

    #[test]
    fn fixed_open_rejects_oversized_key_len() {
        let dir = tempfile::tempdir().unwrap();
        let result = Fixed::open(dir.path(), FixedConfig::test(), 65536, 16);
        assert!(result.is_err());
        let msg = result.err().unwrap().to_string();
        assert!(
            msg.contains("key_len"),
            "expected key_len error, got: {msg}"
        );
    }

    #[test]
    fn fixed_open_rejects_oversized_value_len() {
        let dir = tempfile::tempdir().unwrap();
        let result = Fixed::open(dir.path(), FixedConfig::test(), 8, 65536);
        assert!(result.is_err());
        let msg = result.err().unwrap().to_string();
        assert!(
            msg.contains("value_len"),
            "expected value_len error, got: {msg}"
        );
    }

    #[cfg(feature = "replication")]
    #[test]
    fn fixed_write_discard_uses_real_key_for_replication_delete() {
        use crate::durability::DurabilityInner;
        use crate::fixed_replication::FixedReplicationEvent;

        let dir = tempfile::tempdir().unwrap();
        let fixed = Fixed::open(dir.path(), FixedConfig::test(), 8, 16).unwrap();
        let mut inner = fixed.lock_shard(0);
        let (producer, mut consumer) = rtrb::RingBuffer::new(4);
        inner.replication_tx = Some(producer);

        inner.write_discard(0, b"key_0001").unwrap();

        match consumer.pop().unwrap() {
            FixedReplicationEvent::Delete { key, .. } => assert_eq!(key, b"key_0001"),
            _ => panic!("expected delete event"),
        }
    }

    #[test]
    fn clean_recovery_skips_merged_gap_slot() {
        use crate::fixed::slot::pack_meta;
        let dir = tempdir().unwrap();
        let path = dir.path().join("db");

        // Write 3 valid occupied slots, then mark slot 1 FREE in `versions` only
        // (disk keeps its valid occupied bytes). Clean close records the sidecar
        // from the mutated versions, so the sidecar disagrees with disk at slot 1.
        {
            let fixed = Fixed::open(&path, test_fixed_config(), 8, 16).unwrap();
            {
                let mut inner = sync::lock(&fixed.engine.shards()[0].inner);
                for i in 0..3u64 {
                    let id = inner.alloc_slot().unwrap();
                    assert_eq!(id, i as u32);
                    let mut val = [0u8; 16];
                    val[..8].copy_from_slice(&(i * 100).to_be_bytes());
                    inner.write_slot(id, &i.to_be_bytes(), &val).unwrap();
                }
                // Sidecar will say slot 1 is FREE; disk still holds a valid occupied slot.
                inner.versions[1] = pack_meta(STATUS_FREE, version_of(inner.versions[1]));
                inner.bitmap.clear(1);
            }
            fixed.engine.close().unwrap();
        }

        let fixed = Fixed::open(&path, test_fixed_config(), 8, 16).unwrap();
        let seen = std::sync::Mutex::new(Vec::new());
        let seen_ref = &seen;
        fixed
            .recover_entries(move |_shard_idx| {
                move |k: &[u8], _v: &[u8], _id: u32| {
                    seen_ref.lock().unwrap().push(k.to_vec());
                    None
                }
            })
            .unwrap();
        let mut seen = seen.into_inner().unwrap();
        seen.sort();

        assert_eq!(
            seen,
            vec![0u64.to_be_bytes().to_vec(), 2u64.to_be_bytes().to_vec()]
        );
        let inner = sync::lock(&fixed.engine.shards()[0].inner);
        assert_eq!(
            status_of(inner.versions[1]),
            STATUS_FREE,
            "gap slot untouched"
        );
        assert!(!inner.bitmap.is_set(1), "gap slot not added to bitmap");
    }

    #[test]
    fn clean_recovery_crc_fail_frees_slot() {
        use std::os::unix::fs::FileExt;
        let dir = tempdir().unwrap();
        let path = dir.path().join("db");

        {
            let fixed = Fixed::open(&path, test_fixed_config(), 8, 16).unwrap();
            {
                let mut inner = sync::lock(&fixed.engine.shards()[0].inner);
                for i in 0..3u64 {
                    let id = inner.alloc_slot().unwrap();
                    inner.write_slot(id, &i.to_be_bytes(), &[9u8; 16]).unwrap();
                }
            }
            fixed.engine.close().unwrap(); // sidecar: slots 0,1,2 OCCUPIED
        }

        // Corrupt slot 1 value bytes: offset = HEADER + 1*slot_size + 8 (hdr) + 8 (key).
        let data = path.join("shard_000").join("fixed.data");
        let ss = crate::fixed::slot::slot_size(8, 16) as u64;
        let off = crate::fixed::shard::HEADER_SIZE + ss + 8 + 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 fixed = Fixed::open(&path, test_fixed_config(), 8, 16).unwrap();
        let seen = std::sync::Mutex::new(Vec::new());
        let seen_ref = &seen;
        fixed
            .recover_entries(move |_shard_idx| {
                move |k: &[u8], _v: &[u8], _id: u32| {
                    seen_ref.lock().unwrap().push(k.to_vec());
                    None
                }
            })
            .unwrap();
        let mut seen = seen.into_inner().unwrap();
        seen.sort();

        assert_eq!(
            seen,
            vec![0u64.to_be_bytes().to_vec(), 2u64.to_be_bytes().to_vec()]
        );
        let inner = sync::lock(&fixed.engine.shards()[0].inner);
        assert_eq!(
            status_of(inner.versions[1]),
            STATUS_FREE,
            "CRC-fail slot -> FREE"
        );
        assert!(!inner.bitmap.is_set(1));
    }

    #[test]
    fn recover_entries_factory_called_once_per_shard() {
        use std::sync::atomic::{AtomicUsize, Ordering};
        let dir = tempdir().unwrap();
        let path = dir.path().join("db");
        let cfg = FixedConfig {
            shard_count: 2,
            grow_step: 16,
            ..FixedConfig::test()
        };

        {
            let fixed = Fixed::open(&path, cfg.clone(), 8, 16).unwrap();
            {
                let mut s0 = fixed.lock_shard(0);
                for i in 0..2u64 {
                    let id = s0.alloc_slot().unwrap();
                    s0.write_slot(id, &(1000 + i).to_be_bytes(), &[1u8; 16])
                        .unwrap();
                }
            }
            {
                let mut s1 = fixed.lock_shard(1);
                for i in 0..3u64 {
                    let id = s1.alloc_slot().unwrap();
                    s1.write_slot(id, &(2000 + i).to_be_bytes(), &[2u8; 16])
                        .unwrap();
                }
            }
            fixed.engine.close().unwrap();
        }

        let fixed = Fixed::open(&path, cfg, 8, 16).unwrap();
        let calls = AtomicUsize::new(0);
        let per_shard: Vec<std::sync::Mutex<Vec<Vec<u8>>>> =
            (0..2).map(|_| std::sync::Mutex::new(Vec::new())).collect();
        let calls_ref = &calls;
        let per_shard_ref = &per_shard;
        fixed
            .recover_entries(move |shard_idx| {
                calls_ref.fetch_add(1, Ordering::Relaxed);
                move |k: &[u8], _v: &[u8], _id: u32| {
                    per_shard_ref[shard_idx].lock().unwrap().push(k.to_vec());
                    None
                }
            })
            .unwrap();

        assert_eq!(
            calls.load(Ordering::Relaxed),
            2,
            "factory called once per shard"
        );
        let mut s0: Vec<u64> = per_shard[0]
            .lock()
            .unwrap()
            .iter()
            .map(|k| u64::from_be_bytes(k[..8].try_into().unwrap()))
            .collect();
        let mut s1: Vec<u64> = per_shard[1]
            .lock()
            .unwrap()
            .iter()
            .map(|k| u64::from_be_bytes(k[..8].try_into().unwrap()))
            .collect();
        s0.sort_unstable();
        s1.sort_unstable();
        assert_eq!(
            s0,
            vec![1000, 1001],
            "shard 0 visitor sees only shard 0 keys"
        );
        assert_eq!(
            s1,
            vec![2000, 2001, 2002],
            "shard 1 visitor sees only shard 1 keys"
        );
    }
}