armdb 0.12.0

//! Crash-atomic transaction log (T5, part 1).
//!
//! Spec: `docs/superpowers/specs/26-08-05-crash-atomic-multi-collection-tx.md`.
//!
//! # What this module is
//!
//! The commit carrier for a multi-collection transaction. A record becomes
//! durable here *before* any collection is touched (no-steal), so recovery can
//! replay whatever the collections are missing, and an error before the commit
//! point leaves nothing behind anywhere.
//!
//! # Ordering rules
//!
//! - Replay order is `tx_id` order, never physical position: after a
//!   copy-forward an old record lives in the newest segment.
//! - A torn record is legal only at the tail of the newest segment; anywhere
//!   else it fails the open.
//! - A CRC-valid record from an unknown format version or with unknown flags
//!   fails the open. Garbage that merely *looks* like one is torn.
//!
//! # Durability rules
//!
//! - Segment create/unlink is followed by a directory fsync, as everywhere else
//!   in the crate (`engine.rs:36-45`, `recovery.rs:249-265`).
//! - A partial append is truncated back to the previous end; if that repair
//!   fails the log is poisoned and rejects further commits until reopen.
//! - `sync_upto` only fsyncs the active segment: sealing a segment fsyncs it,
//!   so a record in a sealed segment is already durable.
//! - Transaction ids come from a persisted ceiling, so they never repeat — not
//!   even after a checkpoint deletes every record.
//!
//! # What is NOT here
//!
//! Encryption of records (part 3), and everything that knows about collections:
//! the participant seam, the commit protocol, replay into collections and the
//! checkpoint cut (part 2). This module has no dependency on collections at
//! all, which is why it can be tested on its own.

pub(crate) mod group;
pub(crate) mod record;
pub(crate) mod segment;
pub(crate) mod state;

#[cfg(feature = "encryption")]
pub(crate) mod crypt;

/// The cipher a record is sealed with, as the rest of the log names it.
///
/// Without the `encryption` feature this is an **uninhabited** type, so
/// `Option<&LogCipher>` is statically `None` and every encryption branch is
/// dead code the optimiser drops. This keeps one signature for both builds —
/// `#[cfg]` on a function *parameter* would compile, but `#[cfg]` on the
/// matching *argument* at the call site does not, so the alias is what makes
/// the feature-free build spell the same calls.
#[cfg(feature = "encryption")]
pub(crate) type LogCipher = crypt::RecordCipher;

#[cfg(not(feature = "encryption"))]
#[derive(Debug)]
pub(crate) enum LogCipher {}

/// Unreachable by construction — a value of an uninhabited type cannot exist —
/// but present so the encryption branches type-check in the feature-free build.
#[cfg(not(feature = "encryption"))]
impl LogCipher {
    pub(crate) fn seal(&self, _tx_id: u64, _aad: &[u8], _plaintext: &[u8]) -> DbResult<Vec<u8>> {
        match *self {}
    }
    pub(crate) fn open(&self, _tx_id: u64, _aad: &[u8], _sealed: &[u8]) -> DbResult<Vec<u8>> {
        match *self {}
    }
}

use std::collections::{BTreeMap, BTreeSet};
use std::path::{Path, PathBuf};

use crate::error::{DbError, DbResult};
use crate::io::direct::sync_dir;
use crate::sync::Mutex;
use crate::txlog::group::SyncGroup;
use crate::txlog::record::TxRecord;
use crate::txlog::segment::{Segment, list_segment_ids};
use crate::txlog::state::TxIdAllocator;

#[derive(Debug, Clone, Copy)]
pub(crate) struct TxLogConfig {
    pub segment_bytes: u64,
    pub id_window: u64,
    #[cfg(feature = "encryption")]
    pub log_encryption_key: Option<[u8; 32]>,
}

impl Default for TxLogConfig {
    fn default() -> Self {
        Self {
            segment_bytes: 16 << 20,
            id_window: 4096,
            #[cfg(feature = "encryption")]
            log_encryption_key: None,
        }
    }
}

/// Which segment each known record lives in — the checkpoint needs it to decide
/// what can be unlinked and what must be copied forward.
struct RecordSlot {
    segment: u32,
    record: TxRecord,
    applied: bool,
}

struct Writer {
    active: Segment,
    /// Sealed segments, oldest first.
    sealed: Vec<u32>,
    poisoned: Option<String>,
}

pub(crate) struct TxLog {
    dir: PathBuf,
    cfg: TxLogConfig,
    ids: Mutex<TxIdAllocator>,
    writer: Mutex<Writer>,
    group: SyncGroup,
    records: Mutex<BTreeMap<u64, RecordSlot>>,
    /// Commit ids between log append and `mark_applied` — drives the cut.
    in_flight: Mutex<BTreeSet<u64>>,
    /// Ids read off disk at open. Fixed for the life of the log: only these
    /// records are candidates for open-time replay settling, which is a
    /// statement about a *previous* process's work. Records appended by this
    /// process settle through their own commit.
    recovered: BTreeSet<u64>,
    /// `None` in a build without the `encryption` feature, where `LogCipher` is
    /// uninhabited — so the field costs nothing and needs no `#[cfg]`.
    cipher: Option<LogCipher>,
}

/// A record that reached the log. `seq` orders it against other appends for
/// group commit.
#[derive(Debug, Clone, Copy)]
pub(crate) struct Committed {
    #[allow(dead_code)] // read by part 2 committers after `sync_upto`
    pub tx_id: u64,
    pub seq: u64,
}

/// Counts fsyncs of the log file (tests only): the group-commit assertion needs
/// to see that N committers cost fewer than N syncs.
#[cfg(test)]
pub(crate) static TXLOG_FSYNCS: std::sync::atomic::AtomicUsize =
    std::sync::atomic::AtomicUsize::new(0);

/// Skip the unlink half of a checkpoint (tests only): reproduces a crash
/// between copy-forward and unlink, where both copies exist.
#[cfg(test)]
pub(crate) static SKIP_UNLINK: std::sync::atomic::AtomicBool =
    std::sync::atomic::AtomicBool::new(false);

impl TxLog {
    pub(crate) fn open(root: &Path, cfg: TxLogConfig) -> DbResult<Self> {
        let dir = root.join("txlog");
        if !dir.is_dir() {
            std::fs::create_dir_all(&dir)?;
            sync_dir(root)?;
        }

        #[cfg(not(feature = "encryption"))]
        let cipher: Option<LogCipher> = None;
        #[cfg(feature = "encryption")]
        let cipher = match cfg.log_encryption_key {
            Some(key) => {
                // The log keeps its own salt next to its segments: a collection's
                // `db.salt` lives in that collection's directory (`engine.rs`),
                // and the log outlives any single collection.
                //
                // Generate only when the sidecar is absent (first encrypted open).
                // A present but wrong-sized/unreadable salt must fail closed —
                // rewriting it would orphan every sealed record under the old key.
                let salt_path = dir.join("db.salt");
                let salt = if salt_path.exists() {
                    crate::salt::read(&dir)?
                } else {
                    let s = crate::salt::generate()?;
                    crate::salt::write(&dir, &s)?;
                    s
                };
                Some(crypt::RecordCipher::derive(&key, &salt)?)
            }
            None => None,
        };

        let mut ids = TxIdAllocator::open(&dir, cfg.id_window)?;
        let segment_ids = list_segment_ids(&dir)?;

        let mut records: BTreeMap<u64, RecordSlot> = BTreeMap::new();
        let mut newest_torn_at: Option<u64> = None;
        for (pos, &sid) in segment_ids.iter().enumerate() {
            let seg = Segment::open(&dir, sid)?;
            let scanned = seg.scan(cipher.as_ref())?;
            let is_newest = pos + 1 == segment_ids.len();
            if let Some(off) = scanned.torn_at {
                if !is_newest {
                    return Err(DbError::FormatMismatch(format!(
                        "txlog segment {sid:06} has a torn record at offset {off}; \
                         only the newest segment may end torn"
                    )));
                }
                newest_torn_at = Some(off);
            }
            for record in scanned.records {
                // A record copied forward exists twice until the old segment is
                // unlinked; the newer segment wins, both copies are identical.
                records.insert(
                    record.tx_id,
                    RecordSlot {
                        segment: sid,
                        record,
                        applied: false,
                    },
                );
            }
        }

        // The ceiling normally covers everything on disk: it is persisted before
        // any id inside its window is handed out, so a crash cannot leave a
        // record above it. A *restore* can — `state` is a separate file from the
        // segments. Reissuing a recovered id would overwrite that record here
        // and, under encryption, repeat its GCM nonce under the same key.
        if let Some(&highest) = records.keys().next_back() {
            ids.ensure_above(highest)?;
        }

        let (mut active, sealed) = match segment_ids.last() {
            Some(&last) => (
                Segment::open(&dir, last)?,
                segment_ids[..segment_ids.len() - 1].to_vec(),
            ),
            None => (Segment::create(&dir, 1)?, Vec::new()),
        };
        if let Some(off) = newest_torn_at {
            active.truncate_to(off)?;
        }

        let recovered: BTreeSet<u64> = records.keys().copied().collect();

        Ok(Self {
            dir,
            cfg,
            ids: Mutex::new(ids),
            writer: Mutex::new(Writer {
                active,
                sealed,
                poisoned: None,
            }),
            group: SyncGroup::new(),
            records: Mutex::new(records),
            in_flight: Mutex::new(BTreeSet::new()),
            recovered,
            cipher,
        })
    }

    pub(crate) fn is_encrypted(&self) -> bool {
        self.cipher.is_some()
    }

    /// Was this record read off disk at open, rather than appended by this
    /// process? Only those are settled by open-time replay.
    pub(crate) fn is_recovered(&self, tx_id: u64) -> bool {
        self.recovered.contains(&tx_id)
    }

    /// Records recovered at open plus records committed since, ordered by
    /// `tx_id`. Replay order is this order, never physical position.
    pub(crate) fn pending(&self) -> Vec<TxRecord> {
        crate::sync::lock(&self.records)
            .values()
            .map(|slot| slot.record.clone())
            .collect()
    }

    pub(crate) fn total_bytes(&self) -> u64 {
        let w = crate::sync::lock(&self.writer);
        let mut total = w.active.end_offset();
        for &sid in &w.sealed {
            if let Ok(meta) = std::fs::metadata(crate::txlog::segment::segment_path(&self.dir, sid))
            {
                total += meta.len();
            }
        }
        total
    }

    pub(crate) fn next_tx_id(&self) -> DbResult<u64> {
        crate::sync::lock(&self.ids).next_id()
    }

    pub(crate) fn append(&self, rec: &TxRecord) -> DbResult<Committed> {
        let mut bytes = Vec::new();
        crate::txlog::record::encode_record_with(rec, self.cipher.as_ref(), &mut bytes)?;

        let mut w = crate::sync::lock(&self.writer);
        if let Some(reason) = &w.poisoned {
            return Err(DbError::TxLogPoisoned(reason.clone()));
        }

        // Rotate first, so a record never straddles two segments. Sealing
        // fsyncs the outgoing segment, which is what lets `sync_upto` fsync
        // only the active one.
        if w.active.end_offset() > 0
            && w.active.end_offset() + bytes.len() as u64 > self.cfg.segment_bytes
        {
            w.active.sync()?;
            let next_id = w.active.id() + 1;
            let sealed = std::mem::replace(&mut w.active, Segment::create(&self.dir, next_id)?);
            w.sealed.push(sealed.id());
        }

        let segment_id = w.active.id();
        match w.active.append(&bytes)? {
            crate::txlog::segment::AppendOutcome::Written { .. } => {}
            crate::txlog::segment::AppendOutcome::FailedAndTruncated(e) => return Err(e),
            crate::txlog::segment::AppendOutcome::FailedAndPoisoned(e) => {
                w.poisoned = Some(e.to_string());
                return Err(e);
            }
        }
        let seq = self.group.record_write();
        // Register the record before releasing the writer lock. Checkpoint
        // holds `writer` then reads `records`; if we drop `writer` first, a
        // concurrent checkpoint can rewrite/truncate the active segment without
        // this tail record and leave a committed-but-unreplayable hole.
        crate::sync::lock(&self.records).insert(
            rec.tx_id,
            RecordSlot {
                segment: segment_id,
                record: rec.clone(),
                applied: false,
            },
        );
        crate::sync::lock(&self.in_flight).insert(rec.tx_id);
        drop(w);

        Ok(Committed {
            tx_id: rec.tx_id,
            seq,
        })
    }

    pub(crate) fn sync_upto(&self, c: &Committed) -> DbResult<()> {
        self.group.sync_upto(c.seq, &|| {
            #[cfg(test)]
            TXLOG_FSYNCS.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            crate::sync::lock(&self.writer).active.sync()
        })
    }

    pub(crate) fn mark_applied(&self, tx_id: u64) {
        if let Some(slot) = crate::sync::lock(&self.records).get_mut(&tx_id) {
            slot.applied = true;
        }
        self.clear_in_flight(tx_id);
    }

    /// Drop an in-flight pin without marking the record applied. Used when
    /// apply fails after a durable append: the record must stay pending for
    /// replay, but must not freeze `in_flight_floor` forever.
    pub(crate) fn clear_in_flight(&self, tx_id: u64) {
        crate::sync::lock(&self.in_flight).remove(&tx_id);
    }

    /// Highest `tx_id` safe to retire: one below the lowest in-flight commit, or
    /// the highest committed id when nothing is in flight.
    pub(crate) fn in_flight_floor(&self) -> u64 {
        let in_flight = crate::sync::lock(&self.in_flight);
        if let Some(&min) = in_flight.iter().next() {
            return min.saturating_sub(1);
        }
        drop(in_flight);
        crate::sync::lock(&self.records)
            .keys()
            .last()
            .copied()
            .unwrap_or(0)
    }

    /// Retire every record that is applied and at or below `cut`. Sealed
    /// segments holding only such records are unlinked; a segment that still
    /// holds survivors has them copied into a fresh segment first, so one stuck
    /// record pins its own bytes rather than a whole segment. Applied records
    /// in the active segment are retired the same way — via copy-forward when
    /// survivors remain, or by truncating the active tail when they do not.
    pub(crate) fn checkpoint(&self, cut: u64) -> DbResult<usize> {
        let mut w = crate::sync::lock(&self.writer);
        if w.poisoned.is_some() {
            return Ok(0);
        }
        let sealed = w.sealed.clone();
        let active_id = w.active.id();

        let mut records = crate::sync::lock(&self.records);
        let mut survivors: Vec<TxRecord> = Vec::new();
        // `(tx_id, segment)`: a record is only forgotten once the bytes holding
        // it are actually gone, so the segment it lives in has to be carried
        // this far.
        let mut retired_ids: Vec<(u64, u32)> = Vec::new();

        let is_retirable = |segment: u32| sealed.contains(&segment) || segment == active_id;

        for (&tx_id, slot) in records.iter() {
            if !is_retirable(slot.segment) {
                continue;
            }
            if slot.applied && tx_id <= cut {
                retired_ids.push((tx_id, slot.segment));
            } else {
                survivors.push(slot.record.clone());
            }
        }
        if retired_ids.is_empty() {
            return Ok(0);
        }

        // 1. Copy survivors into a new segment and make it durable, before the
        //    old segments are unlinked.
        let mut moved: Vec<(u64, u32)> = Vec::new();
        let mut replaced_active: Option<u32> = None;
        if !survivors.is_empty() {
            let next_id = w.active.id() + 1;
            let mut fresh = Segment::create(&self.dir, next_id)?;
            let fill = (|| -> DbResult<()> {
                for record in &survivors {
                    let mut bytes = Vec::new();
                    crate::txlog::record::encode_record_with(
                        record,
                        self.cipher.as_ref(),
                        &mut bytes,
                    )?;
                    match fresh.append(&bytes)? {
                        crate::txlog::segment::AppendOutcome::Written { .. } => {}
                        crate::txlog::segment::AppendOutcome::FailedAndTruncated(e)
                        | crate::txlog::segment::AppendOutcome::FailedAndPoisoned(e) => {
                            return Err(e);
                        }
                    }
                    moved.push((record.tx_id, next_id));
                }
                fresh.sync()?;
                Ok(())
            })();
            if let Err(e) = fill {
                // Drop the orphan so the next rotation can recreate `next_id`.
                // Checkpoint is not on the process commit path — do not poison the log.
                let _ = fresh.remove(&self.dir);
                return Err(e);
            }
            // Sealing must fsync the outgoing active: `sync_upto` only syncs the
            // new active, matching the append-rotation path.
            if let Err(e) = w.active.sync() {
                let _ = fresh.remove(&self.dir);
                return Err(e);
            }
            // The fresh segment becomes the active one; the previous active is
            // sealed so segment ids stay ordered oldest-to-newest.
            let previous = std::mem::replace(&mut w.active, fresh);
            replaced_active = Some(previous.id());
            w.sealed.push(previous.id());
        } else {
            w.active.truncate_to(0)?;
        }

        // 2. Unlink retired segments; `Segment::remove` fsyncs the directory.
        let mut survivor_segments: std::collections::HashSet<u32> =
            std::collections::HashSet::new();
        for (&tx_id, slot) in records.iter() {
            if !sealed.contains(&slot.segment) {
                continue;
            }
            if !(slot.applied && tx_id <= cut) {
                survivor_segments.insert(slot.segment);
            }
        }

        #[cfg(test)]
        let skip_unlink = SKIP_UNLINK.load(std::sync::atomic::Ordering::SeqCst);
        #[cfg(not(test))]
        let skip_unlink = false;

        // The bookkeeping below has to reflect what actually happened, segment by
        // segment. Returning early on the first failure used to leave `w.sealed`
        // naming segments that were already gone, so the next checkpoint reopened
        // one of them, got `NotFound`, and failed the same way forever — while
        // `maybe_checkpoint` merely logged the error and the log kept growing.
        let mut unlinked: Vec<u32> = Vec::new();
        let mut unlink_err = None;
        for sid in sealed
            .iter()
            .chain(replaced_active.iter())
            .copied()
            .collect::<Vec<_>>()
        {
            if skip_unlink && survivor_segments.contains(&sid) {
                continue;
            }
            let outcome = match Segment::open(&self.dir, sid) {
                Ok(seg) => seg.remove(&self.dir),
                // Already gone — an earlier checkpoint unlinked it and then
                // failed on a later one. Accepting that is what lets the retry
                // make progress.
                Err(DbError::Io(e)) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
                Err(e) => Err(e),
            };
            match outcome {
                Ok(()) => unlinked.push(sid),
                Err(e) => {
                    unlink_err = Some(e);
                    break;
                }
            }
        }
        w.sealed.retain(|sid| !unlinked.contains(sid));

        // A record is forgotten only when the bytes holding it are gone: the
        // segment was unlinked, or it was the active one and got truncated back
        // to zero. Dropping a record whose segment survived a failed unlink
        // would leave nothing retirable, so the retry would return `Ok(0)`
        // before reaching the unlink loop and the file would stay until a
        // restart scanned it back in as pending.
        let truncated_active = survivors.is_empty().then_some(active_id);
        let cleared =
            |segment: u32| unlinked.contains(&segment) || truncated_active == Some(segment);
        let mut retired = 0usize;
        for (tx_id, segment) in retired_ids {
            if cleared(segment) {
                records.remove(&tx_id);
                retired += 1;
            }
        }
        for (tx_id, segment) in moved {
            if let Some(slot) = records.get_mut(&tx_id) {
                slot.segment = segment;
            }
        }
        match unlink_err {
            Some(e) => Err(e),
            None => Ok(retired),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::txlog::record::{OpKind, TxOp, TxRecord, VersionTag, encode_record};
    use crate::txlog::segment::{Segment, segment_path};

    pub(super) fn rec(tx_id: u64) -> TxRecord {
        TxRecord {
            tx_id,
            ops: vec![TxOp {
                collection_id: 1,
                name: "t".to_string(),
                schema_fingerprint: 42,
                key: tx_id.to_le_bytes().to_vec(),
                kind: OpKind::Put,
                value: vec![7; 16],
                version: VersionTag::Bitcask { gsn: tx_id },
            }],
        }
    }

    fn write_segment(dir: &std::path::Path, id: u32, tx_ids: &[u64]) {
        let mut seg = Segment::create(dir, id).unwrap();
        for &tx in tx_ids {
            let mut b = Vec::new();
            encode_record(&rec(tx), &mut b).unwrap();
            seg.append(&b).unwrap();
        }
    }

    #[test]
    fn txlog_open_sorts_by_tx_id_across_segments() {
        let root = tempfile::tempdir().unwrap();
        let dir = root.path().join("txlog");
        std::fs::create_dir_all(&dir).unwrap();
        // Segment 2 holds an older record: exactly what copy-forward produces.
        write_segment(&dir, 1, &[10, 11]);
        write_segment(&dir, 2, &[3, 12]);

        let log = TxLog::open(root.path(), TxLogConfig::default()).unwrap();
        let ids: Vec<u64> = log.pending().iter().map(|r| r.tx_id).collect();
        assert_eq!(ids, vec![3, 10, 11, 12]);
    }

    /// `state` is a separate file from the segments; a partial restore can lose
    /// it while the segments (and, under encryption, the salt) survive. Handing
    /// out a `tx_id` that is already sealed on disk would repeat an AES-GCM
    /// nonce under the same key and overwrite the recovered record.
    #[test]
    fn txlog_open_never_reissues_a_recovered_tx_id() {
        let root = tempfile::tempdir().unwrap();
        let dir = root.path().join("txlog");
        std::fs::create_dir_all(&dir).unwrap();
        write_segment(&dir, 1, &[1, 2, 3]);
        assert!(!dir.join("state").exists(), "the ceiling was lost");

        let log = TxLog::open(root.path(), TxLogConfig::default()).unwrap();
        let next = log.next_tx_id().unwrap();
        assert!(
            next > 3,
            "tx_id {next} is already sealed on disk — reusing it repeats a nonce"
        );
    }

    #[test]
    fn txlog_open_deduplicates_a_copy_forward_crash() {
        let root = tempfile::tempdir().unwrap();
        let dir = root.path().join("txlog");
        std::fs::create_dir_all(&dir).unwrap();
        write_segment(&dir, 1, &[5, 6]);
        write_segment(&dir, 2, &[5]); // copied forward, old segment not yet unlinked

        let log = TxLog::open(root.path(), TxLogConfig::default()).unwrap();
        let ids: Vec<u64> = log.pending().iter().map(|r| r.tx_id).collect();
        assert_eq!(ids, vec![5, 6]);
    }

    #[test]
    fn txlog_open_accepts_a_torn_tail_in_the_newest_segment() {
        let root = tempfile::tempdir().unwrap();
        let dir = root.path().join("txlog");
        std::fs::create_dir_all(&dir).unwrap();
        write_segment(&dir, 1, &[1, 2]);
        let mut f = std::fs::OpenOptions::new()
            .append(true)
            .open(segment_path(&dir, 1))
            .unwrap();
        std::io::Write::write_all(&mut f, &[0xCD; 30]).unwrap();

        let log = TxLog::open(root.path(), TxLogConfig::default()).unwrap();
        assert_eq!(log.pending().len(), 2);
    }

    #[test]
    fn txlog_open_truncates_torn_tail_before_append() {
        let root = tempfile::tempdir().unwrap();
        let cfg = TxLogConfig::default();

        // 1. Create log, append a good record, sync.
        let first_tx_id = {
            let log = TxLog::open(root.path(), cfg).unwrap();
            let tx_id = log.next_tx_id().unwrap();
            let c = log.append(&rec(tx_id)).unwrap();
            log.sync_upto(&c).unwrap();
            tx_id
        };

        // 2. Append raw torn bytes after end_offset.
        let dir = root.path().join("txlog");
        let seg_path = segment_path(&dir, 1);
        let good_len = std::fs::metadata(&seg_path).unwrap().len();
        let mut f = std::fs::OpenOptions::new()
            .append(true)
            .open(&seg_path)
            .unwrap();
        std::io::Write::write_all(&mut f, &[0xAB; 40]).unwrap();

        // 3. Reopen via TxLog::open (truncates torn tail).
        let log = TxLog::open(root.path(), cfg).unwrap();
        assert_eq!(log.pending().len(), 1);
        assert_eq!(
            std::fs::metadata(&seg_path).unwrap().len(),
            good_len,
            "torn tail must be truncated on open"
        );

        // 4. Append a new record.
        let second_tx_id = log.next_tx_id().unwrap();
        let c = log.append(&rec(second_tx_id)).unwrap();
        log.sync_upto(&c).unwrap();
        drop(log);

        // 5. Drop + reopen again.
        let reopened = TxLog::open(root.path(), cfg).unwrap();

        // 6. Both original and new record visible.
        let ids: Vec<u64> = reopened.pending().iter().map(|r| r.tx_id).collect();
        assert_eq!(ids, vec![first_tx_id, second_tx_id]);
    }

    #[test]
    fn txlog_open_rejects_a_torn_record_in_an_older_segment() {
        let root = tempfile::tempdir().unwrap();
        let dir = root.path().join("txlog");
        std::fs::create_dir_all(&dir).unwrap();
        write_segment(&dir, 1, &[1]);
        write_segment(&dir, 2, &[2]);
        let mut f = std::fs::OpenOptions::new()
            .append(true)
            .open(segment_path(&dir, 1))
            .unwrap();
        std::io::Write::write_all(&mut f, &[0xCD; 30]).unwrap();

        let err = TxLog::open(root.path(), TxLogConfig::default())
            .err()
            .expect("a torn record before the newest segment is corruption");
        assert!(format!("{err}").contains("torn"), "{err}");
    }

    #[test]
    fn txlog_open_creates_the_directory_on_a_fresh_database() {
        let root = tempfile::tempdir().unwrap();
        let log = TxLog::open(root.path(), TxLogConfig::default()).unwrap();
        assert!(root.path().join("txlog").is_dir());
        assert!(log.pending().is_empty());
        assert_eq!(log.total_bytes(), 0);
    }

    #[test]
    fn txlog_append_rotates_and_keeps_every_record_readable() {
        let root = tempfile::tempdir().unwrap();
        let cfg = TxLogConfig {
            segment_bytes: 512, // forces rotation after a couple of records
            ..TxLogConfig::default()
        };
        let log = TxLog::open(root.path(), cfg).unwrap();
        for _ in 0..10 {
            let tx_id = log.next_tx_id().unwrap();
            log.append(&rec(tx_id)).unwrap();
        }
        assert!(
            crate::txlog::segment::list_segment_ids(&root.path().join("txlog"))
                .unwrap()
                .len()
                > 1,
            "the log must have rotated"
        );

        drop(log);
        let reopened = TxLog::open(root.path(), cfg).unwrap();
        assert_eq!(reopened.pending().len(), 10);
    }

    #[test]
    fn txlog_append_rejects_an_oversized_record() {
        let root = tempfile::tempdir().unwrap();
        let log = TxLog::open(root.path(), TxLogConfig::default()).unwrap();
        let mut big = rec(1);
        big.ops[0].value = vec![0u8; crate::txlog::record::MAX_RECORD_BYTES];
        assert!(log.append(&big).is_err());
    }

    #[test]
    fn txlog_append_poisons_after_a_failed_repair() {
        let root = tempfile::tempdir().unwrap();
        let log = TxLog::open(root.path(), TxLogConfig::default()).unwrap();

        crate::txlog::segment::FAIL_APPEND_AFTER.store(10, std::sync::atomic::Ordering::SeqCst);
        crate::txlog::segment::FAIL_TRUNCATE.store(true, std::sync::atomic::Ordering::SeqCst);
        let first = log.append(&rec(1));
        crate::txlog::segment::FAIL_APPEND_AFTER
            .store(usize::MAX, std::sync::atomic::Ordering::SeqCst);
        crate::txlog::segment::FAIL_TRUNCATE.store(false, std::sync::atomic::Ordering::SeqCst);

        assert!(first.is_err(), "the failed append must surface");
        let err = log.append(&rec(2)).expect_err("the log stays poisoned");
        assert!(matches!(err, crate::DbError::TxLogPoisoned(_)), "{err}");
    }

    #[test]
    fn txlog_sync_upto_syncs_the_log_file() {
        let root = tempfile::tempdir().unwrap();
        let log = TxLog::open(root.path(), TxLogConfig::default()).unwrap();
        let before = TXLOG_FSYNCS.load(std::sync::atomic::Ordering::SeqCst);
        let tx_id = log.next_tx_id().unwrap();
        let c = log.append(&rec(tx_id)).unwrap();
        log.sync_upto(&c).unwrap();
        assert!(TXLOG_FSYNCS.load(std::sync::atomic::Ordering::SeqCst) > before);
    }

    #[test]
    fn txlog_checkpoint_retires_only_applied_records_under_the_cut() {
        let root = tempfile::tempdir().unwrap();
        let cfg = TxLogConfig {
            segment_bytes: 256,
            ..TxLogConfig::default()
        };
        let log = TxLog::open(root.path(), cfg).unwrap();

        let mut ids = Vec::new();
        for _ in 0..6 {
            let tx_id = log.next_tx_id().unwrap();
            log.append(&rec(tx_id)).unwrap();
            ids.push(tx_id);
        }
        // Applied: all but the fifth. Cut excludes the sixth.
        for (i, id) in ids.iter().enumerate() {
            if i != 4 {
                log.mark_applied(*id);
            }
        }
        let cut = ids[4];

        let retired = log.checkpoint(cut).unwrap();
        assert_eq!(retired, 4, "ids 1-4 are applied and under the cut");

        let left: Vec<u64> = log.pending().iter().map(|r| r.tx_id).collect();
        assert_eq!(left, vec![ids[4], ids[5]]);

        drop(log);
        let reopened = TxLog::open(root.path(), cfg).unwrap();
        let after: Vec<u64> = reopened.pending().iter().map(|r| r.tx_id).collect();
        assert_eq!(after, vec![ids[4], ids[5]], "survivors outlive the restart");
    }

    /// A checkpoint that fails part-way through unlinking must leave a state the
    /// next checkpoint can work from.
    ///
    /// Unlinking walks the sealed list; an error after some segments are already
    /// gone used to return before the sealed list and the records' locations were
    /// updated. The next checkpoint then reopened a segment that no longer
    /// existed, got `NotFound`, and failed the same way forever — so the log grew
    /// without bound while `maybe_checkpoint` only logged the error.
    #[test]
    fn txlog_checkpoint_retries_after_an_unlink_failure() {
        use crate::txlog::segment::FAIL_REMOVE_SEGMENT;

        let root = tempfile::tempdir().unwrap();
        let cfg = TxLogConfig {
            segment_bytes: 128,
            ..TxLogConfig::default()
        };
        let log = TxLog::open(root.path(), cfg).unwrap();

        // One record per segment, so several sealed segments are retired at once.
        let mut ids = Vec::new();
        for _ in 0..5 {
            let tx_id = log.next_tx_id().unwrap();
            log.append(&rec(tx_id)).unwrap();
            log.mark_applied(tx_id);
            ids.push(tx_id);
        }
        let sealed_ids = crate::sync::lock(&log.writer).sealed.clone();
        assert!(
            sealed_ids.len() >= 2,
            "the scenario needs several sealed segments, got {sealed_ids:?}"
        );

        // The second sealed segment refuses to go.
        FAIL_REMOVE_SEGMENT.store(sealed_ids[1], std::sync::atomic::Ordering::SeqCst);
        let err = match log.checkpoint(*ids.last().unwrap()) {
            Err(e) => e,
            Ok(_) => panic!("the injected unlink failure must surface"),
        };
        assert!(
            format!("{err}").contains("injected segment remove failure"),
            "{err}"
        );
        FAIL_REMOVE_SEGMENT.store(u32::MAX, std::sync::atomic::Ordering::SeqCst);

        // The next checkpoint must not trip over the segments already unlinked,
        // and must finish the work the failed one left — a record whose segment
        // survived has to stay retirable, or the retry has nothing to do and the
        // file lives on until a restart scans it back in as pending.
        log.checkpoint(*ids.last().unwrap())
            .expect("a checkpoint must be able to resume after a failed unlink");
        for sid in &sealed_ids {
            assert!(
                !crate::txlog::segment::segment_path(&log.dir, *sid).exists(),
                "segment {sid} must be gone after the retry"
            );
        }
        assert!(
            log.pending().is_empty(),
            "every applied record under the cut must retire"
        );

        // And the restart must not resurrect them.
        drop(log);
        let reopened = TxLog::open(root.path(), cfg).unwrap();
        assert!(
            reopened.pending().is_empty(),
            "retired records must not come back from a segment left on disk"
        );
    }

    #[test]
    fn txlog_checkpoint_copy_forward_crash_leaves_no_duplicate_after_reopen() {
        let root = tempfile::tempdir().unwrap();
        let cfg = TxLogConfig {
            segment_bytes: 256,
            ..TxLogConfig::default()
        };
        let log = TxLog::open(root.path(), cfg).unwrap();

        let mut ids = Vec::new();
        for _ in 0..6 {
            let tx_id = log.next_tx_id().unwrap();
            log.append(&rec(tx_id)).unwrap();
            ids.push(tx_id);
        }
        log.mark_applied(ids[0]);
        log.mark_applied(ids[1]);

        // Crash between copy-forward and unlink: both copies on disk.
        SKIP_UNLINK.store(true, std::sync::atomic::Ordering::SeqCst);
        let _ = log.checkpoint(ids[1]).unwrap();
        SKIP_UNLINK.store(false, std::sync::atomic::Ordering::SeqCst);

        drop(log);
        let reopened = TxLog::open(root.path(), cfg).unwrap();
        let after: Vec<u64> = reopened.pending().iter().map(|r| r.tx_id).collect();
        let mut expected: Vec<u64> = ids[2..].to_vec();
        expected.sort_unstable();
        assert_eq!(
            after, expected,
            "each surviving record appears exactly once"
        );
    }

    #[test]
    fn txlog_checkpoint_keeps_records_whose_participants_never_opened() {
        let root = tempfile::tempdir().unwrap();
        let log = TxLog::open(root.path(), TxLogConfig::default()).unwrap();
        let tx_id = log.next_tx_id().unwrap();
        log.append(&rec(tx_id)).unwrap();
        // Never marked applied — a participant was not opened this session.
        assert_eq!(log.checkpoint(tx_id + 100).unwrap(), 0);
        assert_eq!(log.pending().len(), 1);
    }

    #[test]
    fn txlog_checkpoint_fsyncs_outgoing_active_before_seal() {
        let root = tempfile::tempdir().unwrap();
        let cfg = TxLogConfig {
            segment_bytes: 256,
            ..TxLogConfig::default()
        };
        let log = TxLog::open(root.path(), cfg).unwrap();
        let mut ids = Vec::new();
        for _ in 0..6 {
            let tx_id = log.next_tx_id().unwrap();
            log.append(&rec(tx_id)).unwrap();
            ids.push(tx_id);
        }
        log.mark_applied(ids[0]);
        log.mark_applied(ids[1]);

        let before = crate::txlog::segment::SEGMENT_SYNCS.load(std::sync::atomic::Ordering::SeqCst);
        assert!(log.checkpoint(ids[1]).unwrap() > 0);
        let after = crate::txlog::segment::SEGMENT_SYNCS.load(std::sync::atomic::Ordering::SeqCst);
        assert!(
            after >= before + 2,
            "copy-forward must sync the fresh segment and the outgoing active, before={before} after={after}"
        );
    }

    #[test]
    fn txlog_checkpoint_removes_orphan_on_sync_failure() {
        let root = tempfile::tempdir().unwrap();
        let cfg = TxLogConfig {
            segment_bytes: 256,
            ..TxLogConfig::default()
        };
        let log = TxLog::open(root.path(), cfg).unwrap();
        let mut ids = Vec::new();
        for _ in 0..6 {
            let tx_id = log.next_tx_id().unwrap();
            log.append(&rec(tx_id)).unwrap();
            ids.push(tx_id);
        }
        log.mark_applied(ids[0]);
        log.mark_applied(ids[1]);

        let dir = root.path().join("txlog");
        let active_id = *crate::txlog::segment::list_segment_ids(&dir)
            .unwrap()
            .last()
            .unwrap();
        let orphan_id = active_id + 1;

        crate::txlog::segment::FAIL_SYNC.store(true, std::sync::atomic::Ordering::SeqCst);
        let err = log.checkpoint(ids[1]);
        crate::txlog::segment::FAIL_SYNC.store(false, std::sync::atomic::Ordering::SeqCst);
        assert!(err.is_err(), "checkpoint must surface the sync failure");
        assert!(
            !crate::txlog::segment::segment_path(&dir, orphan_id).exists(),
            "failed copy-forward must not leave an orphan segment"
        );

        // Appends (and later rotations) must still work — log is not poisoned.
        let tx_id = log.next_tx_id().unwrap();
        log.append(&rec(tx_id)).unwrap();
    }

    #[test]
    fn txlog_restart_returns_every_committed_record_in_order() {
        let root = tempfile::tempdir().unwrap();
        let cfg = TxLogConfig {
            segment_bytes: 1024,
            id_window: 64,
            ..TxLogConfig::default()
        };

        let mut written = Vec::new();
        {
            let log = TxLog::open(root.path(), cfg).unwrap();
            for _ in 0..50 {
                let tx_id = log.next_tx_id().unwrap();
                let c = log.append(&rec(tx_id)).unwrap();
                log.sync_upto(&c).unwrap();
                written.push(tx_id);
            }
        }

        let log = TxLog::open(root.path(), cfg).unwrap();
        let seen: Vec<u64> = log.pending().iter().map(|r| r.tx_id).collect();
        assert_eq!(
            seen, written,
            "every committed record returns, in tx_id order"
        );

        // Ids never repeat after a restart, even inside a reserved window.
        let next = log.next_tx_id().unwrap();
        assert!(
            next > *written.last().unwrap(),
            "{next} must exceed every id already used"
        );
    }

    #[test]
    fn txlog_restart_keeps_the_stuck_record_and_drops_the_rest() {
        let root = tempfile::tempdir().unwrap();
        let cfg = TxLogConfig {
            segment_bytes: 1024,
            id_window: 64,
            ..TxLogConfig::default()
        };

        // Each record is well under 128 bytes, so 30 of them fill at least two
        // 1 KiB segments and leave a third active. A checkpoint retires only
        // SEALED segments by design, which is why the assertions are about the
        // first record (certainly sealed) and the stuck one, not an exact count.
        let (stuck, first);
        {
            let log = TxLog::open(root.path(), cfg).unwrap();
            let mut ids = Vec::new();
            for _ in 0..30 {
                let tx_id = log.next_tx_id().unwrap();
                log.append(&rec(tx_id)).unwrap();
                ids.push(tx_id);
            }
            // One record's participant was never opened, so it is never applied.
            stuck = ids[7];
            first = ids[0];
            for id in &ids {
                if *id != stuck {
                    log.mark_applied(*id);
                }
            }
            let cut = *ids.last().unwrap();
            let retired = log.checkpoint(cut).unwrap();
            assert!(retired > 0, "the oldest sealed segment must retire");
        }

        let log = TxLog::open(root.path(), cfg).unwrap();
        let left: Vec<u64> = log.pending().iter().map(|r| r.tx_id).collect();
        assert!(
            left.contains(&stuck),
            "an unapplied record is never retired"
        );
        assert!(
            !left.contains(&first),
            "an applied record under the cut is retired"
        );
        assert!(left.len() < 30, "the checkpoint made progress: {left:?}");
    }

    #[cfg(feature = "encryption")]
    #[test]
    fn txlog_generates_its_salt_once_and_reuses_it() {
        let dir = tempfile::tempdir().unwrap();
        let cfg = TxLogConfig {
            log_encryption_key: Some([0x5A; 32]),
            ..TxLogConfig::default()
        };
        let log = TxLog::open(dir.path(), cfg).unwrap();
        assert!(log.is_encrypted());
        let salt = std::fs::read(dir.path().join("txlog").join("db.salt")).unwrap();
        drop(log);

        let _log = TxLog::open(dir.path(), cfg).unwrap();
        assert_eq!(
            std::fs::read(dir.path().join("txlog").join("db.salt")).unwrap(),
            salt,
            "a second open must not regenerate the salt — that would orphan every record"
        );
    }

    #[cfg(feature = "encryption")]
    #[test]
    fn txlog_refuses_to_overwrite_a_malformed_salt() {
        let dir = tempfile::tempdir().unwrap();
        let cfg = TxLogConfig {
            log_encryption_key: Some([0x5A; 32]),
            ..TxLogConfig::default()
        };
        // First open creates a valid salt.
        let _ = TxLog::open(dir.path(), cfg).unwrap();
        let salt_path = dir.path().join("txlog").join("db.salt");
        let good = std::fs::read(&salt_path).unwrap();
        // Corrupt: valid prefix plus a trailing byte (still "exists").
        let mut bad = good.clone();
        bad.push(0xFF);
        std::fs::write(&salt_path, &bad).unwrap();

        let err = TxLog::open(dir.path(), cfg)
            .err()
            .expect("malformed salt must fail the open");
        assert!(
            matches!(err, DbError::FormatMismatch(_)),
            "malformed salt must fail closed, got {err}"
        );
        assert_eq!(
            std::fs::read(&salt_path).unwrap(),
            bad,
            "open must not rewrite a present salt sidecar"
        );
    }
}