armdb 0.7.0

use std::fs;
use std::sync::Arc;
use std::time::Duration;

use zerocopy::FromBytes;

use crate::Key;
use crate::disk_loc::DiskLoc;
use crate::entry::{EntryHeader, entry_size};
use crate::error::DbResult;
use crate::hint;
use crate::io::direct;
use crate::shard::{ImmutableFile, Shard};

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

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

fn dir_fsync(dir: &std::path::Path) -> DbResult<()> {
    direct::sync_dir(dir)
}

pub trait CompactionIndex<K: Key>: Send + Sync {
    /// If the current index points to `old_loc`, it is updated to `new_loc` and returns true.
    fn update_if_match(&self, key: &K, old_loc: DiskLoc, new_loc: DiskLoc) -> bool;

    /// Invalidate cached blocks for a file after compaction replaces its contents.
    fn invalidate_blocks(&self, _shard_id: u8, _file_id: u32, _total_bytes: u64) {}

    /// Returns true if the key currently exists in the index (i.e. has a live Put).
    fn contains_key(&self, key: &K) -> bool;

    /// Returns true iff the index currently maps `key` to exactly `loc` — i.e.
    /// the entry at `loc` is the live (latest) version of `key`.
    ///
    /// `shard_id` is the shard being compacted. Every key scanned during a shard's
    /// compaction routes to that shard, so sharded (map) indexes use `shard_id`
    /// directly instead of recomputing `shard_for(key)` per entry; tree indexes
    /// are global and ignore it.
    ///
    /// Used by compaction to skip superseded entries before copying them into a
    /// new file. A superseded entry can never become live again (a newer version
    /// with a higher GSN exists), so a `false` result is permanently safe to act
    /// on even while writers run concurrently. A `true` result may still race with
    /// a concurrent overwrite; the post-copy [`update_if_match`](Self::update_if_match)
    /// CAS is the authoritative guard that re-marks the copy dead if it lost the race.
    fn is_live(&self, shard_id: u8, key: &K, loc: DiskLoc) -> bool;
}

/// Guard that prevents compaction from removing files with unreplicated entries.
#[cfg(feature = "replication")]
pub trait CompactionGuard: Send + Sync {
    /// Minimum GSN replicated to all followers for this shard.
    /// Files containing entries with GSN above this value must not be compacted.
    ///
    /// # Watermark policy
    ///
    /// The value starts at `0` and advances monotonically (`fetch_max`) from a
    /// connected follower's acks; it **never decreases** and carries no follower
    /// identity. This yields two deliberate extremes (see `docs/design.md`,
    /// "Политика `min_replicated_gsn` watermark"):
    ///
    /// - **No follower ever acked → `0`**: every file has `max_gsn >= 0`, so
    ///   `compact_shard_guarded` is a no-op until the first ack. Guarded
    ///   compaction intentionally blocks (dead bytes grow) rather than risk
    ///   dropping un-replicated data. Use `compact_shard` / [`NoReplicationGuard`]
    ///   (which returns `u64::MAX`) when no follower is expected.
    /// - **Replacement/fresh follower**: the watermark still reflects the *old*
    ///   follower's progress. This is safe because catch-up emits a GSN-sorted
    ///   stream (k-way merge in `ShardLogReader`), so the client's stale-GSN
    ///   watermark stays correct even over files compacted before the fresh
    ///   follower connected.
    ///
    /// Follower-identity reset and a no-ack compaction timeout are policy knobs
    /// left unimplemented until a concrete deployment needs them.
    fn min_replicated_gsn(&self, shard_id: u8) -> u64;
}

/// No-op guard — allows compaction of all files.
#[cfg(feature = "replication")]
pub struct NoReplicationGuard;

#[cfg(feature = "replication")]
impl CompactionGuard for NoReplicationGuard {
    fn min_replicated_gsn(&self, _shard_id: u8) -> u64 {
        u64::MAX
    }
}

/// Returns true when the next write of `needed` bytes at `write_offset` would
/// exceed `max_file_size`. Extracted as a pure function for unit-testability.
fn should_rotate_output(write_offset: u64, needed: u64, max_file_size: u64) -> bool {
    write_offset + needed > max_file_size
}

pub fn compact_shard<K: Key, I: CompactionIndex<K>>(
    shard: &Shard,
    index: &I,
    threshold: f64,
) -> DbResult<usize> {
    compact_shard_inner::<K, I>(shard, index, threshold, u64::MAX)
}

/// Compact with replication guard — skip files whose max GSN >= min_replicated_gsn.
#[cfg(feature = "replication")]
pub fn compact_shard_guarded<K: Key, I: CompactionIndex<K>>(
    shard: &Shard,
    index: &I,
    threshold: f64,
    guard: &dyn CompactionGuard,
) -> DbResult<usize> {
    let min_gsn = guard.min_replicated_gsn(shard.id);
    compact_shard_inner::<K, I>(shard, index, threshold, min_gsn)
}

/// State for the currently open compaction output file.
struct OutputState {
    file_id: u32,
    tmp_file: fs::File,
    write_offset: u64,
    #[cfg(feature = "encryption")]
    enc_state: Option<EncryptionState>,
}

#[cfg(feature = "encryption")]
struct EncryptionState {
    cipher: Arc<PageCipher>,
    page_buf: Box<[u8; 4096]>,
    page_offset: usize,
    pages_written: u64,
    tag_list: Vec<[u8; 16]>,
}

#[cfg(feature = "encryption")]
impl EncryptionState {
    fn flush_page(&mut self, file: &fs::File, file_id: u32) -> DbResult<()> {
        let tag = self
            .cipher
            .encrypt_page(file_id, self.pages_written, &mut *self.page_buf)?;
        direct::pwrite_at(file, &*self.page_buf, self.pages_written * 4096)?;
        self.tag_list.push(tag);
        self.pages_written += 1;
        self.page_buf.fill(0);
        self.page_offset = 0;
        Ok(())
    }

    fn write_bytes(&mut self, file: &fs::File, file_id: u32, data: &[u8]) -> DbResult<()> {
        let mut remaining = data;
        while !remaining.is_empty() {
            let space = 4096 - self.page_offset;
            let chunk = remaining.len().min(space);
            self.page_buf[self.page_offset..self.page_offset + chunk]
                .copy_from_slice(&remaining[..chunk]);
            self.page_offset += chunk;
            remaining = &remaining[chunk..];
            if self.page_offset == 4096 {
                self.flush_page(file, file_id)?;
            }
        }
        Ok(())
    }
}

/// A finalized compaction output ready to be registered as immutable.
struct FinalizedOutput {
    file_id: u32,
    total_bytes: u64,
    final_data_path: std::path::PathBuf,
    #[cfg(feature = "encryption")]
    final_tag_path: Option<std::path::PathBuf>,
}

/// RAII cleanup for compaction outputs (C-02).
///
/// A compaction pass writes one or more output files. Each is first written to
/// `{id}.data.tmp` (`+ {id}.tags.tmp` when encrypted), then `finalize_output`
/// renames it to the final `{id}.data` (`+ {id}.tags`). A finalized output is
/// **not** live until it is registered into `inner.immutable`; between finalize
/// and registration it is an orphan on disk.
///
/// While *armed*, dropping this guard removes:
/// * every finalized-but-unregistered output's final `.data` (and `.tags`), and
/// * the current in-progress output's `.data.tmp` (and `.tags.tmp`).
///
/// It is [`disarm`](Self::disarm)ed the moment the outputs are registered as
/// immutable — the point of no return — so a later error (hint write, flush,
/// old-file deletion) never deletes a file that is already serving reads. Every
/// early `?`-exit and the C-01 anomalous-break aborts share this one path,
/// leaving all *source* files untouched.
struct OutputGuard {
    dir: std::path::PathBuf,
    /// File ids whose final `.data` exists but is not yet registered.
    finalized_ids: Vec<u32>,
    /// The in-progress output whose `.data.tmp` is still open, if any.
    current_tmp_id: Option<u32>,
    armed: bool,
}

impl OutputGuard {
    fn new(dir: std::path::PathBuf) -> Self {
        Self {
            dir,
            finalized_ids: Vec::new(),
            current_tmp_id: None,
            armed: true,
        }
    }

    /// Record the id of the output currently being written to `{id}.data.tmp`.
    fn set_current(&mut self, file_id: u32) {
        self.current_tmp_id = Some(file_id);
    }

    /// Record that `file_id` was finalized (tmp renamed to the final `.data`)
    /// but is not yet registered; the previous "current" tmp is gone.
    fn mark_finalized(&mut self, file_id: u32) {
        self.current_tmp_id = None;
        self.finalized_ids.push(file_id);
    }

    /// The current in-progress tmp has already been removed by the caller
    /// (empty-output path); stop tracking it.
    fn clear_current(&mut self) {
        self.current_tmp_id = None;
    }

    /// Stop tracking — the outputs are now registered and must not be removed.
    fn disarm(&mut self) {
        self.armed = false;
    }
}

impl Drop for OutputGuard {
    fn drop(&mut self) {
        if !self.armed {
            return;
        }

        // A missing file is expected here (it may never have been created); any
        // *other* unlink error is surfaced rather than silently swallowed, so a
        // cleanup that leaves an orphan behind is at least observable in logs.
        fn remove_logged(path: std::path::PathBuf) {
            match fs::remove_file(&path) {
                Ok(()) => {}
                Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
                Err(e) => {
                    tracing::warn!(?path, err = %e, "compaction cleanup failed to remove orphan")
                }
            }
        }

        // P1-7: remove each finalized output's `.data`, then (encrypted only)
        // fsync the directory *before* removing its `.tags`. A `.data` without a
        // sibling `.tags` is a poison pill that makes recovery hard-fail, so the
        // `.data` removal must be committed before the `.tags` removal — otherwise
        // a crash on a metadata-reordering filesystem could persist the `.tags`
        // unlink while losing the `.data` unlink, resurfacing exactly that shape.
        for fid in &self.finalized_ids {
            remove_logged(self.dir.join(format!("{fid:06}.data")));
        }
        #[cfg(feature = "encryption")]
        {
            let _ = dir_fsync(&self.dir);
            for fid in &self.finalized_ids {
                remove_logged(self.dir.join(format!("{fid:06}.tags")));
            }
        }
        if let Some(id) = self.current_tmp_id {
            remove_logged(self.dir.join(format!("{id:06}.data.tmp")));
            #[cfg(feature = "encryption")]
            remove_logged(self.dir.join(format!("{id:06}.tags.tmp")));
        }
        #[cfg(feature = "encryption")]
        let _ = dir_fsync(&self.dir);
    }
}

// ---------------------------------------------------------------------------
// Test-only fault-injection seams (compiled out of production builds).
//
// These live in `src/` so the in-file unit tests can drive the exact internal
// crash windows the C-01/C-02/C-03 fixes protect against. `#[cfg(test)]` keeps
// them out of the normal `cdylib`/integration builds entirely.
// ---------------------------------------------------------------------------
#[cfg(test)]
thread_local! {
    /// C-03: when set, `finalize_output` returns an error *between* the two
    /// encrypted renames, modelling a crash there. With the [`OutputGuard`]
    /// this leaves exactly the on-disk state a real crash would, so a reopen
    /// exercises the rename ordering.
    static CRASH_BETWEEN_RENAMES: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
    /// C-02: `open_output` call counter (1-based).
    static OPEN_OUTPUT_CALLS: std::cell::Cell<u32> = const { std::cell::Cell::new(0) };
    /// C-02: the `open_output` call index at which to inject a failure
    /// (`0` = never). Failing on call 2 models a fault right after the first
    /// output was finalized during a rotation.
    static FAIL_OPEN_OUTPUT_AT: std::cell::Cell<u32> = const { std::cell::Cell::new(0) };
}

fn compact_shard_inner<K: Key, I: CompactionIndex<K>>(
    shard: &Shard,
    index: &I,
    threshold: f64,
    min_replicated_gsn: u64,
) -> DbResult<usize> {
    let mut files_to_compact = Vec::new();

    // 1. Find immutable files exceeding the garbage ratio threshold
    #[cfg(feature = "encryption")]
    let cipher_opt: Option<Arc<PageCipher>>;
    let max_file_size: u64;
    let max_files_per_pass: usize;
    let cooldown_ids: Vec<u32>;
    {
        let mut inner = shard.lock();
        max_file_size = inner.max_file_size;
        max_files_per_pass = inner.compaction_max_files_per_pass;
        cooldown_ids = std::mem::take(&mut inner.last_compaction_output_ids);
        #[cfg(feature = "encryption")]
        {
            cipher_opt = inner.cipher.clone();
        }
        for file in &inner.immutable {
            let total = file.total_bytes;
            let dead = inner.dead_bytes.get(&file.file_id).copied().unwrap_or(0);
            if total > 0
                && (dead as f64 / total as f64) > threshold
                && !cooldown_ids.contains(&file.file_id)
            {
                files_to_compact.push(file.clone());
            }
        }
    }

    if files_to_compact.is_empty() {
        return Ok(0);
    }

    // Filter out files that haven't been fully replicated to all followers
    if min_replicated_gsn < u64::MAX {
        files_to_compact.retain(|file| {
            let hint_path = shard.dir().join(format!("{:06}.hint", file.file_id));
            match file_max_gsn(&hint_path, file.file_id, size_of::<K>()) {
                Some(max_gsn) => max_gsn < min_replicated_gsn,
                None => false, // No hint → conservative, skip
            }
        });
        if files_to_compact.is_empty() {
            return Ok(0);
        }
    }

    files_to_compact.sort_by_key(|f| f.file_id);
    // Bound per-pass work to `max_files_per_pass` files (config-tunable). Sorting
    // by file_id first means we retire the oldest, most-superseded files. `0` =
    // unlimited: drain the entire backlog in one pass — needed when a continuous
    // writer would otherwise create files faster than a capped pass can reclaim
    // them (the file creation rate outpacing removal makes on-disk size track
    // total bytes written rather than the live set).
    if max_files_per_pass > 0 {
        files_to_compact.truncate(max_files_per_pass);
    }
    let compact_start = std::time::Instant::now();

    let old_file_ids: Vec<u32> = files_to_compact.iter().map(|f| f.file_id).collect();

    // Allocate the first output file_id under the lock.
    let first_file_id = {
        let mut inner = shard.lock();
        inner.allocate_file_id()?
    };

    let mut output = open_output(shard, first_file_id)?;
    // RAII cleanup for finalized-but-unregistered outputs and the in-progress
    // tmp (C-02). Disarmed once the outputs are registered as immutable.
    let mut cleanup = OutputGuard::new(shard.dir().to_path_buf());
    cleanup.set_current(first_file_id);
    let serving_direct = shard.effective_direct_io();

    // Entries accumulated for deferred index update and dead-byte accounting.
    const BATCH_SIZE: usize = 256;

    struct BatchEntry<K> {
        key: K,
        gsn: u64,
        crc32: u32,
        old_loc: DiskLoc,
        new_loc: DiskLoc,
        is_tombstone: bool,
    }

    let mut batch: Vec<BatchEntry<K>> = Vec::with_capacity(BATCH_SIZE);
    // Outputs that have been written and fsynced but not yet registered as immutable.
    let mut pending_outputs: Vec<FinalizedOutput> = Vec::new();

    // Scan old files (bulk read fds; serving fds stay for live reads until removal).
    for old_arc in &files_to_compact {
        let data_path = shard.dir().join(format!("{:06}.data", old_arc.file_id));
        let scan_file = direct::open_bulk_read(&data_path)?;
        let file = &scan_file;
        let file_len = old_arc.total_bytes;
        let mut offset: u64 = 0;

        // Build reader for this file (encrypted or plain)
        #[cfg(feature = "encryption")]
        let read_fn: Box<ReadFn> =
            if let (Some(cipher), Some(_tag_file)) = (&cipher_opt, &old_arc.tag_file) {
                let c = cipher.clone();
                let fid = old_arc.file_id;
                let tp = tags::tags_path_for_data(&old_arc.path);
                let tf = Arc::new(TagFile::open_read(&tp)?);
                Box::new(move |f, o, l| direct::pread_value_encrypted(f, &tf, &c, fid, o, l))
            } else {
                Box::new(direct::pread_value)
            };
        #[cfg(not(feature = "encryption"))]
        let read_fn: Box<ReadFn> = Box::new(direct::pread_value);

        while offset + size_of::<EntryHeader>() as u64 <= file_len {
            // C-01: a header we cannot read or parse mid-file is corruption, not
            // a clean end. Aborting (and preserving every source via the
            // OutputGuard) is the only safe response — silently `break`ing here
            // would retire this file and permanently drop every live entry that
            // still follows the damaged region. Encrypted files reach this on a
            // single flipped byte (the page GCM tag then fails to authenticate).
            let header_bytes = match read_fn(file, offset, size_of::<EntryHeader>()) {
                Ok(b) => b,
                Err(_) => return Err(crate::error::DbError::CorruptedEntry { offset }),
            };
            let header = match EntryHeader::read_from_bytes(&header_bytes) {
                Ok(h) => h,
                Err(_) => return Err(crate::error::DbError::CorruptedEntry { offset }),
            };

            // Skip clean zero page padding (plain or encrypted); only stop the
            // file scan when the padding reaches EOF (a *clean* end).
            if header.gsn == 0 && header.crc32 == 0 && header.value_len == 0 {
                if crate::entry::check_page_padding_at(
                    file,
                    offset + size_of::<EntryHeader>() as u64,
                    read_fn.as_ref(),
                )? {
                    const PAGE_SIZE: u64 = 4096;
                    let next_page =
                        (offset + size_of::<EntryHeader>() as u64).div_ceil(PAGE_SIZE) * PAGE_SIZE;
                    if next_page >= file_len {
                        break; // clean end: zero padding runs to EOF
                    }
                    offset = next_page;
                    continue;
                }
                // C-01: a zero header whose page is *not* clean padding is
                // corruption. Abort rather than retire the file and drop the
                // live entries beyond it.
                return Err(crate::error::DbError::CorruptedEntry { offset });
            }

            let total = entry_size(size_of::<K>(), header.value_len);
            if offset + total > file_len {
                // C-01: the entry claims to extend past EOF — a torn write or a
                // corrupt `value_len` (this path runs *before* the CRC check, so
                // it would otherwise bypass the V-10 guard). An immutable file is
                // fully fsynced before it is sealed, so this can only be
                // corruption: abort and preserve the source.
                return Err(crate::error::DbError::CorruptedEntry { offset });
            }

            let old_loc = DiskLoc::new(
                old_arc.file_id,
                (offset + size_of::<EntryHeader>() as u64 + size_of::<K>() as u64) as u32,
                header.value_len,
            );

            let entry_bytes = read_fn(file, offset, total as usize)?;

            // Verify CRC integrity *before* trusting the entry for the liveness
            // decision or copying it into the compacted output (V-10 fix). This
            // must run before the liveness skip below: a corrupt key would
            // otherwise be parsed into the wrong key, misclassified as superseded,
            // and silently dropped when the source file is removed — instead of
            // surfacing as a CrcMismatch that aborts compaction and preserves the
            // source file.
            let key_start = size_of::<EntryHeader>();
            let key_end = key_start + size_of::<K>();
            let val_end = key_end + header.value_len as usize;
            let computed_crc = crate::entry::compute_crc32(
                header.gsn,
                header.value_len,
                &entry_bytes[key_start..key_end],
                &entry_bytes[key_end..val_end],
            );
            if computed_crc != header.crc32 {
                // Abort and preserve every source file. The OutputGuard removes
                // the finalized-but-unregistered outputs and the in-progress tmp
                // on drop (V-10 policy; shared with the C-01 aborts above).
                return Err(crate::error::DbError::CrcMismatch {
                    expected: header.crc32,
                    actual: computed_crc,
                });
            }

            let key: K = K::from_bytes(&entry_bytes[key_start..key_end]);
            let is_tombstone = header.is_tombstone();

            // Skip superseded entries: only the live version of each key is
            // carried into the compacted file. The CRC above has already vouched
            // for the key bytes, so this classification is trustworthy. A non-live
            // entry can never become live again (a newer version with a higher GSN
            // exists), so dropping it here is permanently safe even under
            // concurrent writers — the post-copy `update_if_match` CAS below is the
            // authoritative guard for the live-then-overwritten race. Without this
            // filter compaction copies every dead version forward verbatim and
            // reclaims nothing under overwrite/delete churn (output == input size).
            let live = if is_tombstone {
                !index.contains_key(&key)
            } else {
                index.is_live(shard.id, &key, old_loc)
            };
            if !live {
                offset += total;
                continue;
            }

            // Rotate output file if this (live) entry would push past max_file_size.
            // Skip rotation when output is empty: an oversized entry writes into
            // an empty file regardless (defensive guard, not reachable today due
            // to Config::validate invariants).
            if output.write_offset > 0
                && should_rotate_output(output.write_offset, total, max_file_size)
            {
                let next_file_id = {
                    let mut inner = shard.lock();
                    inner.allocate_file_id()?
                };
                let finished = finalize_output(shard, output)?;
                cleanup.mark_finalized(finished.file_id);
                pending_outputs.push(finished);
                output = open_output(shard, next_file_id)?;
                cleanup.set_current(next_file_id);
            }

            // Write to output (streaming encryption or direct).
            #[cfg(feature = "encryption")]
            if let Some(ref mut enc) = output.enc_state {
                enc.write_bytes(&output.tmp_file, output.file_id, &entry_bytes)?;
            } else {
                direct::pwrite_at(&output.tmp_file, &entry_bytes, output.write_offset)?;
            }
            #[cfg(not(feature = "encryption"))]
            direct::pwrite_at(&output.tmp_file, &entry_bytes, output.write_offset)?;

            let value_offset =
                output.write_offset + size_of::<EntryHeader>() as u64 + size_of::<K>() as u64;
            debug_assert!(value_offset <= u32::MAX as u64);
            let new_loc = DiskLoc::new(output.file_id, value_offset as u32, header.value_len);

            batch.push(BatchEntry {
                key,
                gsn: header.gsn,
                crc32: header.crc32,
                old_loc,
                new_loc,
                is_tombstone,
            });

            output.write_offset += total;
            offset += total;
        }
        direct::fadvise_dontneed(&scan_file, 0, file_len);
    }

    // Finalize the last output file only if something was written to it.
    if output.write_offset > 0 {
        let last_output = finalize_output(shard, output)?;
        cleanup.mark_finalized(last_output.file_id);
        pending_outputs.push(last_output);
    } else {
        // Empty output (no live entries in compaction set). Remove the
        // {file_id}.data.tmp (and tags.tmp if encrypted) that open_output
        // already created so no orphans are left on disk.
        let tmp_data_path = shard.dir().join(format!("{:06}.data.tmp", output.file_id));
        let _ = std::fs::remove_file(&tmp_data_path);
        #[cfg(feature = "encryption")]
        {
            let tmp_tags_path = shard.dir().join(format!("{:06}.tags.tmp", output.file_id));
            let _ = std::fs::remove_file(&tmp_tags_path);
        }
        cleanup.clear_current();
    }

    // Critical section 1: Expose all new output files to readers.
    // Open all read handles outside the lock so that a failure cannot leave
    // the shard with partially registered outputs.
    #[cfg(feature = "encryption")]
    let new_immutables: Vec<Arc<ImmutableFile>> = pending_outputs
        .iter()
        .map(|out| {
            let final_file = direct::open_serving_read(&out.final_data_path, serving_direct)?;
            let final_tag_file = if let Some(ref tp) = out.final_tag_path {
                Some(Arc::new(TagFile::open_read(tp)?))
            } else {
                None
            };
            Ok(Arc::new(ImmutableFile {
                file: final_file,
                file_id: out.file_id,
                path: out.final_data_path.clone(),
                total_bytes: out.total_bytes,
                tag_file: final_tag_file,
            }))
        })
        .collect::<DbResult<Vec<_>>>()?;

    #[cfg(not(feature = "encryption"))]
    let new_immutables: Vec<Arc<ImmutableFile>> = pending_outputs
        .iter()
        .map(|out| {
            let final_file = direct::open_serving_read(&out.final_data_path, serving_direct)?;
            Ok(Arc::new(ImmutableFile {
                file: final_file,
                file_id: out.file_id,
                total_bytes: out.total_bytes,
            }))
        })
        .collect::<DbResult<Vec<_>>>()?;

    {
        let mut inner = shard.lock();
        inner.immutable.extend(new_immutables);
        inner.immutable.sort_by_key(|f| f.file_id);
        inner.last_compaction_output_ids = pending_outputs.iter().map(|o| o.file_id).collect();
    }
    // Point of no return: the outputs now serve reads. A later error (hint
    // write, flush, old-file deletion) must not delete them.
    cleanup.disarm();

    let mut compacted_entries = 0;
    let key_len = size_of::<K>();

    // Build per-output-file hint buffers: entries that survive the index update
    // are kept; superseded entries are accounted as dead bytes.
    // The hint_data already stored in each output contains ALL entries written
    // to that file. We rebuild per-file live hint buffers during the index pass.
    let mut live_hint_data: std::collections::HashMap<u32, Vec<u8>> = pending_outputs
        .iter()
        .map(|o| (o.file_id, Vec::new()))
        .collect();

    for chunk in batch.chunks(BATCH_SIZE) {
        let mut inner = shard.lock();
        for entry in chunk {
            if entry.is_tombstone {
                if index.contains_key(&entry.key) {
                    inner.add_dead_bytes(
                        entry.new_loc.file_id,
                        entry_size(size_of::<K>(), entry.new_loc.len),
                    );
                } else {
                    compacted_entries += 1;
                    if let Some(buf) = live_hint_data.get_mut(&entry.new_loc.file_id) {
                        append_hint_entry(
                            buf,
                            entry.gsn,
                            &entry.key,
                            entry.new_loc.offset as u64,
                            entry.new_loc.len,
                            entry.crc32,
                            key_len,
                        );
                    }
                }
            } else if index.update_if_match(&entry.key, entry.old_loc, entry.new_loc) {
                compacted_entries += 1;
                if let Some(buf) = live_hint_data.get_mut(&entry.new_loc.file_id) {
                    append_hint_entry(
                        buf,
                        entry.gsn,
                        &entry.key,
                        entry.new_loc.offset as u64,
                        entry.new_loc.len,
                        entry.crc32,
                        key_len,
                    );
                }
            } else {
                inner.add_dead_bytes(
                    entry.new_loc.file_id,
                    entry_size(size_of::<K>(), entry.new_loc.len),
                );
            }
        }
    }

    // Critical section 2: Remove old files and invalidate cache entries atomically (V-11 fix).
    {
        let mut inner = shard.lock();
        inner
            .immutable
            .retain(|f| !old_file_ids.contains(&f.file_id));
        for fid in &old_file_ids {
            inner.dead_bytes.remove(fid);
        }
        for old_arc in &files_to_compact {
            index.invalidate_blocks(shard.id, old_arc.file_id, old_arc.total_bytes);
        }
    }

    for old_arc in &files_to_compact {
        direct::fadvise_dontneed(&old_arc.file, 0, old_arc.total_bytes);
    }

    // Write hint files for all output files. Each output is sealed (immutable),
    // so its logical end is fixed: stamp the v3 header with `out.total_bytes` so
    // recovery's tail-padding check can validate the file (see `hint_replay`).
    for out in &pending_outputs {
        let entries = live_hint_data.remove(&out.file_id).unwrap_or_default();
        let hint_data = hint::hint_file_bytes(&entries, out.total_bytes, size_of::<K>())?;
        let hint_path = shard.dir().join(format!("{:06}.hint", out.file_id));
        hint::write_hint_file(&hint_path, &hint_data)?;
    }
    dir_fsync(shard.dir())?;

    // Crash-consistency: a source file about to be deleted may hold the last
    // *durable* version of a key whose newer version is still only in the active
    // write buffer (Bitcask `append_entry` buffers without disk I/O). The liveness
    // skip above drops that old version from the compacted output, so deleting the
    // source while the newer value is unflushed would lose the key entirely on a
    // crash before flush. Flush + fsync the active file first so every key the
    // index points at has a durable on-disk home before its old copy is removed.
    shard.flush()?;

    // Delete old data, hint, and tag files
    for fid in &old_file_ids {
        let _ = fs::remove_file(shard.dir().join(format!("{fid:06}.data")));
        let _ = fs::remove_file(shard.dir().join(format!("{fid:06}.hint")));
        #[cfg(feature = "encryption")]
        let _ = fs::remove_file(shard.dir().join(format!("{fid:06}.tags")));
    }
    dir_fsync(shard.dir())?;

    let elapsed = compact_start.elapsed().as_secs_f64();
    metrics::counter!("armdb.compaction.runs").increment(1);
    metrics::counter!("armdb.compaction.entries").increment(compacted_entries as u64);
    metrics::histogram!("armdb.compaction.duration_seconds").record(elapsed);
    tracing::info!(
        entries = compacted_entries,
        files = old_file_ids.len(),
        elapsed_ms = (elapsed * 1000.0) as u64,
        "compaction complete"
    );
    Ok(compacted_entries)
}

/// Open a new compaction output file, returning an `OutputState` ready for writing.
fn open_output(shard: &Shard, file_id: u32) -> DbResult<OutputState> {
    #[cfg(test)]
    {
        let n = OPEN_OUTPUT_CALLS.with(|c| {
            let v = c.get() + 1;
            c.set(v);
            v
        });
        if FAIL_OPEN_OUTPUT_AT.with(|c| c.get()) == n {
            return Err(crate::error::DbError::Internal("test: open_output fault"));
        }
    }
    let tmp_path = shard.dir().join(format!("{file_id:06}.data.tmp"));
    match fs::remove_file(&tmp_path) {
        Ok(()) => {}
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
        Err(e) => return Err(e.into()),
    }
    let tmp_file = direct::open_bulk_write(&tmp_path)?;
    Ok(OutputState {
        file_id,
        tmp_file,
        write_offset: 0,
        #[cfg(feature = "encryption")]
        enc_state: shard.lock().cipher.clone().map(|cipher| EncryptionState {
            cipher,
            page_buf: Box::new([0u8; 4096]),
            page_offset: 0,
            pages_written: 0,
            tag_list: Vec::new(),
        }),
    })
}

/// Fsync, rename tmp → final, and return a `FinalizedOutput` descriptor.
/// Does NOT register the file as immutable — that happens later under the lock.
fn finalize_output(shard: &Shard, output: OutputState) -> DbResult<FinalizedOutput> {
    let file_id = output.file_id;
    let tmp_path = shard.dir().join(format!("{file_id:06}.data.tmp"));
    let final_data_path = shard.dir().join(format!("{file_id:06}.data"));

    #[cfg(feature = "encryption")]
    let (total_bytes, final_tag_path) = match output.enc_state {
        Some(mut enc) => {
            // Flush the final partial page if data remains.
            if enc.page_offset > 0 {
                enc.flush_page(&output.tmp_file, file_id)?;
            }

            let tp = shard.dir().join(format!("{file_id:06}.tags.tmp"));
            match fs::remove_file(&tp) {
                Ok(()) => {}
                Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
                Err(e) => return Err(e.into()),
            }
            let tf = TagFile::open_write(&tp)?;
            tf.write_tags(0, &enc.tag_list)?;
            tf.sync()?;

            let final_tags_path = shard.dir().join(format!("{file_id:06}.tags"));
            direct::fsync(&output.tmp_file)?;
            // C-03: rename the tag file into place *before* the data file. A
            // crash between the two then leaves an orphan `.tags` (inert — no
            // `.data`, so recovery never registers it) instead of an orphan
            // `.data` without its `.tags` (a poison pill: `make_reader` hard-fails
            // "tag file missing" and the whole DB refuses to open). The dir_fsync
            // between the renames pins that order on metadata-reordering fs'es.
            fs::rename(&tp, &final_tags_path)?;
            dir_fsync(shard.dir())?;
            #[cfg(test)]
            if CRASH_BETWEEN_RENAMES.with(|c| c.get()) {
                return Err(crate::error::DbError::Internal(
                    "test: crash between finalize renames",
                ));
            }
            fs::rename(&tmp_path, &final_data_path)?;
            dir_fsync(shard.dir())?;

            (enc.pages_written * 4096, Some(final_tags_path))
        }
        None => {
            direct::fsync(&output.tmp_file)?;
            fs::rename(&tmp_path, &final_data_path)?;
            dir_fsync(shard.dir())?;
            (output.write_offset, None)
        }
    };

    #[cfg(not(feature = "encryption"))]
    {
        direct::fsync(&output.tmp_file)?;
        fs::rename(&tmp_path, &final_data_path)?;
        dir_fsync(shard.dir())?;
    }

    Ok(FinalizedOutput {
        file_id,
        #[cfg(feature = "encryption")]
        total_bytes,
        #[cfg(not(feature = "encryption"))]
        total_bytes: output.write_offset,
        final_data_path,
        #[cfg(feature = "encryption")]
        final_tag_path,
    })
}

/// Background compaction handle.
pub struct Compactor {
    stop: crate::shutdown::ShutdownSignal,
    handle: Option<std::thread::JoinHandle<()>>,
}

impl Compactor {
    /// Start a background compaction thread with its own shutdown signal.
    pub fn start(
        compact_fn: impl Fn() -> DbResult<usize> + Send + 'static,
        interval: Duration,
    ) -> Self {
        Self::start_with_signal(compact_fn, interval, crate::shutdown::ShutdownSignal::new())
    }

    /// Start a background compaction thread controlled by an external shutdown
    /// signal. When the signal fires the thread wakes up immediately instead of
    /// waiting for the full sleep interval.
    pub fn start_with_signal(
        compact_fn: impl Fn() -> DbResult<usize> + Send + 'static,
        interval: Duration,
        signal: crate::shutdown::ShutdownSignal,
    ) -> Self {
        let stop = signal.clone();
        let handle = std::thread::spawn(move || {
            while !stop.is_shutdown() {
                if stop.wait_timeout(interval) {
                    break;
                }
                match compact_fn() {
                    Ok(n) if n > 0 => tracing::info!(entries = n, "compaction cycle"),
                    Err(e) => tracing::error!(error = %e, "compaction error"),
                    _ => {}
                }
            }
        });
        Self {
            stop: signal,
            handle: Some(handle),
        }
    }

    pub fn stop(&mut self) {
        self.stop.shutdown();
        if let Some(h) = self.handle.take() {
            let _ = h.join();
        }
    }
}

impl Drop for Compactor {
    fn drop(&mut self) {
        self.stop();
    }
}

/// Append a hint entry (GSN | Key | Offset | Len) to the buffer.
/// Same binary format as `hint::generate_hint_data_dyn`.
fn append_hint_entry<K: Key>(
    buf: &mut Vec<u8>,
    gsn: u64,
    key: &K,
    value_offset: u64,
    value_len: u32,
    crc32: u32,
    _key_len: usize,
) {
    buf.extend_from_slice(&gsn.to_ne_bytes());
    buf.extend_from_slice(key.as_bytes());
    buf.extend_from_slice(&value_offset.to_ne_bytes());
    buf.extend_from_slice(&value_len.to_ne_bytes());
    buf.extend_from_slice(&crc32.to_ne_bytes());
}

/// Read the maximum GSN from a hint file. Returns None if the hint file
/// doesn't exist or can't be parsed.
///
/// Recovery validates the complete v3 file before compaction can observe it, so
/// this guard intentionally reads only the checksummed fixed-size header.
fn file_max_gsn(hint_path: &std::path::Path, _file_id: u32, _key_len: usize) -> Option<u64> {
    let header = hint::read_hint_header(hint_path).ok()??;
    Some(header.max_gsn)
}

#[cfg(test)]
mod compaction_output_rotation_tests {
    use super::*;

    #[test]
    fn output_should_rotate_when_offset_plus_needed_exceeds_limit() {
        assert!(!should_rotate_output(0, 4096, 64 * 4096));
        assert!(!should_rotate_output(60 * 4096, 4096, 64 * 4096));
        assert!(should_rotate_output(63 * 4096, 4096 + 1, 64 * 4096));
        assert!(should_rotate_output(
            u32::MAX as u64 - 10,
            100,
            u32::MAX as u64
        ));
    }
}

#[cfg(test)]
mod compaction_padding_tests {
    use super::*;
    use crate::Config;
    use crate::ConstTree;
    use crate::entry::serialize_entry;
    use crate::hint::{generate_hint_data_dyn, hint_path_for_data, write_hint_file};
    use tempfile::tempdir;

    fn install_padded_immutable_file(shard_dir: &std::path::Path) -> u64 {
        let key_a = 1u64.to_be_bytes();
        let key_b = 2u64.to_be_bytes();
        let val_a_old = [0xAAu8; 8];
        let val_a_new = [0xCCu8; 8];
        let val_b = [0xBBu8; 8];

        let e1 = serialize_entry(1, &key_a, &val_a_old, false);
        let e2 = serialize_entry(3, &key_a, &val_a_new, false);
        let e3 = serialize_entry(4, &key_b, &val_b, false);

        // Layout: stale entry A, zero-pad to page boundary, live entry A, entry B.
        let mut file_bytes = Vec::new();
        file_bytes.extend_from_slice(&e1);
        let pad = 4096 - (file_bytes.len() % 4096);
        file_bytes.extend(std::iter::repeat_n(0u8, pad));
        file_bytes.extend_from_slice(&e2);
        file_bytes.extend_from_slice(&e3);
        let file_len = file_bytes.len() as u64;

        let path = shard_dir.join("000001.data");
        let f = direct::open_bulk_write(&path).unwrap();
        direct::pwrite_at(&f, &file_bytes, 0).unwrap();
        direct::fsync(&f).unwrap();
        drop(f);

        let rf = direct::open_bulk_read(&path).unwrap();
        let hint_data = generate_hint_data_dyn(&rf, file_len, size_of::<[u8; 8]>()).unwrap();
        write_hint_file(&hint_path_for_data(&path), &hint_data).unwrap();

        let path2 = shard_dir.join("000002.data");
        drop(direct::open_serving_write(&path2, false).unwrap());

        file_len
    }

    #[test]
    fn compaction_skips_mid_file_padding() {
        let dir = tempdir().unwrap();
        let mut config = Config::test();
        config.shard_count = 1;
        config.compaction_threshold = 0.0;

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

        let shard_dir = dir.path().join("shard_000");
        let _file_len = install_padded_immutable_file(&shard_dir);

        let tree = ConstTree::<[u8; 8], 8>::open(dir.path(), config).unwrap();
        assert_eq!(tree.len(), 2);

        let compacted = tree.compact().unwrap();
        assert!(
            compacted > 0,
            "compaction should run on padded immutable file"
        );

        let key_a = 1u64.to_be_bytes();
        let key_b = 2u64.to_be_bytes();
        assert_eq!(tree.get(&key_a), Some([0xCC; 8]));
        assert_eq!(tree.get(&key_b), Some([0xBB; 8]));

        let hint_path = shard_dir.join("000003.hint");
        if hint_path.exists() {
            let data = std::fs::read(&hint_path).unwrap();
            let entry_sz = crate::hint::hint_entry_size(size_of::<[u8; 8]>());
            // v3 hint = fixed header + N whole entries.
            let (_, entries) = crate::hint::parse_hint_header(&data).expect("v3 hint header");
            assert!(
                !entries.is_empty() && entries.len().is_multiple_of(entry_sz),
                "compaction hint entries {} not aligned to entry size {entry_sz}",
                entries.len()
            );
            let parsed: Vec<_> = crate::hint::parse_hint_entries::<[u8; 8]>(&data).collect();
            assert!(!parsed.is_empty());
        }
    }
}

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

    #[test]
    fn file_max_gsn_returns_true_max_not_last_entry() {
        let dir = tempdir().unwrap();
        let hint_path = dir.path().join("000001.hint");
        let key_len = 8;
        let entry_size = crate::hint::hint_entry_size(key_len);

        // Build a hint file with 3 entries: GSN 100, 300, 150
        // The true max is 300 (middle entry), not 150 (last entry).
        let mut data = Vec::new();
        for &gsn in &[100u64, 300u64, 150u64] {
            let start = data.len();
            data.resize(start + entry_size, 0);
            data[start..start + 8].copy_from_slice(&gsn.to_ne_bytes());
        }
        // Wrap the raw entries in a v3 header, then corrupt a payload byte.
        // Header-only max-GSN lookup must remain unaffected by payload damage.
        let mut hint_data = hint::hint_file_bytes(&data, 0, key_len).unwrap();
        hint_data[hint::HINT_HEADER_SIZE + entry_size] ^= 0xff;
        std::fs::write(&hint_path, hint_data).unwrap();

        let result = file_max_gsn(&hint_path, 1, key_len);
        assert_eq!(result, Some(300));
    }

    #[test]
    fn file_max_gsn_strips_tombstone_bit() {
        let dir = tempdir().unwrap();
        let hint_path = dir.path().join("000002.hint");
        let key_len = 8;
        let entry_size = crate::hint::hint_entry_size(key_len);

        let tombstone_gsn = 200u64 | crate::entry::TOMBSTONE_BIT;
        let mut data = vec![0u8; entry_size * 2];
        data[0..8].copy_from_slice(&100u64.to_ne_bytes());
        data[entry_size..entry_size + 8].copy_from_slice(&tombstone_gsn.to_ne_bytes());
        std::fs::write(
            &hint_path,
            hint::hint_file_bytes(&data, 0, key_len).unwrap(),
        )
        .unwrap();

        let result = file_max_gsn(&hint_path, 2, key_len);
        assert_eq!(result, Some(200));
    }
}

#[cfg(test)]
mod compaction_data_loss_tests {
    use super::*;
    use crate::Config;
    use crate::ConstTree;
    use crate::entry::{EntryHeader, compute_crc32, serialize_entry};
    use crate::hint::{hint_path_for_data, write_hint_file};
    use tempfile::tempdir;
    use zerocopy::IntoBytes;

    /// Install an immutable `000001.data` (+ hint) whose scan-order layout is
    /// `[stale A][live A][poison header][live B]`, plus an empty active
    /// `000002.data`. The poison header claims a `value_len` that overruns EOF,
    /// so a compaction scan takes the `offset + total > file_len` path *before*
    /// the CRC check — the exact anomalous break C-01 must turn into an abort.
    ///
    /// The hint is hand-built to reference A(live) and B so both are live in the
    /// index: a real data scan would stop at the poison and never index B, but
    /// here B is a legitimate live entry whose only durable home is this file.
    fn install_corrupt_immutable_file(
        shard_dir: &std::path::Path,
        key_a: &[u8; 8],
        val_a_new: &[u8; 8],
        key_b: &[u8; 8],
        val_b: &[u8; 8],
    ) {
        let val_a_old = [0xAAu8; 8];

        let e_a_stale = serialize_entry(1, key_a, &val_a_old, false); // 32B @0
        let e_a_live = serialize_entry(3, key_a, val_a_new, false); //   32B @32
        let poison = EntryHeader {
            gsn: 7,
            crc32: 0x1,
            value_len: 0xFFFF_FF00, // entry_size overruns EOF -> anomalous break
        };
        let e_b = serialize_entry(4, key_b, val_b, false); //             32B @80

        let mut file_bytes = Vec::new();
        file_bytes.extend_from_slice(&e_a_stale);
        file_bytes.extend_from_slice(&e_a_live);
        file_bytes.extend_from_slice(poison.as_bytes()); // 16B @64..80
        let b_off = file_bytes.len(); // 80
        file_bytes.extend_from_slice(&e_b);
        let file_len = file_bytes.len() as u64; // 112: logical end for the v3 hint

        let path = shard_dir.join("000001.data");
        let f = direct::open_bulk_write(&path).unwrap();
        direct::pwrite_at(&f, &file_bytes, 0).unwrap();
        direct::fsync(&f).unwrap();
        drop(f);

        // Hand-built hint: value_offset = entry_start + header(16) + key(8).
        let mut hint = Vec::new();
        append_hint_entry::<[u8; 8]>(
            &mut hint,
            1,
            key_a,
            24,
            8,
            compute_crc32(1, 8, key_a, &val_a_old),
            8,
        );
        append_hint_entry::<[u8; 8]>(
            &mut hint,
            3,
            key_a,
            56,
            8,
            compute_crc32(3, 8, key_a, val_a_new),
            8,
        );
        append_hint_entry::<[u8; 8]>(
            &mut hint,
            4,
            key_b,
            (b_off + 24) as u64,
            8,
            compute_crc32(4, 8, key_b, val_b),
            8,
        );
        write_hint_file(
            &hint_path_for_data(&path),
            &hint::hint_file_bytes(&hint, file_len, 8).unwrap(),
        )
        .unwrap();

        // Empty active file so the corrupt file stays immutable, not active.
        let path2 = shard_dir.join("000002.data");
        drop(direct::open_serving_write(&path2, false).unwrap());
    }

    /// C-01: a corrupt entry mid-file must make compaction abort and preserve the
    /// source, never silently retire the file and drop the live entry after it.
    #[test]
    fn compaction_aborts_on_corruption_and_preserves_live_tail() {
        let dir = tempdir().unwrap();
        let mut config = Config::test();
        config.shard_count = 1;
        config.compaction_threshold = 0.0;

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

        let shard_dir = dir.path().join("shard_000");
        let key_a = 1u64.to_be_bytes();
        let key_b = 2u64.to_be_bytes();
        let val_a_new = [0xCCu8; 8];
        let val_b = [0xBBu8; 8];
        install_corrupt_immutable_file(&shard_dir, &key_a, &val_a_new, &key_b, &val_b);

        let tree = ConstTree::<[u8; 8], 8>::open(dir.path(), config.clone()).unwrap();
        assert_eq!(tree.len(), 2, "both live entries recovered from the hint");
        assert_eq!(tree.get(&key_b), Some(val_b));

        // Compaction must abort — never silently retire the file and drop B.
        let result = tree.compact();
        assert!(
            result.is_err(),
            "compaction must abort on the corrupt entry, got {result:?}"
        );
        assert!(
            shard_dir.join("000001.data").exists(),
            "source file must be preserved on abort (holds live tail entry B)"
        );
        drop(tree);

        // Reopen: B's only durable home (000001) survived, so it is still there.
        let tree = ConstTree::<[u8; 8], 8>::open(dir.path(), config).unwrap();
        assert_eq!(
            tree.get(&key_b),
            Some(val_b),
            "live tail entry B must survive a corruption-aborted compaction"
        );
        assert_eq!(tree.get(&key_a), Some(val_a_new));
    }

    fn count_data_files(dir: &std::path::Path) -> (usize, usize) {
        let mut data = 0;
        let mut tmp = 0;
        for e in std::fs::read_dir(dir).unwrap() {
            let name = e.unwrap().file_name();
            let name = name.to_string_lossy();
            if name.ends_with(".data.tmp") {
                tmp += 1;
            } else if name.ends_with(".data") {
                data += 1;
            }
        }
        (data, tmp)
    }

    /// C-02: an error after the first output has been finalized (during a
    /// rotation) but before it is registered must not leave an orphan `.data`
    /// or `.data.tmp` on disk — the OutputGuard removes them.
    #[test]
    fn compaction_rotation_failure_leaves_no_orphan_outputs() {
        let dir = tempdir().unwrap();
        let mut config = Config::test();
        config.shard_count = 1;
        config.max_file_size = 4096;
        config.write_buffer_size = 4096;
        config.compaction_threshold = 0.1;

        let tree = ConstTree::<[u8; 8], 8>::open(dir.path(), config).unwrap();

        // Fill several immutable files, then overwrite a quarter of the keys so
        // the oldest files carry dead bytes *and* a large live majority — enough
        // live entries that compacting them rotates the output more than once.
        for i in 0..512u64 {
            tree.put(&i.to_be_bytes(), &i.to_be_bytes()).unwrap();
        }
        for i in (0..512u64).step_by(4) {
            tree.put(&i.to_be_bytes(), &(i + 1_000_000).to_be_bytes())
                .unwrap();
        }
        tree.flush().unwrap();

        let shard_dir = dir.path().join("shard_000");
        let (data_before, _) = count_data_files(&shard_dir);

        // Fail on the 2nd open_output — right after the first output has been
        // finalized during a rotation, before it is registered as immutable.
        OPEN_OUTPUT_CALLS.with(|c| c.set(0));
        FAIL_OPEN_OUTPUT_AT.with(|c| c.set(2));
        let result = tree.compact();
        FAIL_OPEN_OUTPUT_AT.with(|c| c.set(0));

        assert!(
            result.is_err(),
            "the injected rotation-open fault must surface, got {result:?}"
        );
        assert!(
            OPEN_OUTPUT_CALLS.with(|c| c.get()) >= 2,
            "test precondition: compaction must rotate the output at least once"
        );

        let (data_after, tmp_after) = count_data_files(&shard_dir);
        assert_eq!(
            tmp_after, 0,
            "no leftover .data.tmp after the aborted compaction"
        );
        assert_eq!(
            data_after, data_before,
            "OutputGuard must remove the finalized-but-unregistered orphan output"
        );
    }
}

/// C-03: a crash between the two encrypted `finalize_output` renames must leave
/// a recoverable DB. With tags renamed first, the crash leaves an inert orphan
/// `.tags`; before the fix it left a `.data` without its `.tags`, and the DB
/// refused to reopen ("tag file missing").
#[cfg(all(test, feature = "encryption"))]
mod compaction_encrypted_crash_tests {
    use super::*;
    use crate::Config;
    use crate::ConstTree;
    use tempfile::tempdir;

    #[test]
    fn crash_between_finalize_renames_reopens_cleanly() {
        let dir = tempdir().unwrap();
        let key = [0x42u8; 32];
        let mut config = Config::test();
        config.encryption_key = Some(key);
        config.shard_count = 1;
        config.max_file_size = 8192;
        config.write_buffer_size = 8192;
        config.compaction_threshold = 0.1;

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

        // Model a crash between the tag and data renames of the first finalized
        // encrypted output.
        CRASH_BETWEEN_RENAMES.with(|c| c.set(true));
        let result = tree.compact();
        CRASH_BETWEEN_RENAMES.with(|c| c.set(false));
        assert!(
            result.is_err(),
            "the injected crash must surface, got {result:?}"
        );

        // The orphaned compaction output holds the *highest* file id right after
        // the crash, so on its own it would reopen as the (masked) active file.
        // Model the real trigger — concurrent writers that keep rotating the
        // active file past the output's id — by appending enough fresh keys to
        // rotate the active file. The orphan is then a genuine *immutable* file
        // on reopen: with the old rename order it is a `.data` without `.tags`
        // (make_reader hard-fails "tag file missing"); with the fix it is an
        // inert `.tags` with no `.data`.
        for i in 1000..1400u64 {
            tree.put(&i.to_be_bytes(), &i.to_be_bytes()).unwrap();
        }
        drop(tree);

        // With the fix (tags renamed first), the crash left an inert orphan
        // `.tags` and every source intact, so the DB reopens and reads all data.
        // Before the fix it left a `.data` without its `.tags`, and the reopen
        // failed in `make_reader` with "tag file missing".
        let tree = ConstTree::<[u8; 8], 8>::open(dir.path(), config)
            .expect("DB must reopen after a crash between finalize renames");
        for i in 0..500u64 {
            let expected = if i < 250 {
                (i + 1000).to_be_bytes()
            } else {
                i.to_be_bytes()
            };
            assert_eq!(
                tree.get(&i.to_be_bytes()).unwrap(),
                expected,
                "post-crash reopen mismatch at key {i}"
            );
        }
    }
}