seaweedfs/seaweed-volume/src/storage/needle_map.rs

//! NeedleMapper: index mapping NeedleId -> (Offset, Size).
//!
//! Two implementations:
//!   - `CompactNeedleMap`: in-memory segmented sorted arrays (~10 bytes/entry)
//!   - `RedbNeedleMap`: disk-backed via redb (low RAM, slightly slower)
//!
//! The `NeedleMap` enum wraps both and provides a uniform interface.
//! Loaded from .idx file on volume mount. Supports Get, Put, Delete with
//! metrics tracking (file count, byte count, deleted count, deleted bytes).

use std::collections::BTreeMap;
use std::io::{self, Read, Seek, Write};
use std::path::Path;
use std::sync::atomic::{AtomicI64, AtomicU64, Ordering};

#[cfg(feature = "redb-experimental-cursor")]
use std::ops::Bound;

mod compact_map;
pub mod file_pool;
mod idx_metric;
pub mod sorted_file;
use compact_map::CompactMap;
use idx_metric::metrics_from_idx;
use sorted_file::SortedFileNeedleMap;

use redb::{Database, Durability, ReadableDatabase, ReadableTable, TableDefinition};

use crate::storage::idx;
use crate::storage::needle::needle::get_actual_size;
use crate::storage::types::*;

// ============================================================================
// NeedleValue
// ============================================================================

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct NeedleValue {
    pub offset: Offset,
    pub size: Size,
}

/// Packed size of a NeedleValue in redb storage: OFFSET_SIZE + SIZE_SIZE.
const PACKED_NEEDLE_VALUE_SIZE: usize = OFFSET_SIZE + SIZE_SIZE;

/// Pack an (Offset, Size) pair into bytes for redb storage.
/// Layout: [offset OFFSET_SIZE bytes] [size 4 bytes big-endian]
fn pack_needle_value(nv: &NeedleValue) -> [u8; PACKED_NEEDLE_VALUE_SIZE] {
    let mut buf = [0u8; PACKED_NEEDLE_VALUE_SIZE];
    nv.offset.to_bytes(&mut buf[..OFFSET_SIZE]);
    nv.size.to_bytes(&mut buf[OFFSET_SIZE..]);
    buf
}

/// Unpack bytes from redb storage into (Offset, Size).
fn unpack_needle_value(bytes: &[u8; PACKED_NEEDLE_VALUE_SIZE]) -> NeedleValue {
    NeedleValue {
        offset: Offset::from_bytes(&bytes[..OFFSET_SIZE]),
        size: Size::from_bytes(&bytes[OFFSET_SIZE..]),
    }
}

fn packed_to_needle_value(bytes: &[u8]) -> Option<NeedleValue> {
    if bytes.len() != PACKED_NEEDLE_VALUE_SIZE {
        return None;
    }
    let mut arr = [0u8; PACKED_NEEDLE_VALUE_SIZE];
    arr.copy_from_slice(bytes);
    Some(unpack_needle_value(&arr))
}

// ============================================================================
// NeedleMapMetric
// ============================================================================

/// Metrics tracking for needle map operations.
#[derive(Debug, Default)]
pub struct NeedleMapMetric {
    pub file_count: AtomicI64,
    pub file_byte_count: AtomicU64,
    pub deletion_count: AtomicI64,
    pub deletion_byte_count: AtomicU64,
    pub max_file_key: AtomicU64,
    /// Largest (offset.to_actual_offset() + get_actual_size(size, version))
    /// observed during the load walk. Used at volume load to verify that no
    /// .idx entry references bytes past the end of .dat (issue #8928)
    /// without paying for a second linear scan.
    pub max_needle_end: AtomicI64,
}

impl NeedleMapMetric {
    /// Update metrics based on a Put operation (additive-only, matching Go's logPut).
    fn on_put(&self, key: NeedleId, old: Option<&NeedleValue>, new_size: Size) {
        self.maybe_set_max_file_key(key);
        // Go: always LogFileCounter(newSize) which does FileCounter++ and FileByteCounter += newSize
        self.file_count.fetch_add(1, Ordering::Relaxed);
        self.file_byte_count
            .fetch_add(new_size.0 as u64, Ordering::Relaxed);
        // Go: if oldSize > 0 && oldSize.IsValid() { LogDeletionCounter(oldSize) }
        if let Some(old_val) = old
            && old_val.size.0 > 0
            && old_val.size.is_valid()
        {
            self.deletion_count.fetch_add(1, Ordering::Relaxed);
            self.deletion_byte_count
                .fetch_add(old_val.size.0 as u64, Ordering::Relaxed);
        }
    }

    /// Update metrics based on a Delete operation (additive-only, matching Go's logDelete).
    fn on_delete(&self, old: &NeedleValue) {
        if old.size.0 > 0 {
            self.deletion_count.fetch_add(1, Ordering::Relaxed);
            self.deletion_byte_count
                .fetch_add(old.size.0 as u64, Ordering::Relaxed);
        }
    }

    fn maybe_set_max_file_key(&self, key: NeedleId) {
        let key_val: u64 = key.into();
        loop {
            let current = self.max_file_key.load(Ordering::Relaxed);
            if key_val <= current {
                break;
            }
            if self
                .max_file_key
                .compare_exchange(current, key_val, Ordering::Relaxed, Ordering::Relaxed)
                .is_ok()
            {
                break;
            }
        }
    }

    /// Update `max_needle_end` if this entry's (offset + actual size) exceeds
    /// the running maximum. Skips deleted/zero-offset entries because they
    /// don't reserve space in .dat.
    fn maybe_set_max_needle_end(&self, offset: Offset, size: Size, version: Version) {
        if offset.is_zero() || !size.is_valid() {
            return;
        }
        let end = offset.to_actual_offset() + get_actual_size(size, version);
        loop {
            let current = self.max_needle_end.load(Ordering::Relaxed);
            if end <= current {
                break;
            }
            if self
                .max_needle_end
                .compare_exchange(current, end, Ordering::Relaxed, Ordering::Relaxed)
                .is_ok()
            {
                break;
            }
        }
    }
}

// ============================================================================
// NeedleMapKind
// ============================================================================

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum NeedleMapKind {
    InMemory,
    Redb,
    RedbMedium,
    RedbLarge,
}

impl NeedleMapKind {
    /// Bytes of redb page cache to give ONE volume's index database.
    ///
    /// A volume server opens a separate redb database per volume, so the
    /// process-wide ceiling is roughly (open volumes x this budget). redb's
    /// own default is 1 GiB per database, which with hundreds of volumes
    /// grows without practical bound as traffic touches them (#11179).
    /// These tiers mirror the Go server's LevelDB sizing (block cache +
    /// write buffer of 3/6/12 MiB), rounded to powers of two.
    pub fn redb_cache_bytes(self) -> usize {
        const MIB: usize = 1024 * 1024;
        match self {
            // InMemory never opens redb; return the smallest tier so a
            // caller that does not branch on kind still gets a sane bound.
            NeedleMapKind::InMemory | NeedleMapKind::Redb => 4 * MIB,
            NeedleMapKind::RedbMedium => 8 * MIB,
            NeedleMapKind::RedbLarge => 16 * MIB,
        }
    }
}

// ============================================================================
// IdxFileWriter trait
// ============================================================================

/// Trait for appending to an index file.
///
/// The file is opened without append mode and each row is written at the
/// current `idx_file_offset` — the same positioned-write model the Go
/// server uses — because an append-mode handle cannot truncate on Windows,
/// where the std library keeps it strictly append-only.
pub trait IdxFileWriter: Write + Seek + Send + Sync {
    fn sync_all(&self) -> io::Result<()>;
    /// Truncate the file to `len` bytes. Used to remove an orphan .idx row
    /// left by a failed redb commit so `idx_file_offset` stays a contiguous
    /// replay watermark.
    fn truncate_to(&mut self, len: u64) -> io::Result<()>;
}

impl IdxFileWriter for std::fs::File {
    fn sync_all(&self) -> io::Result<()> {
        std::fs::File::sync_all(self)
    }
    fn truncate_to(&mut self, len: u64) -> io::Result<()> {
        self.set_len(len)
    }
}

// ============================================================================
// CompactNeedleMap (in-memory)
// ============================================================================

/// In-memory needle map backed by a CompactMap (segmented sorted arrays).
/// Uses ~10 bytes per entry instead of ~40-48 bytes with HashMap.
/// The .idx file is kept open for append-only writes.
pub struct CompactNeedleMap {
    map: CompactMap,
    metric: NeedleMapMetric,
    idx_file: Option<Box<dyn IdxFileWriter>>,
    idx_file_offset: u64,
    /// The file holds bytes past `idx_file_offset` that must be trimmed
    /// before another row can land aligned.
    idx_torn: bool,
}

impl Default for CompactNeedleMap {
    fn default() -> Self {
        Self::new()
    }
}

impl CompactNeedleMap {
    /// Create a new empty in-memory map.
    pub fn new() -> Self {
        CompactNeedleMap {
            map: CompactMap::new(),
            metric: NeedleMapMetric::default(),
            idx_file: None,
            idx_file_offset: 0,
            idx_torn: false,
        }
    }

    /// Load from an .idx file, building the in-memory map.
    pub fn load_from_idx<R: Read + Seek>(reader: &mut R, version: Version) -> io::Result<Self> {
        let mut nm = CompactNeedleMap::new();
        idx::walk_index_file(reader, 0, |key, offset, size| {
            // A read-only load attaches no writer, so this is its only size.
            nm.idx_file_offset += NEEDLE_MAP_ENTRY_SIZE as u64;
            nm.metric.maybe_set_max_needle_end(offset, size, version);
            if offset.is_zero() || size.is_deleted() {
                nm.delete_from_map(key);
            } else {
                nm.set_internal(key, NeedleValue { offset, size });
            }
            Ok(())
        })?;
        Ok(nm)
    }

    /// Set the index file for append-only writes.
    pub fn set_idx_file(&mut self, file: Box<dyn IdxFileWriter>, offset: u64) {
        self.idx_file = Some(file);
        self.idx_file_offset = offset;
        self.idx_torn = false;
    }

    /// True when an .idx file writer is attached. A read-only load leaves
    /// this `false` — set_writable() must reattach a writer or subsequent
    /// puts silently skip the disk append.
    pub fn has_idx_writer(&self) -> bool {
        self.idx_file.is_some()
    }

    // ---- Map operations ----

    /// Insert or update an entry. Appends to .idx file if present.
    pub fn put(&mut self, key: NeedleId, offset: Offset, size: Size) -> io::Result<()> {
        // Persist to idx file BEFORE mutating in-memory state for crash consistency
        self.append_to_index_file(key, offset, size)?;
        if self.idx_file.is_some() {
            self.idx_file_offset += NEEDLE_MAP_ENTRY_SIZE as u64;
        }

        let old = self.map.get(key);
        self.metric.on_put(key, old.as_ref(), size);
        self.map.set(key, offset, size);
        Ok(())
    }

    /// Write one row to the .idx file at `idx_file_offset`. A row left
    /// half-written by a failed write is trimmed back to the offset so the
    /// next row still lands aligned; while the trim keeps failing no row is
    /// written at all, or it would sit off alignment and parse as garbage on
    /// load. The offset itself is advanced by the caller once the row counts.
    fn append_to_index_file(
        &mut self,
        key: NeedleId,
        offset: Offset,
        size: Size,
    ) -> io::Result<()> {
        let Some(idx_file) = self.idx_file.as_mut() else {
            return Ok(());
        };
        if self.idx_torn {
            match idx_file.truncate_to(self.idx_file_offset) {
                Ok(()) => self.idx_torn = false,
                Err(e) => {
                    return Err(io::Error::other(format!(
                        "index file still holds a torn row: {e}"
                    )));
                }
            }
        }
        idx_file.seek(io::SeekFrom::Start(self.idx_file_offset))?;
        if let Err(e) = idx::write_index_entry(idx_file, key, offset, size) {
            if let Err(te) = idx_file.truncate_to(self.idx_file_offset) {
                self.idx_torn = true;
                tracing::warn!("failed to trim torn .idx row: {}", te);
            }
            return Err(e);
        }
        Ok(())
    }

    /// Look up a needle.
    pub fn get(&self, key: NeedleId) -> Option<NeedleValue> {
        self.map.get(key)
    }

    /// Mark a needle as deleted. Appends tombstone to .idx file.
    /// Matches Go's NeedleMap.Delete: ALWAYS writes tombstone to idx and
    /// increments deletion counter, even if needle doesn't exist or is
    /// already deleted (important for replication).
    pub fn delete(&mut self, key: NeedleId, offset: Offset) -> io::Result<Option<Size>> {
        // Go unconditionally calls nm.m.Delete(), nm.logDelete(), nm.appendToIndexFile()
        let deleted_bytes = self.map.delete(key).unwrap_or(Size(0));

        // Match Go's logDelete -> LogDeletionCounter: only increment when oldSize > 0.
        // Go does NOT decrement FileCounter/FileByteCounter in Delete;
        // live counts are computed as FileCounter - DeletionCounter.
        if deleted_bytes.0 > 0 {
            self.metric.deletion_count.fetch_add(1, Ordering::Relaxed);
            self.metric
                .deletion_byte_count
                .fetch_add(deleted_bytes.0 as u64, Ordering::Relaxed);
        }

        // Always write tombstone to idx file (matching Go)
        self.append_to_index_file(key, offset, TOMBSTONE_FILE_SIZE)?;
        if self.idx_file.is_some() {
            self.idx_file_offset += NEEDLE_MAP_ENTRY_SIZE as u64;
        }

        if deleted_bytes.0 > 0 {
            Ok(Some(deleted_bytes))
        } else {
            Ok(None)
        }
    }

    // ---- Internal helpers ----

    /// Insert into map during loading (no idx file write).
    fn set_internal(&mut self, key: NeedleId, nv: NeedleValue) {
        let old = self.map.get(key);
        self.metric.on_put(key, old.as_ref(), nv.size);
        self.map.set(key, nv.offset, nv.size);
    }

    /// Remove from map during loading (handle deletions in idx walk).
    /// Matches Go's doLoading else branch: always increments DeletionCounter,
    /// and adds old size bytes to DeletionByteCounter.
    fn delete_from_map(&mut self, key: NeedleId) {
        self.metric.maybe_set_max_file_key(key);
        // Go's CompactMap.Delete returns old size (0 if not found or already deleted).
        // Go's doLoading always does DeletionCounter++ and DeletionByteCounter += uint64(oldSize).
        let old_size = self.map.get(key).map(|nv| nv.size).unwrap_or(Size(0));
        // Go unconditionally increments DeletionCounter
        self.metric.deletion_count.fetch_add(1, Ordering::Relaxed);
        // Go adds uint64(oldSize) which for valid sizes adds the value, for 0/negative adds 0
        if old_size.0 > 0 {
            self.metric
                .deletion_byte_count
                .fetch_add(old_size.0 as u64, Ordering::Relaxed);
        }
        self.map.remove(key);
    }

    // ---- Metrics accessors ----

    pub fn content_size(&self) -> u64 {
        self.metric.file_byte_count.load(Ordering::Relaxed)
    }

    pub fn deleted_size(&self) -> u64 {
        self.metric.deletion_byte_count.load(Ordering::Relaxed)
    }

    pub fn file_count(&self) -> i64 {
        self.metric.file_count.load(Ordering::Relaxed)
    }

    pub fn deleted_count(&self) -> i64 {
        self.metric.deletion_count.load(Ordering::Relaxed)
    }

    pub fn max_file_key(&self) -> NeedleId {
        NeedleId(self.metric.max_file_key.load(Ordering::Relaxed))
    }

    /// Largest (offset + actual size) seen during the load walk; 0 if the
    /// map is empty. See `NeedleMapMetric::maybe_set_max_needle_end`.
    pub fn max_needle_end(&self) -> i64 {
        self.metric.max_needle_end.load(Ordering::Relaxed)
    }

    pub fn index_file_size(&self) -> u64 {
        self.idx_file_offset
    }

    /// Sync index file to disk.
    pub fn sync(&self) -> io::Result<()> {
        if let Some(ref idx_file) = self.idx_file {
            idx_file.sync_all()?;
        }
        Ok(())
    }

    /// Close index file.
    pub fn close(&mut self) {
        let _ = self.sync();
        self.idx_file = None;
    }

    /// Save the in-memory map to an index file, sorted by needle ID ascending.
    pub fn save_to_idx(&self, path: &str) -> io::Result<()> {
        let mut file = std::fs::OpenOptions::new()
            .write(true)
            .create(true)
            .truncate(true)
            .open(path)?;

        self.map.ascending_visit(|id, nv| {
            if nv.size.is_valid() {
                idx::write_index_entry(&mut file, id, nv.offset, nv.size)
            } else {
                Ok(())
            }
        })?;
        file.sync_all()?;
        Ok(())
    }

    /// Visit all entries in ascending order by needle ID.
    pub fn ascending_visit<F, E>(&self, f: F) -> Result<(), E>
    where
        F: FnMut(NeedleId, &NeedleValue) -> Result<(), E>,
    {
        self.map.ascending_visit(f)
    }
}

// ============================================================================
// RedbNeedleMap (disk-backed via redb)
// ============================================================================

/// redb table: NeedleId (u64) -> packed [offset(4) + size(4)]
const NEEDLE_TABLE: TableDefinition<u64, &[u8]> = TableDefinition::new("needles");

/// Metadata table: stores the .idx file size that was used to build this redb.
/// Key "idx_size" -> u64 byte offset. Used to detect whether the .rdb can be
/// reused on restart or needs a full/incremental rebuild.
const META_TABLE: TableDefinition<&str, u64> = TableDefinition::new("meta");
const META_IDX_SIZE: &str = "idx_size";

/// Writes between two durable redb checkpoints. Every non-durable commit
/// leaves an entry in redb's transaction tracker and pins the pages that
/// were on disk at the last durable commit; only a durable commit clears
/// both. Without a cadence they grow for the life of the process (#11179).
/// The map only counts; the volume takes the checkpoint (see
/// `RedbNeedleMap::checkpoint_due`) because the .dat must be flushed first.
const REDB_CHECKPOINT_INTERVAL: u32 = 1000;

/// Disk-backed needle map using redb.
/// Low memory usage — data lives on disk behind a small, bounded redb page
/// cache sized by `NeedleMapKind::redb_cache_bytes`.
pub struct RedbNeedleMap {
    /// `Option` so a non-`TransactionPoisoned` commit error can drop the
    /// `Database` (releasing the file lock) before reopening from `.idx`.
    db: Option<Database>,
    /// Path of the `.rdb` file — needed to reopen after a commit error.
    rdb_path: String,
    /// Volume version, for replaying `.idx` rows on reopen.
    version: Version,
    /// redb page-cache size, for reopening with the same configuration.
    cache_bytes: usize,
    metric: NeedleMapMetric,
    idx_file: Option<Box<dyn IdxFileWriter>>,
    idx_file_offset: u64,
    /// The file holds bytes past `idx_file_offset` that must be trimmed
    /// before another row can land aligned.
    idx_torn: bool,
    /// Puts/deletes since the last durable checkpoint.
    writes_since_checkpoint: u32,
}

impl RedbNeedleMap {
    /// Begin a write transaction with `Durability::None` (no fsync).
    /// The .idx file is the source of truth for crash recovery: a crash
    /// loses at most the writes since the last checkpoint from redb, and
    /// the next load replays them from .idx.
    fn begin_write_no_fsync(db: &Database) -> io::Result<redb::WriteTransaction> {
        let mut txn = db
            .begin_write()
            .map_err(|e| io::Error::other(format!("redb begin_write: {}", e)))?;
        let _ = txn.set_durability(Durability::None);
        Ok(txn)
    }

    /// Return a reference to the open `Database`, or an error if a previous
    /// non-poisoned commit error followed by a failed reopen left it closed.
    fn db_or_err(&self) -> io::Result<&Database> {
        self.db.as_ref().ok_or_else(|| {
            io::Error::other(
                "redb database is closed after a commit error and failed reopen from .idx",
            )
        })
    }

    /// True once `REDB_CHECKPOINT_INTERVAL` puts/deletes have been committed
    /// non-durably. The caller (the volume) must then flush the .dat and
    /// call [`checkpoint`](Self::checkpoint): a checkpoint makes the index
    /// durable, so the bytes it points at have to be on disk before it.
    pub fn checkpoint_due(&self) -> bool {
        self.writes_since_checkpoint >= REDB_CHECKPOINT_INTERVAL
    }

    /// Make the table durable and record how much of the .idx it reflects.
    /// Precondition: the .dat the index points into has been flushed.
    ///
    /// When `sync_idx` is true the .idx file is fsynced first — the recorded
    /// size must never exceed what is on disk, or the reload would have to
    /// rebuild from scratch. A caller that has already fsynced the .idx (e.g.
    /// the volume's `flush_idx` on the fsync=true write path) may pass false
    /// to avoid a redundant fsync.
    pub fn checkpoint(&mut self, sync_idx: bool) -> io::Result<()> {
        let txn = self.begin_checkpoint(sync_idx)?;
        txn.commit()
            .map_err(|e| io::Error::other(format!("redb commit: {}", e)))?;
        self.writes_since_checkpoint = 0;
        Ok(())
    }

    /// Begin a durable transaction (fsync on commit) that also records how
    /// much of the .idx the table reflects, so a reload replays only the
    /// tail appended after it. When `sync_idx` is true the .idx is fsynced
    /// first (see [`checkpoint`](Self::checkpoint)).
    /// Durable commit uses `set_quick_repair(true)` so the next open after a
    /// crash does not scan the whole file.
    fn begin_checkpoint(&self, sync_idx: bool) -> io::Result<redb::WriteTransaction> {
        if sync_idx {
            self.sync()?;
        }
        let mut txn = self
            .db_or_err()?
            .begin_write()
            .map_err(|e| io::Error::other(format!("redb begin_write: {}", e)))?;
        txn.set_quick_repair(true);
        if self.idx_file.is_some() {
            let mut meta = txn
                .open_table(META_TABLE)
                .map_err(|e| io::Error::other(format!("redb open meta: {}", e)))?;
            meta.insert(META_IDX_SIZE, self.idx_file_offset)
                .map_err(|e| io::Error::other(format!("redb insert meta: {}", e)))?;
        }
        Ok(txn)
    }

    /// Create a new redb-backed needle map at the given path.
    /// The database file will be created if it does not exist.
    /// `cache_bytes` bounds redb's page cache for this one database.
    pub fn new(db_path: &str, cache_bytes: usize) -> io::Result<Self> {
        let db = Database::builder()
            .set_cache_size(cache_bytes)
            .create(db_path)
            .map_err(|e| io::Error::other(format!("redb create error: {}", e)))?;

        // Ensure tables exist
        let txn = Self::begin_write_no_fsync(&db)?;
        {
            let _table = txn
                .open_table(NEEDLE_TABLE)
                .map_err(|e| io::Error::other(format!("redb open_table: {}", e)))?;
            let _meta = txn
                .open_table(META_TABLE)
                .map_err(|e| io::Error::other(format!("redb open_table meta: {}", e)))?;
        }
        txn.commit()
            .map_err(|e| io::Error::other(format!("redb commit: {}", e)))?;

        Ok(RedbNeedleMap {
            db: Some(db),
            rdb_path: db_path.to_string(),
            version: Version::current(),
            cache_bytes,
            metric: NeedleMapMetric::default(),
            idx_file: None,
            idx_file_offset: 0,
            idx_torn: false,
            writes_since_checkpoint: 0,
        })
    }

    /// Save the .idx file size into redb metadata so we can detect whether
    /// the .rdb is up-to-date on the next startup.
    fn save_idx_size_meta(&self, idx_size: u64) -> io::Result<()> {
        let txn = Self::begin_write_no_fsync(self.db_or_err()?)?;
        {
            let mut meta = txn
                .open_table(META_TABLE)
                .map_err(|e| io::Error::other(format!("redb open meta: {}", e)))?;
            meta.insert(META_IDX_SIZE, idx_size)
                .map_err(|e| io::Error::other(format!("redb insert meta: {}", e)))?;
        }
        txn.commit()
            .map_err(|e| io::Error::other(format!("redb commit meta: {}", e)))?;
        Ok(())
    }

    /// Read the stored .idx file size from redb metadata.
    fn read_idx_size_meta(&self) -> io::Result<Option<u64>> {
        let txn = self
            .db_or_err()?
            .begin_read()
            .map_err(|e| io::Error::other(format!("redb begin_read: {}", e)))?;
        let meta = txn
            .open_table(META_TABLE)
            .map_err(|e| io::Error::other(format!("redb open meta: {}", e)))?;
        // experimental-api-5 drops inherent ReadOnlyTable::get ('static guard).
        // ReadableTable::get guard borrows `meta`; edition 2024 drops the tail
        // expression's temporaries before `meta`, so no extra binding is needed.
        match meta.get(META_IDX_SIZE) {
            Ok(Some(guard)) => Ok(Some(guard.value())),
            Ok(None) => Ok(None),
            Err(e) => Err(io::Error::other(format!("redb get meta: {}", e))),
        }
    }

    /// Test-only read of META `idx_size` through the live handle. See
    /// [`test_support::live_meta_idx_size`] for why durability tests use
    /// this instead of copying the open `.rdb`.
    #[cfg(test)]
    pub(crate) fn live_meta_idx_size(&self) -> Option<u64> {
        self.read_idx_size_meta().unwrap()
    }

    /// Load from an .idx file, reusing an existing .rdb if it is consistent.
    ///
    /// Strategy:
    /// 1. Try to open existing .rdb and read its stored .idx size
    /// 2. If .idx size matches → reuse .rdb, rebuild metrics from scan
    /// 3. If .idx is larger → replay new entries incrementally
    /// 4. Otherwise (missing, corrupted, .idx smaller) → full rebuild
    pub fn load_from_idx<R: Read + Seek>(
        db_path: &str,
        reader: &mut R,
        version: Version,
        cache_bytes: usize,
    ) -> io::Result<Self> {
        let idx_size = reader.seek(io::SeekFrom::End(0))?;
        reader.seek(io::SeekFrom::Start(0))?;

        // Try to reuse existing .rdb
        if Path::new(db_path).exists() {
            if let Ok(nm) = Self::try_reuse_rdb(db_path, reader, idx_size, version, cache_bytes) {
                return Ok(nm);
            }
            // Reuse failed — fall through to full rebuild
            reader.seek(io::SeekFrom::Start(0))?;
        }

        Self::full_rebuild(db_path, reader, idx_size, version, cache_bytes)
    }

    /// Try to reuse an existing .rdb file. Returns Ok if successful,
    /// Err if a full rebuild is needed.
    fn try_reuse_rdb<R: Read + Seek>(
        db_path: &str,
        reader: &mut R,
        idx_size: u64,
        version: Version,
        cache_bytes: usize,
    ) -> io::Result<Self> {
        let db = Database::builder()
            .set_cache_size(cache_bytes)
            .open(db_path)
            .map_err(|e| io::Error::other(format!("redb open: {}", e)))?;

        let mut nm = RedbNeedleMap {
            db: Some(db),
            rdb_path: db_path.to_string(),
            version,
            cache_bytes,
            metric: NeedleMapMetric::default(),
            idx_file: None,
            idx_file_offset: 0,
            idx_torn: false,
            writes_since_checkpoint: 0,
        };

        let stored_idx_size = nm
            .read_idx_size_meta()?
            .ok_or_else(|| io::Error::other("no idx_size in redb meta"))?;

        if stored_idx_size > idx_size {
            // .idx shrank — corrupted or truncated, need full rebuild
            return Err(io::Error::other("idx file smaller than stored size"));
        }

        // Counters come from the whole .idx history, never from the table,
        // so the replay below is free to re-apply rows the table already
        // holds (redb flushes on drop even without a checkpoint).
        nm.metric = metrics_from_idx(reader, version)?;

        if stored_idx_size < idx_size {
            // .idx grew — replay new entries incrementally
            let start_entry = stored_idx_size / NEEDLE_MAP_ENTRY_SIZE as u64;
            let txn = Self::begin_write_no_fsync(nm.db.as_ref().unwrap())?;
            {
                let mut table = txn
                    .open_table(NEEDLE_TABLE)
                    .map_err(|e| io::Error::other(format!("redb open_table: {}", e)))?;
                idx::walk_index_file(reader, start_entry, |key, offset, size| {
                    let key_u64: u64 = key.into();
                    if offset.is_zero() || size.is_deleted() {
                        // Delete: store a tombstone (negative size, original
                        // offset) over a live value; already deleted is a no-op.
                        if let Ok(Some(old)) = nm.get_via_table(&table, key_u64)
                            && old.size.is_valid()
                        {
                            let deleted_nv = NeedleValue {
                                offset: old.offset,
                                size: Size(-(old.size.0)),
                            };
                            let packed = pack_needle_value(&deleted_nv);
                            table
                                .insert(key_u64, packed.as_slice())
                                .map_err(|e| io::Error::other(format!("redb insert: {}", e)))?;
                        }
                    } else {
                        let packed = pack_needle_value(&NeedleValue { offset, size });
                        table
                            .insert(key_u64, packed.as_slice())
                            .map_err(|e| io::Error::other(format!("redb insert: {}", e)))?;
                    }
                    Ok(())
                })?;
            }
            txn.commit()
                .map_err(|e| io::Error::other(format!("redb commit: {}", e)))?;

            nm.save_idx_size_meta(idx_size)?;
        }

        Ok(nm)
    }

    /// Look up a needle value using an already-open table reference.
    /// Used during incremental replay to avoid opening separate read transactions.
    fn get_via_table(
        &self,
        table: &redb::Table<u64, &[u8]>,
        key_u64: u64,
    ) -> io::Result<Option<NeedleValue>> {
        match table.get(key_u64) {
            Ok(Some(guard)) => Ok(packed_to_needle_value(guard.value())),
            Ok(None) => Ok(None),
            Err(e) => Err(io::Error::other(format!("redb get: {}", e))),
        }
    }

    /// Full rebuild: prefer a fresh `.rdb`, then rebuild from the entire `.idx`.
    ///
    /// `Database::create` opens an existing file (`truncate(false)`). Unlink
    /// is best-effort: if it fails (sticky bit, foreign owner) but the path is
    /// still writable, clear the leftover needles table before inserting.
    /// Live keys come from a `BTreeMap` so inserts are in needle-id order and
    /// redb 4.2.0 packs leaves, without a second Vec + sort scratch.
    fn full_rebuild<R: Read + Seek>(
        db_path: &str,
        reader: &mut R,
        idx_size: u64,
        version: Version,
        cache_bytes: usize,
    ) -> io::Result<Self> {
        let unlinked = match std::fs::remove_file(db_path) {
            Ok(()) => true,
            Err(e) if e.kind() == io::ErrorKind::NotFound => true,
            Err(e) => {
                tracing::warn!("redb unlink before rebuild failed: {}", e);
                false
            }
        };

        let mut nm = match RedbNeedleMap::new(db_path, cache_bytes) {
            Ok(nm) => nm,
            Err(e) => {
                let _ = std::fs::remove_file(db_path);
                return Err(e);
            }
        };
        nm.version = version;

        let result = (|| -> io::Result<()> {
            // Seek independently of walk_index_file. Run before the write txn
            // so a metric-read error does not unlink a committed rebuild.
            nm.metric = metrics_from_idx(reader, version)?;

            let mut entries: BTreeMap<NeedleId, NeedleValue> = BTreeMap::new();
            idx::walk_index_file(reader, 0, |key, offset, size| {
                if offset.is_zero() || size.is_deleted() {
                    entries.remove(&key);
                } else {
                    entries.insert(key, NeedleValue { offset, size });
                }
                Ok(())
            })?;

            let txn = Self::begin_write_no_fsync(nm.db.as_ref().unwrap())?;
            {
                let mut table = txn
                    .open_table(NEEDLE_TABLE)
                    .map_err(|e| io::Error::other(format!("redb open_table: {}", e)))?;
                if !unlinked {
                    table
                        .retain(|_, _| false)
                        .map_err(|e| io::Error::other(format!("redb retain: {}", e)))?;
                }

                #[cfg(not(feature = "redb-experimental-cursor"))]
                {
                    for (key, nv) in &entries {
                        let key_u64: u64 = (*key).into();
                        let packed = pack_needle_value(nv);
                        table
                            .insert(key_u64, packed.as_slice())
                            .map_err(|e| io::Error::other(format!("redb insert: {}", e)))?;
                    }
                }
                #[cfg(feature = "redb-experimental-cursor")]
                {
                    let mut cursor =
                        table
                            .upper_bound_mut(Bound::<u64>::Unbounded)
                            .map_err(|e| {
                                io::Error::new(
                                    io::ErrorKind::Other,
                                    format!("redb upper_bound_mut: {}", e),
                                )
                            })?;
                    for (key, nv) in &entries {
                        let key_u64: u64 = (*key).into();
                        let packed = pack_needle_value(nv);
                        cursor
                            .insert_before(key_u64, packed.as_slice())
                            .map_err(|e| {
                                io::Error::new(
                                    io::ErrorKind::Other,
                                    format!("redb insert_before: {}", e),
                                )
                            })?;
                    }
                    cursor.close().map_err(|e| {
                        io::Error::new(io::ErrorKind::Other, format!("redb cursor close: {}", e))
                    })?;
                }
            }
            txn.commit()
                .map_err(|e| io::Error::other(format!("redb commit: {}", e)))?;

            nm.save_idx_size_meta(idx_size)?;
            Ok(())
        })();

        match result {
            Ok(()) => Ok(nm),
            Err(e) => {
                drop(nm);
                let _ = std::fs::remove_file(db_path);
                Err(e)
            }
        }
    }

    /// Set the index file for append-only writes.
    pub fn set_idx_file(&mut self, file: Box<dyn IdxFileWriter>, offset: u64) {
        self.idx_file = Some(file);
        self.idx_file_offset = offset;
        self.idx_torn = false;
    }

    /// True when an .idx file writer is attached. See CompactNeedleMap.
    pub fn has_idx_writer(&self) -> bool {
        self.idx_file.is_some()
    }

    // ---- Map operations ----

    /// Insert or update an entry. Writes to idx file first, then redb.
    pub fn put(&mut self, key: NeedleId, offset: Offset, size: Size) -> io::Result<()> {
        // Persist to idx file BEFORE mutating redb state for crash consistency.
        // The offset is advanced only after the redb commit succeeds: a failed
        // commit leaves an orphan row in .idx that redb doesn't reflect, and
        // advancing the offset here would let a later checkpoint record it as
        // reflected, making the reload skip it permanently.
        self.append_to_index_file(key, offset, size)?;

        let key_u64: u64 = key.into();
        let packed = pack_needle_value(&NeedleValue { offset, size });

        let old = {
            let db = match self.db_or_err() {
                Ok(db) => db,
                Err(e) => {
                    self.truncate_idx_to_offset();
                    return Err(e);
                }
            };
            let txn = Self::begin_write_no_fsync(db);
            let txn = match txn {
                Ok(txn) => txn,
                Err(e) => {
                    self.truncate_idx_to_offset();
                    return Err(e);
                }
            };
            // Insert and extract the old value inside a block so the table
            // and AccessGuard are dropped before commit (which moves txn).
            let old = {
                let mut table = match txn.open_table(NEEDLE_TABLE) {
                    Ok(t) => t,
                    Err(e) => {
                        self.truncate_idx_to_offset();
                        return Err(io::Error::other(format!("redb open_table: {}", e)));
                    }
                };
                match table.insert(key_u64, packed.as_slice()) {
                    Ok(prev) => prev.and_then(|g| packed_to_needle_value(g.value())),
                    Err(e) => {
                        self.truncate_idx_to_offset();
                        return Err(io::Error::other(format!("redb insert: {}", e)));
                    }
                }
            };
            match txn.commit() {
                Ok(()) => old,
                Err(redb::CommitError::TransactionPoisoned) => {
                    // Transaction rolled back, database still usable:
                    // truncate the orphan .idx row.
                    self.truncate_idx_to_offset();
                    return Err(io::Error::other(
                        "redb commit: Transaction was poisoned by a panic",
                    ));
                }
                Err(e) => {
                    // Non-poisoned commit error: the change may be
                    // visible and redb refuses further writes. Keep
                    // the .idx row (do NOT truncate) and reopen from
                    // .idx to repair redb's internal state.
                    let err = io::Error::other(format!("redb commit: {}", e));
                    if let Err(reopen_err) = self.reopen_from_idx() {
                        tracing::warn!("redb reopen after put commit error failed: {}", reopen_err);
                    }
                    return Err(err);
                }
            }
        };

        if self.idx_file.is_some() {
            self.idx_file_offset += NEEDLE_MAP_ENTRY_SIZE as u64;
        }
        self.writes_since_checkpoint = self.writes_since_checkpoint.saturating_add(1);
        self.metric.on_put(key, old.as_ref(), size);
        Ok(())
    }

    /// Write one row to the .idx file at `idx_file_offset`. A row left
    /// half-written by a failed write is trimmed back to the offset so the
    /// next row still lands aligned; while the trim keeps failing no row is
    /// written at all, or it would sit off alignment and parse as garbage on
    /// load. The offset itself is advanced by the caller once the row counts.
    fn append_to_index_file(
        &mut self,
        key: NeedleId,
        offset: Offset,
        size: Size,
    ) -> io::Result<()> {
        let Some(idx_file) = self.idx_file.as_mut() else {
            return Ok(());
        };
        if self.idx_torn {
            match idx_file.truncate_to(self.idx_file_offset) {
                Ok(()) => self.idx_torn = false,
                Err(e) => {
                    return Err(io::Error::other(format!(
                        "index file still holds a torn row: {e}"
                    )));
                }
            }
        }
        idx_file.seek(io::SeekFrom::Start(self.idx_file_offset))?;
        if let Err(e) = idx::write_index_entry(idx_file, key, offset, size) {
            if let Err(te) = idx_file.truncate_to(self.idx_file_offset) {
                self.idx_torn = true;
                tracing::warn!("failed to trim torn .idx row: {}", te);
            }
            return Err(e);
        }
        Ok(())
    }

    /// Look up a needle. A redb failure is an ERROR, not an absent needle:
    /// answering "not found" would turn a database problem into a read miss
    /// and let a delete report success without recording a tombstone.
    pub fn get(&self, key: NeedleId) -> io::Result<Option<NeedleValue>> {
        let key_u64: u64 = key.into();
        self.get_internal(key_u64)
    }

    /// Internal get that returns io::Result for error propagation.
    fn get_internal(&self, key_u64: u64) -> io::Result<Option<NeedleValue>> {
        let txn = self
            .db_or_err()?
            .begin_read()
            .map_err(|e| io::Error::other(format!("redb begin_read: {}", e)))?;
        let table = txn
            .open_table(NEEDLE_TABLE)
            .map_err(|e| io::Error::other(format!("redb open_table: {}", e)))?;
        // experimental-api-5 drops inherent ReadOnlyTable::get ('static guard).
        // ReadableTable::get guard borrows `table`; edition 2024 drops the tail
        // expression's temporaries before `table`, so no extra binding is needed.
        match table.get(key_u64) {
            Ok(Some(guard)) => Ok(packed_to_needle_value(guard.value())),
            Ok(None) => Ok(None),
            Err(e) => Err(io::Error::other(format!("redb get: {}", e))),
        }
    }

    /// Mark a needle as deleted. Appends tombstone to .idx file, negates size in redb.
    pub fn delete(&mut self, key: NeedleId, offset: Offset) -> io::Result<Option<Size>> {
        let key_u64: u64 = key.into();
        let txn = Self::begin_write_no_fsync(self.db_or_err()?)?;
        let mut table = txn
            .open_table(NEEDLE_TABLE)
            .map_err(|e| io::Error::other(format!("redb open_table: {}", e)))?;
        let old = match table.get(key_u64) {
            Ok(Some(guard)) => packed_to_needle_value(guard.value()),
            Ok(None) => None,
            Err(e) => {
                return Err(io::Error::other(format!("redb get: {}", e)));
            }
        };
        let Some(old) = old.filter(|nv| nv.size.is_valid()) else {
            drop(table);
            return Ok(None);
        };

        self.append_to_index_file(key, offset, TOMBSTONE_FILE_SIZE)?;

        let deleted_nv = NeedleValue {
            offset: old.offset,
            size: Size(-(old.size.0)),
        };
        let packed = pack_needle_value(&deleted_nv);
        let insert_res = table.insert(key_u64, packed.as_slice()).map(|_| ());
        drop(table);
        if let Err(e) = insert_res {
            self.truncate_idx_to_offset();
            return Err(io::Error::other(format!("redb insert: {}", e)));
        }
        match txn.commit() {
            Ok(()) => {}
            Err(redb::CommitError::TransactionPoisoned) => {
                // Transaction rolled back, database still usable:
                // truncate the orphan .idx row.
                self.truncate_idx_to_offset();
                return Err(io::Error::other(
                    "redb commit: Transaction was poisoned by a panic",
                ));
            }
            Err(e) => {
                // Non-poisoned commit error: the change may be visible
                // and redb refuses further writes. Keep the .idx row
                // (do NOT truncate) and reopen from .idx to repair
                // redb's internal state.
                let err = io::Error::other(format!("redb commit: {}", e));
                if let Err(reopen_err) = self.reopen_from_idx() {
                    tracing::warn!(
                        "redb reopen after delete commit error failed: {}",
                        reopen_err
                    );
                }
                return Err(err);
            }
        }
        if self.idx_file.is_some() {
            self.idx_file_offset += NEEDLE_MAP_ENTRY_SIZE as u64;
        }
        self.writes_since_checkpoint = self.writes_since_checkpoint.saturating_add(1);
        self.metric.on_delete(&old);
        Ok(Some(old.size))
    }

    // ---- Metrics accessors ----

    pub fn content_size(&self) -> u64 {
        self.metric.file_byte_count.load(Ordering::Relaxed)
    }

    pub fn deleted_size(&self) -> u64 {
        self.metric.deletion_byte_count.load(Ordering::Relaxed)
    }

    pub fn file_count(&self) -> i64 {
        self.metric.file_count.load(Ordering::Relaxed)
    }

    pub fn deleted_count(&self) -> i64 {
        self.metric.deletion_count.load(Ordering::Relaxed)
    }

    pub fn max_file_key(&self) -> NeedleId {
        NeedleId(self.metric.max_file_key.load(Ordering::Relaxed))
    }

    /// Largest (offset + actual size) seen during the load walk; 0 if the
    /// map is empty. See `NeedleMapMetric::maybe_set_max_needle_end`.
    pub fn max_needle_end(&self) -> i64 {
        self.metric.max_needle_end.load(Ordering::Relaxed)
    }

    pub fn index_file_size(&self) -> u64 {
        self.idx_file_offset
    }

    /// Sync index file to disk.
    pub fn sync(&self) -> io::Result<()> {
        if let Some(ref idx_file) = self.idx_file {
            idx_file.sync_all()?;
        }
        Ok(())
    }

    /// Remove any .idx bytes past `idx_file_offset` — the orphan row left by
    /// a failed redb commit. Without this the next successful write appends
    /// after the orphan, `idx_file_offset` advances past it, and a later
    /// checkpoint records an offset that makes the reload skip the orphan.
    /// When the trim fails the file is latched torn so no later row lands
    /// after bytes the map does not reflect.
    fn truncate_idx_to_offset(&mut self) {
        if let Some(ref mut idx_file) = self.idx_file
            && let Err(e) = idx_file.truncate_to(self.idx_file_offset)
        {
            self.idx_torn = true;
            tracing::warn!("failed to truncate orphan .idx row: {}", e);
        }
    }

    /// Reopen the database from `.idx` after a non-`TransactionPoisoned`
    /// commit error.
    ///
    /// redb 4.2.0 can make a `Durability::None` commit visible *before*
    /// returning `Err(CommitError::Storage(..))`. In that state the database
    /// refuses further write transactions, so the map cannot be used as-is.
    /// The `.idx` row that triggered the commit is **not** truncated — the
    /// change may already be in the table — and the database is closed and
    /// reopened from `.idx`, which repairs redb's internal state and replays
    /// any tail the failed commit did not apply.
    ///
    /// The `.idx` writer is preserved and `idx_file_offset` is advanced to
    /// the on-disk `.idx` size so the preserved row is part of the replay
    /// watermark.
    fn reopen_from_idx(&mut self) -> io::Result<()> {
        let idx_path = self
            .rdb_path
            .strip_suffix(".rdb")
            .map(|base| format!("{base}.idx"))
            .ok_or_else(|| {
                io::Error::other(format!(
                    "rdb path lacks .rdb suffix, cannot derive .idx path: {}",
                    self.rdb_path
                ))
            })?;

        // Close the current database first — redb holds an exclusive file
        // lock, so load_from_idx cannot open the same path until we drop it.
        let old_db = self.db.take();
        drop(old_db);

        let read_file = std::fs::OpenOptions::new()
            .read(true)
            .open(&idx_path)
            .map_err(|e| io::Error::other(format!("reopen: open .idx {}: {}", idx_path, e)))?;
        let actual_idx_size = read_file.metadata()?.len();
        let mut reader = io::BufReader::new(read_file);

        let reopened =
            Self::load_from_idx(&self.rdb_path, &mut reader, self.version, self.cache_bytes)?;

        // Preserve the append writer and the paths/version/cache; adopt the
        // repaired database, metrics, and idx_file_offset from the reload.
        self.db = reopened.db;
        self.metric = reopened.metric;
        self.idx_file_offset = actual_idx_size;
        self.idx_torn = reopened.idx_torn;
        // The reopen replayed all rows since the last durable checkpoint
        // non-durably; start the counter fresh.
        self.writes_since_checkpoint = 0;
        Ok(())
    }

    /// Close the index file, checkpointing first so a reload starts from
    /// the recorded .idx size instead of replaying entries the table
    /// already holds.
    pub fn close(&mut self) {
        if let Err(e) = self.checkpoint(true) {
            tracing::warn!("redb checkpoint on close failed: {}", e);
        }
        self.idx_file = None;
    }

    /// Sync the .idx and drop the writer without taking a durable checkpoint.
    /// Used when the .dat flush failed: a checkpoint would make the index
    /// durable with entries that may point past the unflushed .dat tail, so
    /// the reload's max_needle_end check would mark the volume read-only.
    /// Without the checkpoint, META_IDX_SIZE stays at the last successful one
    /// and the reload replays the uncheckpointed tail (redb flushes on drop).
    pub fn close_without_checkpoint(&mut self) {
        if let Err(e) = self.sync() {
            tracing::warn!("redb idx sync on close failed: {}", e);
        }
        self.idx_file = None;
    }

    /// Save the redb contents to an index file, sorted by needle ID ascending.
    pub fn save_to_idx(&self, path: &str) -> io::Result<()> {
        let txn = self
            .db_or_err()?
            .begin_read()
            .map_err(|e| io::Error::other(format!("redb begin_read: {}", e)))?;
        let table = txn
            .open_table(NEEDLE_TABLE)
            .map_err(|e| io::Error::other(format!("redb open_table: {}", e)))?;

        let mut file = std::fs::OpenOptions::new()
            .write(true)
            .create(true)
            .truncate(true)
            .open(path)?;

        // redb iterates in key order (u64 ascending)
        let iter = table
            .iter()
            .map_err(|e| io::Error::other(format!("redb iter: {}", e)))?;

        for entry in iter {
            let (key_guard, val_guard) =
                entry.map_err(|e| io::Error::other(format!("redb iter next: {}", e)))?;
            let key_u64: u64 = key_guard.value();
            let bytes: &[u8] = val_guard.value();
            if let Some(nv) = packed_to_needle_value(bytes)
                && nv.size.is_valid()
            {
                idx::write_index_entry(&mut file, NeedleId(key_u64), nv.offset, nv.size)?;
            }
        }
        file.sync_all()?;
        Ok(())
    }

    /// Visit all entries in ascending order by needle ID.
    pub fn ascending_visit<F, E>(&self, mut f: F) -> Result<(), E>
    where
        F: FnMut(NeedleId, &NeedleValue) -> Result<(), E>,
        E: From<String>,
    {
        let txn = self
            .db_or_err()
            .map_err(|e| format!("redb begin_read: {}", e))?
            .begin_read()
            .map_err(|e| format!("redb begin_read: {}", e))?;
        let table = txn
            .open_table(NEEDLE_TABLE)
            .map_err(|e| format!("redb open_table: {}", e))?;
        let iter = table.iter().map_err(|e| format!("redb iter: {}", e))?;

        for entry in iter {
            let (key_guard, val_guard) = entry.map_err(|e| format!("redb iter next: {}", e))?;
            let key_u64: u64 = key_guard.value();
            let bytes: &[u8] = val_guard.value();
            if let Some(nv) = packed_to_needle_value(bytes) {
                f(NeedleId(key_u64), &nv)?;
            }
        }
        Ok(())
    }

    /// Collect all entries as a Vec for iteration (used by volume.rs iter patterns).
    pub fn collect_entries(&self) -> io::Result<Vec<(NeedleId, NeedleValue)>> {
        let mut result = Vec::new();
        let txn: redb::ReadTransaction = self
            .db_or_err()?
            .begin_read()
            .map_err(|e| io::Error::other(format!("redb begin_read: {e}")))?;
        let table = txn
            .open_table(NEEDLE_TABLE)
            .map_err(|e| io::Error::other(format!("redb open_table: {e}")))?;
        let iter = table
            .iter()
            .map_err(|e| io::Error::other(format!("redb iter: {e}")))?;
        for entry in iter {
            let (key_guard, val_guard) =
                entry.map_err(|e| io::Error::other(format!("redb entry: {e}")))?;
            let key_u64: u64 = key_guard.value();
            let bytes: &[u8] = val_guard.value();
            if let Some(nv) = packed_to_needle_value(bytes) {
                result.push((NeedleId(key_u64), nv));
            }
        }
        Ok(result)
    }
}

// ============================================================================
// NeedleMap enum — unified interface over both implementations
// ============================================================================

/// Unified needle map wrapping either in-memory or redb-backed storage.
pub enum NeedleMap {
    InMemory(CompactNeedleMap),
    Redb(RedbNeedleMap),
    /// Read-only volumes — including every cloud-tiered one — search the sorted
    /// `.sdx` on disk instead of holding an index in RAM. Mirrors Go's
    /// `SortedFileNeedleMap`.
    SortedFile(SortedFileNeedleMap),
}

impl NeedleMap {
    /// Insert or update an entry.
    pub fn put(&mut self, key: NeedleId, offset: Offset, size: Size) -> io::Result<()> {
        match self {
            NeedleMap::InMemory(nm) => nm.put(key, offset, size),
            NeedleMap::Redb(nm) => nm.put(key, offset, size),
            NeedleMap::SortedFile(nm) => nm.put(key, offset, size),
        }
    }

    /// Look up a needle. Disk- and database-backed maps report their own
    /// failures rather than folding them into "not found" — see the notes on
    /// `RedbNeedleMap::get` and `SortedFileNeedleMap::get`.
    pub fn get(&self, key: NeedleId) -> io::Result<Option<NeedleValue>> {
        match self {
            NeedleMap::InMemory(nm) => Ok(nm.get(key)),
            NeedleMap::Redb(nm) => nm.get(key),
            NeedleMap::SortedFile(nm) => nm.get(key),
        }
    }

    /// Mark a needle as deleted.
    pub fn delete(&mut self, key: NeedleId, offset: Offset) -> io::Result<Option<Size>> {
        match self {
            NeedleMap::InMemory(nm) => nm.delete(key, offset),
            NeedleMap::Redb(nm) => nm.delete(key, offset),
            NeedleMap::SortedFile(nm) => nm.delete(key, offset),
        }
    }

    /// Set the index file for append-only writes.
    pub fn set_idx_file(&mut self, file: Box<dyn IdxFileWriter>, offset: u64) {
        match self {
            NeedleMap::InMemory(nm) => nm.set_idx_file(file, offset),
            NeedleMap::Redb(nm) => nm.set_idx_file(file, offset),
            // The sorted map borrows its .idx per append, so there is no
            // long-lived writer to install.
            NeedleMap::SortedFile(_) => {}
        }
    }

    /// True when an .idx file writer is attached.
    pub fn has_idx_writer(&self) -> bool {
        match self {
            NeedleMap::InMemory(nm) => nm.has_idx_writer(),
            NeedleMap::Redb(nm) => nm.has_idx_writer(),
            // Appends open the .idx on demand, so one is always available.
            NeedleMap::SortedFile(_) => true,
        }
    }

    /// Whether the backend wants a durable checkpoint. Only the redb map
    /// commits non-durably between checkpoints; see
    /// `RedbNeedleMap::checkpoint_due`.
    pub fn checkpoint_due(&self) -> bool {
        match self {
            NeedleMap::Redb(nm) => nm.checkpoint_due(),
            NeedleMap::InMemory(_) | NeedleMap::SortedFile(_) => false,
        }
    }

    /// Take the checkpoint `checkpoint_due` asked for. The caller must have
    /// flushed the .dat first. Pass `sync_idx = false` when the .idx has
    /// already been fsynced (e.g. by `flush_idx` on the fsync=true path) to
    /// avoid a redundant fsync.
    pub fn checkpoint(&mut self, sync_idx: bool) -> io::Result<()> {
        match self {
            NeedleMap::Redb(nm) => nm.checkpoint(sync_idx),
            NeedleMap::InMemory(_) | NeedleMap::SortedFile(_) => Ok(()),
        }
    }

    /// Content byte count.
    pub fn content_size(&self) -> u64 {
        match self {
            NeedleMap::InMemory(nm) => nm.content_size(),
            NeedleMap::Redb(nm) => nm.content_size(),
            NeedleMap::SortedFile(nm) => nm.content_size(),
        }
    }

    /// Deleted byte count.
    pub fn deleted_size(&self) -> u64 {
        match self {
            NeedleMap::InMemory(nm) => nm.deleted_size(),
            NeedleMap::Redb(nm) => nm.deleted_size(),
            NeedleMap::SortedFile(nm) => nm.deleted_size(),
        }
    }

    /// Live file count.
    pub fn file_count(&self) -> i64 {
        match self {
            NeedleMap::InMemory(nm) => nm.file_count(),
            NeedleMap::Redb(nm) => nm.file_count(),
            NeedleMap::SortedFile(nm) => nm.file_count(),
        }
    }

    /// Deleted file count.
    pub fn deleted_count(&self) -> i64 {
        match self {
            NeedleMap::InMemory(nm) => nm.deleted_count(),
            NeedleMap::Redb(nm) => nm.deleted_count(),
            NeedleMap::SortedFile(nm) => nm.deleted_count(),
        }
    }

    /// Maximum needle ID seen.
    pub fn max_file_key(&self) -> NeedleId {
        match self {
            NeedleMap::InMemory(nm) => nm.max_file_key(),
            NeedleMap::Redb(nm) => nm.max_file_key(),
            NeedleMap::SortedFile(nm) => nm.max_file_key(),
        }
    }

    /// Skew the live file count away from what the `.idx` holds, so tests
    /// can build a volume whose reported count disagrees with a reload.
    #[cfg(test)]
    pub(crate) fn add_file_count_for_test(&self, delta: i64) {
        let metric = match self {
            NeedleMap::InMemory(nm) => &nm.metric,
            NeedleMap::Redb(nm) => &nm.metric,
            NeedleMap::SortedFile(_) => panic!("sorted-file needle maps are read-only"),
        };
        metric.file_count.fetch_add(delta, Ordering::Relaxed);
    }

    #[cfg(test)]
    pub(crate) fn add_deletion_count_for_test(&self, delta: i64) {
        let metric = match self {
            NeedleMap::InMemory(nm) => &nm.metric,
            NeedleMap::Redb(nm) => &nm.metric,
            NeedleMap::SortedFile(_) => panic!("sorted-file needle maps are read-only"),
        };
        metric.deletion_count.fetch_add(delta, Ordering::Relaxed);
    }

    /// Largest (offset + actual size) seen during the load walk; 0 if the
    /// map is empty. Used at volume load to detect .idx entries that
    /// reference past the end of .dat (issue #8928) without a second scan.
    pub fn max_needle_end(&self) -> i64 {
        match self {
            NeedleMap::InMemory(nm) => nm.max_needle_end(),
            NeedleMap::Redb(nm) => nm.max_needle_end(),
            NeedleMap::SortedFile(nm) => nm.max_needle_end(),
        }
    }

    /// Index file size in bytes.
    pub fn index_file_size(&self) -> u64 {
        match self {
            NeedleMap::InMemory(nm) => nm.index_file_size(),
            NeedleMap::Redb(nm) => nm.index_file_size(),
            NeedleMap::SortedFile(nm) => nm.index_file_size(),
        }
    }

    /// Sync index file to disk.
    pub fn sync(&self) -> io::Result<()> {
        match self {
            NeedleMap::InMemory(nm) => nm.sync(),
            NeedleMap::Redb(nm) => nm.sync(),
            NeedleMap::SortedFile(nm) => nm.sync(),
        }
    }

    /// Close index file.
    pub fn close(&mut self) {
        match self {
            NeedleMap::InMemory(nm) => nm.close(),
            NeedleMap::Redb(nm) => nm.close(),
            NeedleMap::SortedFile(nm) => nm.close(),
        }
    }

    /// Close without checkpointing — sync the .idx and drop the writer only.
    /// See [`RedbNeedleMap::close_without_checkpoint`].
    pub fn close_without_checkpoint(&mut self) {
        match self {
            NeedleMap::InMemory(nm) => nm.close(),
            NeedleMap::Redb(nm) => nm.close_without_checkpoint(),
            NeedleMap::SortedFile(nm) => nm.close(),
        }
    }

    /// Save to an index file.
    pub fn save_to_idx(&self, path: &str) -> io::Result<()> {
        match self {
            NeedleMap::InMemory(nm) => nm.save_to_idx(path),
            NeedleMap::Redb(nm) => nm.save_to_idx(path),
            NeedleMap::SortedFile(nm) => nm.save_to_idx(path),
        }
    }

    /// Visit all entries in ascending order by needle ID.
    pub fn ascending_visit<F, E>(&self, f: F) -> Result<(), E>
    where
        F: FnMut(NeedleId, &NeedleValue) -> Result<(), E>,
        E: From<String>,
    {
        match self {
            NeedleMap::InMemory(nm) => nm.ascending_visit(f),
            NeedleMap::Redb(nm) => nm.ascending_visit(f),
            NeedleMap::SortedFile(nm) => nm.ascending_visit(f),
        }
    }

    /// Iterate all entries. Returns a Vec of (NeedleId, NeedleValue) pairs.
    /// For InMemory this collects via ascending visit; the disk-backed maps read
    /// it back off disk, so a truncated .sdx or a redb read fault surfaces here
    /// as an error. Compaction treats the result as the complete live set, so a
    /// partial scan must never be mistaken for an empty tail.
    pub fn iter_entries(&self) -> io::Result<Vec<(NeedleId, NeedleValue)>> {
        match self {
            NeedleMap::InMemory(nm) => {
                let mut entries = Vec::new();
                // The visitor never fails, so neither can this.
                let _ = nm.ascending_visit(|id, nv| {
                    entries.push((id, *nv));
                    Ok::<(), std::convert::Infallible>(())
                });
                Ok(entries)
            }
            NeedleMap::Redb(nm) => nm.collect_entries(),
            NeedleMap::SortedFile(nm) => nm.iter_entries(),
        }
    }
}

// ============================================================================
// Tests
// ============================================================================

#[cfg(test)]
pub(crate) mod test_support {
    use super::*;

    /// The `.idx` size in the map's META table, read through the live
    /// handle.
    ///
    /// The load path records the `.idx` size with `Durability::None`, and
    /// every `put`/`delete` also commits non-durably, so before the first
    /// checkpoint this is the load-time value (`Some(0)` for a fresh map) —
    /// NOT the crash-durable `None` a copy of the open `.rdb` would show.
    /// A live read is the only portable observation: redb 4.2.0 takes an
    /// exclusive whole-file lock, which is advisory on Unix but mandatory
    /// on Windows, so copying the open `.rdb` fails there with OS error 33.
    ///
    /// It still pins the property under test: the only *durable* META
    /// writer is `checkpoint`, so any value other than the load-time one
    /// proves a checkpoint recorded progress — and the post-checkpoint
    /// value equals the durable one, because checkpoints commit with
    /// `Durability::Immediate`. What is lost vs the old copy: strict crash
    /// fidelity — a hard crash pre-checkpoint would leave META absent
    /// rather than `Some(0)` (loader-equivalent outcomes: full rebuild vs
    /// replay-from-0, both correct). A clean close+reopen cannot recover
    /// that distinction either: dropping the `Database` flushes pending
    /// non-durable commits, so a reopened handle reads `Some(0)` just like
    /// the live one.
    pub(crate) fn live_meta_idx_size(nm: &RedbNeedleMap) -> Option<u64> {
        nm.live_meta_idx_size()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::io::Cursor;

    #[test]
    fn test_needle_map_put_get() {
        let mut nm = CompactNeedleMap::new();
        nm.put(NeedleId(1), Offset::from_actual_offset(0), Size(100))
            .unwrap();
        nm.put(NeedleId(2), Offset::from_actual_offset(128), Size(200))
            .unwrap();

        let v1 = nm.get(NeedleId(1)).unwrap();
        assert_eq!(v1.size, Size(100));

        let v2 = nm.get(NeedleId(2)).unwrap();
        assert_eq!(v2.size, Size(200));

        assert!(nm.get(NeedleId(99)).is_none());
    }

    #[test]
    fn test_needle_map_delete() {
        let mut nm = CompactNeedleMap::new();
        nm.put(NeedleId(1), Offset::from_actual_offset(0), Size(100))
            .unwrap();

        assert_eq!(nm.file_count(), 1);
        assert_eq!(nm.content_size(), 100);

        let deleted = nm
            .delete(NeedleId(1), Offset::from_actual_offset(0))
            .unwrap();
        assert_eq!(deleted, Some(Size(100)));

        // Additive-only: file_count stays at 1 after delete
        assert_eq!(nm.file_count(), 1);
        assert_eq!(nm.deleted_count(), 1);
        assert_eq!(nm.deleted_size(), 100);
    }

    #[test]
    fn test_needle_map_metrics() {
        let mut nm = CompactNeedleMap::new();
        nm.put(NeedleId(1), Offset::from_actual_offset(0), Size(100))
            .unwrap();
        nm.put(NeedleId(2), Offset::from_actual_offset(128), Size(200))
            .unwrap();
        nm.put(NeedleId(3), Offset::from_actual_offset(384), Size(300))
            .unwrap();

        assert_eq!(nm.file_count(), 3);
        assert_eq!(nm.content_size(), 600);
        assert_eq!(nm.max_file_key(), NeedleId(3));

        // Update existing — additive-only: file_count increments, content_size adds
        nm.put(NeedleId(2), Offset::from_actual_offset(700), Size(250))
            .unwrap();
        assert_eq!(nm.file_count(), 4); // 3 + 1 (always increments)
        assert_eq!(nm.content_size(), 850); // 600 + 250 (always adds)

        // Delete — additive-only: file_count unchanged
        nm.delete(NeedleId(1), Offset::from_actual_offset(0))
            .unwrap();
        assert_eq!(nm.file_count(), 4); // unchanged
        assert_eq!(nm.deleted_count(), 2); // 1 from overwrite + 1 from delete
    }

    #[test]
    fn test_needle_map_load_from_idx() {
        // Build an idx file in memory
        // Note: offset 0 is reserved for the SuperBlock, so real needles start at offset >= 8
        let mut idx_data = Vec::new();
        idx::write_index_entry(
            &mut idx_data,
            NeedleId(1),
            Offset::from_actual_offset(8),
            Size(100),
        )
        .unwrap();
        idx::write_index_entry(
            &mut idx_data,
            NeedleId(2),
            Offset::from_actual_offset(128),
            Size(200),
        )
        .unwrap();
        idx::write_index_entry(
            &mut idx_data,
            NeedleId(3),
            Offset::from_actual_offset(384),
            Size(300),
        )
        .unwrap();
        // Delete needle 2
        idx::write_index_entry(
            &mut idx_data,
            NeedleId(2),
            Offset::default(),
            TOMBSTONE_FILE_SIZE,
        )
        .unwrap();

        let mut cursor = Cursor::new(idx_data);
        let nm = CompactNeedleMap::load_from_idx(&mut cursor, Version::current()).unwrap();

        assert!(nm.get(NeedleId(1)).is_some());
        assert!(nm.get(NeedleId(2)).is_none()); // deleted
        assert!(nm.get(NeedleId(3)).is_some());
        // Additive-only: put(1)+put(2)+put(3) = 3, delete doesn't decrement
        assert_eq!(nm.file_count(), 3);
    }

    #[test]
    fn test_needle_map_double_delete() {
        let mut nm = CompactNeedleMap::new();
        nm.put(NeedleId(1), Offset::from_actual_offset(0), Size(100))
            .unwrap();

        let r1 = nm
            .delete(NeedleId(1), Offset::from_actual_offset(0))
            .unwrap();
        assert_eq!(r1, Some(Size(100)));

        // Second delete should return None (already deleted)
        let r2 = nm
            .delete(NeedleId(1), Offset::from_actual_offset(0))
            .unwrap();
        assert_eq!(r2, None);
        assert_eq!(nm.deleted_count(), 1); // not double counted
    }

    // ---- RedbNeedleMap tests ----

    /// The cache budget production gives the smallest redb tier.
    fn redb_test_cache() -> usize {
        NeedleMapKind::Redb.redb_cache_bytes()
    }

    /// Open a redb map on an empty .idx with an append writer attached, the
    /// way `Volume::load_index_redb` opens a writable volume.
    fn open_writable_redb(
        dir: &std::path::Path,
    ) -> (RedbNeedleMap, std::path::PathBuf, std::path::PathBuf) {
        let db_path = dir.join("v.rdb");
        let idx_path = dir.join("v.idx");
        let idx_file = std::fs::OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .truncate(false)
            .open(&idx_path)
            .unwrap();
        let idx_size = idx_file.metadata().unwrap().len();
        let mut reader = std::io::BufReader::new(&idx_file);
        let mut nm = RedbNeedleMap::load_from_idx(
            db_path.to_str().unwrap(),
            &mut reader,
            Version::current(),
            redb_test_cache(),
        )
        .unwrap();
        let writer = std::fs::OpenOptions::new()
            .write(true)
            .open(&idx_path)
            .unwrap();
        nm.set_idx_file(Box::new(writer), idx_size);
        (nm, db_path, idx_path)
    }

    /// .idx records in non-id order: 10, 1, 5, overwrite 1, delete 5.
    fn shuffled_idx_with_overwrite_and_delete() -> Vec<u8> {
        let mut idx_data = Vec::new();
        idx::write_index_entry(
            &mut idx_data,
            NeedleId(10),
            Offset::from_actual_offset(8),
            Size(100),
        )
        .unwrap();
        idx::write_index_entry(
            &mut idx_data,
            NeedleId(1),
            Offset::from_actual_offset(128),
            Size(50),
        )
        .unwrap();
        idx::write_index_entry(
            &mut idx_data,
            NeedleId(5),
            Offset::from_actual_offset(384),
            Size(75),
        )
        .unwrap();
        idx::write_index_entry(
            &mut idx_data,
            NeedleId(1),
            Offset::from_actual_offset(200),
            Size(200),
        )
        .unwrap();
        idx::write_index_entry(
            &mut idx_data,
            NeedleId(5),
            Offset::default(),
            TOMBSTONE_FILE_SIZE,
        )
        .unwrap();
        idx_data
    }

    #[test]
    fn test_redb_full_rebuild_last_write_wins_from_shuffled_idx() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.rdb");
        let idx_data = shuffled_idx_with_overwrite_and_delete();
        let mut cursor = Cursor::new(idx_data);
        let nm = RedbNeedleMap::load_from_idx(
            db_path.to_str().unwrap(),
            &mut cursor,
            Version::current(),
            redb_test_cache(),
        )
        .unwrap();

        let v10 = nm.get(NeedleId(10)).unwrap().unwrap();
        assert_eq!(v10.size, Size(100));
        let v1 = nm.get(NeedleId(1)).unwrap().unwrap();
        assert_eq!(v1.size, Size(200));
        assert_eq!(v1.offset, Offset::from_actual_offset(200));
        assert!(nm.get(NeedleId(5)).unwrap().is_none());
    }

    #[test]
    fn test_redb_full_rebuild_drops_keys_absent_from_idx() {
        let dir = tempfile::tempdir().unwrap();
        let (mut nm, db_path, idx_path) = open_writable_redb(dir.path());
        nm.put(NeedleId(1), Offset::from_actual_offset(8), Size(100))
            .unwrap();
        nm.put(NeedleId(99), Offset::from_actual_offset(128), Size(50))
            .unwrap();
        nm.checkpoint(true).unwrap();
        nm.close();
        drop(nm);

        // idx on disk is two entries; stored idx_size is two entries.
        // Shrink the .idx so try_reuse_rdb rejects (stored > idx) and
        // full_rebuild runs. Key 99 must not survive.
        {
            let f = std::fs::OpenOptions::new()
                .write(true)
                .open(&idx_path)
                .unwrap();
            f.set_len(NEEDLE_MAP_ENTRY_SIZE as u64).unwrap();
        }

        let mut idx = std::fs::File::open(&idx_path).unwrap();
        let reloaded = RedbNeedleMap::load_from_idx(
            db_path.to_str().unwrap(),
            &mut idx,
            Version::current(),
            redb_test_cache(),
        )
        .unwrap();
        assert!(reloaded.get(NeedleId(1)).unwrap().is_some());
        assert!(
            reloaded.get(NeedleId(99)).unwrap().is_none(),
            "full_rebuild must not keep keys that are not in the .idx"
        );
    }

    #[cfg(feature = "redb-experimental-cursor")]
    #[test]
    fn test_redb_full_rebuild_insert_before_reads_back_thousands_of_shuffled_keys() {
        // Enough keys to split leaves. insert_before vs table.insert only
        // diverges across page boundaries (pending-insert buffer / close).
        const N: u64 = 4000;
        let mut idx_data = Vec::new();
        for i in (1..=N).rev() {
            idx::write_index_entry(
                &mut idx_data,
                NeedleId(i),
                Offset::from_actual_offset((8 * i) as i64),
                Size(i as i32),
            )
            .unwrap();
        }
        idx::write_index_entry(
            &mut idx_data,
            NeedleId(1),
            Offset::from_actual_offset(200),
            Size(200),
        )
        .unwrap();
        idx::write_index_entry(
            &mut idx_data,
            NeedleId(2),
            Offset::default(),
            TOMBSTONE_FILE_SIZE,
        )
        .unwrap();

        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.rdb");
        let mut cursor = Cursor::new(idx_data);
        let nm = RedbNeedleMap::load_from_idx(
            db_path.to_str().unwrap(),
            &mut cursor,
            Version::current(),
            redb_test_cache(),
        )
        .unwrap();

        let v1 = nm.get(NeedleId(1)).unwrap().unwrap();
        assert_eq!(v1.size, Size(200));
        assert_eq!(v1.offset, Offset::from_actual_offset(200));
        assert!(nm.get(NeedleId(2)).unwrap().is_none());
        let v_n = nm.get(NeedleId(N)).unwrap().unwrap();
        assert_eq!(v_n.size, Size(N as i32));
        let v3 = nm.get(NeedleId(3)).unwrap().unwrap();
        assert_eq!(v3.size, Size(3));
        assert_eq!(v3.offset, Offset::from_actual_offset(24));

        let mut live = 0u64;
        nm.ascending_visit(|_, _| {
            live += 1;
            Ok::<(), String>(())
        })
        .unwrap();
        assert_eq!(live, N - 1);
    }

    #[test]
    fn test_redb_needle_map_put_get() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.rdb");
        let mut nm = RedbNeedleMap::new(db_path.to_str().unwrap(), redb_test_cache()).unwrap();

        nm.put(NeedleId(1), Offset::from_actual_offset(0), Size(100))
            .unwrap();
        nm.put(NeedleId(2), Offset::from_actual_offset(128), Size(200))
            .unwrap();

        let v1 = nm.get(NeedleId(1)).unwrap().unwrap();
        assert_eq!(v1.size, Size(100));

        let v2 = nm.get(NeedleId(2)).unwrap().unwrap();
        assert_eq!(v2.size, Size(200));

        assert!(nm.get(NeedleId(99)).unwrap().is_none());
    }

    #[test]
    fn test_redb_put_overwrite_metrics_without_prior_get() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.rdb");
        let mut nm = RedbNeedleMap::new(db_path.to_str().unwrap(), redb_test_cache()).unwrap();
        nm.put(NeedleId(1), Offset::from_actual_offset(8), Size(100))
            .unwrap();
        nm.put(NeedleId(1), Offset::from_actual_offset(200), Size(200))
            .unwrap();
        assert_eq!(nm.file_count(), 2);
        assert_eq!(nm.content_size(), 300);
        assert_eq!(nm.deleted_count(), 1);
        assert_eq!(nm.deleted_size(), 100);
        let v = nm.get(NeedleId(1)).unwrap().unwrap();
        assert_eq!(v.size, Size(200));
        assert_eq!(v.offset, Offset::from_actual_offset(200));
    }

    #[test]
    fn test_redb_needle_map_delete() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.rdb");
        let mut nm = RedbNeedleMap::new(db_path.to_str().unwrap(), redb_test_cache()).unwrap();

        nm.put(NeedleId(1), Offset::from_actual_offset(0), Size(100))
            .unwrap();
        assert_eq!(nm.file_count(), 1);
        assert_eq!(nm.content_size(), 100);

        let deleted = nm
            .delete(NeedleId(1), Offset::from_actual_offset(0))
            .unwrap();
        assert_eq!(deleted, Some(Size(100)));

        // Additive-only: file_count stays at 1 after delete
        assert_eq!(nm.file_count(), 1);
        assert_eq!(nm.deleted_count(), 1);
        assert_eq!(nm.deleted_size(), 100);

        // Deleted entry should have negated size
        let nv = nm.get(NeedleId(1)).unwrap().unwrap();
        assert_eq!(nv.size, Size(-100));
    }

    #[test]
    fn test_redb_needle_map_metrics() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.rdb");
        let mut nm = RedbNeedleMap::new(db_path.to_str().unwrap(), redb_test_cache()).unwrap();

        nm.put(NeedleId(1), Offset::from_actual_offset(0), Size(100))
            .unwrap();
        nm.put(NeedleId(2), Offset::from_actual_offset(128), Size(200))
            .unwrap();
        nm.put(NeedleId(3), Offset::from_actual_offset(384), Size(300))
            .unwrap();

        assert_eq!(nm.file_count(), 3);
        assert_eq!(nm.content_size(), 600);
        assert_eq!(nm.max_file_key(), NeedleId(3));

        // Update existing — additive-only: file_count increments, content_size adds
        nm.put(NeedleId(2), Offset::from_actual_offset(700), Size(250))
            .unwrap();
        assert_eq!(nm.file_count(), 4); // 3 + 1 (always increments)
        assert_eq!(nm.content_size(), 850); // 600 + 250 (always adds)

        // Delete — additive-only: file_count unchanged
        nm.delete(NeedleId(1), Offset::from_actual_offset(0))
            .unwrap();
        assert_eq!(nm.file_count(), 4); // unchanged
        assert_eq!(nm.deleted_count(), 2); // 1 from overwrite + 1 from delete
    }

    #[test]
    fn test_redb_needle_map_load_from_idx() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.rdb");

        let mut idx_data = Vec::new();
        idx::write_index_entry(
            &mut idx_data,
            NeedleId(1),
            Offset::from_actual_offset(8),
            Size(100),
        )
        .unwrap();
        idx::write_index_entry(
            &mut idx_data,
            NeedleId(2),
            Offset::from_actual_offset(128),
            Size(200),
        )
        .unwrap();
        idx::write_index_entry(
            &mut idx_data,
            NeedleId(3),
            Offset::from_actual_offset(384),
            Size(300),
        )
        .unwrap();
        // Delete needle 2
        idx::write_index_entry(
            &mut idx_data,
            NeedleId(2),
            Offset::default(),
            TOMBSTONE_FILE_SIZE,
        )
        .unwrap();

        let mut cursor = Cursor::new(idx_data);
        let nm = RedbNeedleMap::load_from_idx(
            db_path.to_str().unwrap(),
            &mut cursor,
            Version::current(),
            redb_test_cache(),
        )
        .unwrap();

        assert!(nm.get(NeedleId(1)).unwrap().is_some());
        assert!(nm.get(NeedleId(2)).unwrap().is_none()); // deleted and removed
        assert!(nm.get(NeedleId(3)).unwrap().is_some());
        // Same history as test_needle_map_load_from_idx: the counters must
        // match what the in-memory map (and a live volume) accumulates.
        assert_eq!(nm.file_count(), 3);
        assert_eq!(nm.deleted_count(), 1);
        assert_eq!(nm.deleted_size(), 200);
    }

    #[test]
    fn test_redb_needle_map_double_delete() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.rdb");
        let mut nm = RedbNeedleMap::new(db_path.to_str().unwrap(), redb_test_cache()).unwrap();

        nm.put(NeedleId(1), Offset::from_actual_offset(0), Size(100))
            .unwrap();

        let r1 = nm
            .delete(NeedleId(1), Offset::from_actual_offset(0))
            .unwrap();
        assert_eq!(r1, Some(Size(100)));

        // Second delete should return None (already deleted)
        let r2 = nm
            .delete(NeedleId(1), Offset::from_actual_offset(0))
            .unwrap();
        assert_eq!(r2, None);
        assert_eq!(nm.deleted_count(), 1); // not double counted
    }

    #[test]
    fn test_redb_delete_wrong_length_value_is_absent_not_panic() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.rdb");
        let mut nm = RedbNeedleMap::new(db_path.to_str().unwrap(), redb_test_cache()).unwrap();
        {
            let txn = RedbNeedleMap::begin_write_no_fsync(nm.db.as_ref().unwrap()).unwrap();
            {
                let mut table = txn.open_table(NEEDLE_TABLE).unwrap();
                table.insert(1u64, &b"xx"[..]).unwrap();
            }
            txn.commit().unwrap();
        }
        let deleted = nm
            .delete(NeedleId(1), Offset::from_actual_offset(0))
            .unwrap();
        assert_eq!(deleted, None);
        assert_eq!(nm.deleted_count(), 0);
    }

    #[test]
    fn test_redb_needle_map_ascending_visit() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.rdb");
        let mut nm = RedbNeedleMap::new(db_path.to_str().unwrap(), redb_test_cache()).unwrap();

        nm.put(NeedleId(3), Offset::from_actual_offset(384), Size(300))
            .unwrap();
        nm.put(NeedleId(1), Offset::from_actual_offset(0), Size(100))
            .unwrap();
        nm.put(NeedleId(2), Offset::from_actual_offset(128), Size(200))
            .unwrap();

        let mut visited = Vec::new();
        nm.ascending_visit(|id, nv| {
            visited.push((id, nv.size));
            Ok::<(), String>(())
        })
        .unwrap();

        assert_eq!(visited.len(), 3);
        assert_eq!(visited[0], (NeedleId(1), Size(100)));
        assert_eq!(visited[1], (NeedleId(2), Size(200)));
        assert_eq!(visited[2], (NeedleId(3), Size(300)));
    }

    #[test]
    fn test_redb_needle_map_save_to_idx() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.rdb");
        let idx_path = dir.path().join("test.idx");

        let mut nm = RedbNeedleMap::new(db_path.to_str().unwrap(), redb_test_cache()).unwrap();
        nm.put(NeedleId(1), Offset::from_actual_offset(8), Size(100))
            .unwrap();
        nm.put(NeedleId(2), Offset::from_actual_offset(128), Size(200))
            .unwrap();
        nm.put(NeedleId(3), Offset::from_actual_offset(384), Size(300))
            .unwrap();
        // Delete needle 2
        nm.delete(NeedleId(2), Offset::from_actual_offset(128))
            .unwrap();

        nm.save_to_idx(idx_path.to_str().unwrap()).unwrap();

        // Load back with CompactNeedleMap to verify
        let mut idx_file = std::fs::File::open(&idx_path).unwrap();
        let loaded = CompactNeedleMap::load_from_idx(&mut idx_file, Version::current()).unwrap();
        assert_eq!(loaded.file_count(), 2); // only live entries
        assert!(loaded.get(NeedleId(1)).is_some());
        assert!(loaded.get(NeedleId(2)).is_none()); // deleted, not saved
        assert!(loaded.get(NeedleId(3)).is_some());
    }

    #[test]
    fn test_pack_unpack_needle_value() {
        let nv = NeedleValue {
            offset: Offset::from_actual_offset(8 * 1000),
            size: Size(4096),
        };
        let packed = pack_needle_value(&nv);
        let unpacked = unpack_needle_value(&packed);
        assert_eq!(
            nv.offset.to_actual_offset(),
            unpacked.offset.to_actual_offset()
        );
        assert_eq!(nv.size, unpacked.size);
    }

    #[test]
    fn test_pack_unpack_negative_size() {
        let nv = NeedleValue {
            offset: Offset::from_actual_offset(8 * 500),
            size: Size(-100),
        };
        let packed = pack_needle_value(&nv);
        let unpacked = unpack_needle_value(&packed);
        assert_eq!(
            nv.offset.to_actual_offset(),
            unpacked.offset.to_actual_offset()
        );
        assert_eq!(nv.size, unpacked.size);
    }

    // ---- NeedleMap enum tests ----

    #[test]
    fn test_needle_map_enum_inmemory() {
        let mut nm = NeedleMap::InMemory(CompactNeedleMap::new());
        nm.put(NeedleId(1), Offset::from_actual_offset(0), Size(100))
            .unwrap();
        assert_eq!(nm.get(NeedleId(1)).unwrap().unwrap().size, Size(100));
        assert_eq!(nm.file_count(), 1);
    }

    #[test]
    fn test_needle_map_enum_redb() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.rdb");
        let mut nm = NeedleMap::Redb(
            RedbNeedleMap::new(db_path.to_str().unwrap(), redb_test_cache()).unwrap(),
        );
        nm.put(NeedleId(1), Offset::from_actual_offset(0), Size(100))
            .unwrap();
        assert_eq!(nm.get(NeedleId(1)).unwrap().unwrap().size, Size(100));
        assert_eq!(nm.file_count(), 1);
    }

    #[test]
    fn test_needle_map_kind_redb_cache_bytes_grows_by_tier() {
        // Per-volume redb cache budgets. These must stay small: a volume
        // server opens one redb database per volume, so the process-wide
        // ceiling is roughly (volumes x budget).
        assert_eq!(NeedleMapKind::Redb.redb_cache_bytes(), 4 * 1024 * 1024);
        assert_eq!(
            NeedleMapKind::RedbMedium.redb_cache_bytes(),
            8 * 1024 * 1024
        );
        assert_eq!(
            NeedleMapKind::RedbLarge.redb_cache_bytes(),
            16 * 1024 * 1024
        );
    }

    #[test]
    fn test_redb_needle_map_tiny_cache_still_serves_all_entries() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.rdb");
        // A cache far smaller than the data forces redb to evict pages on
        // every write and read; every entry must still round-trip.
        let mut nm = RedbNeedleMap::new(db_path.to_str().unwrap(), 64 * 1024).unwrap();
        let n = 3000u64;
        for i in 1..=n {
            nm.put(
                NeedleId(i),
                Offset::from_actual_offset((i * 8) as i64),
                Size(i as i32),
            )
            .unwrap();
        }
        for i in 1..=n {
            let v = nm.get(NeedleId(i)).unwrap().unwrap();
            assert_eq!(v.size, Size(i as i32));
            assert_eq!(v.offset, Offset::from_actual_offset((i * 8) as i64));
        }
        assert_eq!(nm.file_count(), n as i64);
    }

    #[test]
    fn test_redb_checkpoint_is_explicit_and_due_every_interval() {
        use test_support::live_meta_idx_size;

        // Every non-durable redb commit leaves bookkeeping behind until a
        // durable one clears it, so a writable map asks for a checkpoint on
        // a fixed cadence. The map never takes it by itself: the volume has
        // to flush the .dat first, then call checkpoint().
        const EXPECTED_INTERVAL: u64 = 1000;
        let dir = tempfile::tempdir().unwrap();
        let (mut nm, db_path, _idx_path) = open_writable_redb(dir.path());
        for i in 1..EXPECTED_INTERVAL {
            nm.put(
                NeedleId(i),
                Offset::from_actual_offset((i * 8) as i64),
                Size(1),
            )
            .unwrap();
            assert!(!nm.checkpoint_due(), "due after only {i} writes");
        }
        nm.put(
            NeedleId(EXPECTED_INTERVAL),
            Offset::from_actual_offset((EXPECTED_INTERVAL * 8) as i64),
            Size(1),
        )
        .unwrap();
        assert!(nm.checkpoint_due());
        // No checkpoint taken yet: META still holds the load-time .idx size.
        // put() only commits non-durably, so the live value is unchanged.
        assert_eq!(
            live_meta_idx_size(&nm),
            Some(0),
            "put() must not record checkpoint progress"
        );

        nm.checkpoint(true).unwrap();
        assert!(!nm.checkpoint_due());
        assert_eq!(
            live_meta_idx_size(&nm),
            Some(EXPECTED_INTERVAL * NEEDLE_MAP_ENTRY_SIZE as u64),
            "checkpoint records how much of the .idx the table reflects"
        );

        // Everything is durable after the checkpoint, so the map is closed
        // first and the snapshot sees the same bytes on every platform.
        // (Copying while open fails on Windows, where redb's file lock is
        // mandatory: what a crash leaves.)
        drop(nm);
        let crash_copy = dir.path().join("crash.rdb");
        std::fs::copy(&db_path, &crash_copy).unwrap();
        let db = Database::open(&crash_copy).unwrap();
        let txn = db.begin_read().unwrap();
        let table = txn.open_table(NEEDLE_TABLE).unwrap();
        assert!(
            table.get(EXPECTED_INTERVAL).unwrap().is_some(),
            "entries up to the checkpoint are durable"
        );
    }

    #[test]
    fn test_redb_close_records_idx_size_so_reload_does_not_double_count() {
        let dir = tempfile::tempdir().unwrap();
        let (mut nm, db_path, idx_path) = open_writable_redb(dir.path());
        for i in 1..=5u64 {
            nm.put(
                NeedleId(i),
                Offset::from_actual_offset((i * 8) as i64),
                Size(1),
            )
            .unwrap();
        }
        nm.close();
        drop(nm);

        // A clean close leaves the table durable; the recorded .idx size
        // must match it, or the reload replays the same 5 entries on top.
        let mut idx = std::fs::File::open(&idx_path).unwrap();
        let reloaded = RedbNeedleMap::load_from_idx(
            db_path.to_str().unwrap(),
            &mut idx,
            Version::current(),
            redb_test_cache(),
        )
        .unwrap();
        assert_eq!(reloaded.file_count(), 5);
        assert_eq!(reloaded.deleted_count(), 0);
    }

    #[test]
    fn test_redb_reload_with_stale_idx_size_does_not_double_count() {
        let dir = tempfile::tempdir().unwrap();
        let (mut nm, db_path, idx_path) = open_writable_redb(dir.path());
        for i in 1..=5u64 {
            nm.put(
                NeedleId(i),
                Offset::from_actual_offset((i * 8) as i64),
                Size(1),
            )
            .unwrap();
        }
        // Drop without close(): redb makes the table durable on drop, but the
        // recorded .idx size stays at its load-time value (0), so the reload
        // replays all 5 entries over rows the table already holds.
        drop(nm);

        let mut idx = std::fs::File::open(&idx_path).unwrap();
        let reloaded = RedbNeedleMap::load_from_idx(
            db_path.to_str().unwrap(),
            &mut idx,
            Version::current(),
            redb_test_cache(),
        )
        .unwrap();
        assert_eq!(reloaded.file_count(), 5);
        assert_eq!(reloaded.deleted_count(), 0);
        assert_eq!(
            reloaded.get(NeedleId(5)).unwrap().unwrap().offset,
            Offset::from_actual_offset(40)
        );
    }

    #[test]
    fn test_redb_reload_metrics_keep_overwrite_and_delete_history() {
        // garbage_level() is deleted_size / content_size. Every load path
        // must rebuild both from the whole .idx history, the way the live
        // counters accumulate, not from the table's final state, or the
        // bytes of overwritten and deleted needles stop counting as garbage
        // after a restart.
        for (close_first, rebuild) in [(true, false), (false, false), (true, true)] {
            let dir = tempfile::tempdir().unwrap();
            let (mut nm, db_path, idx_path) = open_writable_redb(dir.path());
            nm.put(NeedleId(1), Offset::from_actual_offset(8), Size(100))
                .unwrap();
            // Overwrite: the first 100 bytes become garbage.
            nm.put(NeedleId(1), Offset::from_actual_offset(200), Size(200))
                .unwrap();
            nm.put(NeedleId(2), Offset::from_actual_offset(500), Size(50))
                .unwrap();
            nm.delete(NeedleId(2), Offset::from_actual_offset(600))
                .unwrap();
            let live = (
                nm.file_count(),
                nm.content_size(),
                nm.deleted_count(),
                nm.deleted_size(),
            );
            assert_eq!(live, (3, 350, 2, 150));
            if close_first {
                nm.close();
            }
            drop(nm);
            if rebuild {
                std::fs::remove_file(&db_path).unwrap();
            }

            let mut idx = std::fs::File::open(&idx_path).unwrap();
            let reloaded = RedbNeedleMap::load_from_idx(
                db_path.to_str().unwrap(),
                &mut idx,
                Version::current(),
                redb_test_cache(),
            )
            .unwrap();
            let after = (
                reloaded.file_count(),
                reloaded.content_size(),
                reloaded.deleted_count(),
                reloaded.deleted_size(),
            );
            assert_eq!(after, live, "close_first={close_first} rebuild={rebuild}");
            assert_eq!(reloaded.get(NeedleId(1)).unwrap().unwrap().size, Size(200));
            assert!(
                reloaded
                    .get(NeedleId(2))
                    .unwrap()
                    .is_none_or(|v| v.size.is_deleted())
            );
        }
    }
}