powercut 0.1.0

//! The tier model: what survives which kind of failure, and how a captured
//! copy of it is built.

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

use crate::scratch::{self, ScratchDir};
use crate::witness::{canonical, AssumedDurable, DurabilityWitness, SyncKind, SyncRecord};
use crate::{Error, Result};

/// Which failure `powercut` is modelling.
///
/// The two tiers answer genuinely different questions and it is worth being
/// blunt about the gap between them, because most hand-rolled durability tests
/// measure tier 1 and describe it as tier 2.
///
/// | tier | models | keeps | a store that never calls `fsync` |
/// |---|---|---|---|
/// | [`Tier::ProcessCrash`] | `SIGKILL`, panic, OOM kill, OS crash | every byte the app wrote to the filesystem | **passes** |
/// | [`Tier::PowerCut`] | the plug coming out | only bytes a successful sync certified | fails |
///
/// Tier 1 reads through the OS page cache, so it can only ever prove *"the
/// write reached the file"*. Tier 2 is the one that means *"the write reached
/// stable storage"*.
///
/// ```
/// # use powercut::Tier;
/// assert_eq!(Tier::PowerCut.as_str(), "power-cut");
/// assert!(Tier::ProcessCrash.models().contains("page cache"));
/// ```
#[non_exhaustive]
#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub enum Tier {
    /// Tier 1 — a byte-for-byte copy of the store, taken at the ack.
    ///
    /// Models a process or OS crash: the machine keeps running long enough for
    /// the page cache to reach the filesystem, so anything the application
    /// wrote survives whether or not it was ever synced.
    ProcessCrash,
    /// Tier 2 — only the bytes a successful sync certified, truncated to the
    /// length each file had when it was synced.
    ///
    /// Models a real power cut: the page cache is gone.
    PowerCut,
}

impl Tier {
    /// A short, stable label for reports and baselines.
    ///
    /// ```
    /// # use powercut::Tier;
    /// assert_eq!(Tier::ProcessCrash.as_str(), "process-crash");
    /// ```
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Tier::ProcessCrash => "process-crash",
            Tier::PowerCut => "power-cut",
        }
    }

    /// One sentence on what this tier models, for failure messages.
    ///
    /// ```
    /// # use powercut::Tier;
    /// assert!(Tier::PowerCut.models().contains("fsync"));
    /// ```
    #[must_use]
    pub const fn models(self) -> &'static str {
        match self {
            Tier::ProcessCrash => {
                "a process or OS crash: the page cache still reaches the filesystem, \
                 so every byte the application wrote survives — including bytes no \
                 fsync ever certified"
            }
            Tier::PowerCut => {
                "a real power cut: the page cache is lost, so only bytes that a \
                 successful fsync certified before the ack survive"
            }
        }
    }
}

/// Why a file earned its place in a captured crash state.
///
/// ```
/// # use powercut::{Barrier, SyncKind, SyncRecord, Warrant};
/// let w = Warrant::Synced(SyncRecord::new("db", SyncKind::File, Barrier::new(1), Some(16)));
/// assert!(w.is_measured());
/// assert!(!Warrant::Assumed { reason: "setup".into() }.is_measured());
/// ```
#[non_exhaustive]
#[derive(Clone, Debug)]
pub enum Warrant {
    /// A successful sync certified it. This is the only *measured* warrant.
    Synced(SyncRecord),
    /// [`Tier::ProcessCrash`] keeps everything on the filesystem; no sync was
    /// required and none is claimed.
    OnFilesystem,
    /// Someone asserted it was durable. See [`AssumedDurable`].
    Assumed {
        /// Why the file was assumed durable.
        reason: String,
    },
}

impl Warrant {
    /// `true` only when a real, successful sync is behind this file.
    ///
    /// ```
    /// # use powercut::Warrant;
    /// assert!(!Warrant::OnFilesystem.is_measured());
    /// ```
    #[must_use]
    pub fn is_measured(&self) -> bool {
        matches!(self, Warrant::Synced(_))
    }

    /// A one-line rendering for reports.
    ///
    /// ```
    /// # use powercut::Warrant;
    /// assert_eq!(Warrant::OnFilesystem.describe(), "on the filesystem at the ack");
    /// ```
    #[must_use]
    pub fn describe(&self) -> String {
        match self {
            Warrant::Synced(r) => format!(
                "fsync({}) at barrier {}{}",
                r.kind.as_str(),
                r.at,
                r.durable_len
                    .map(|l| format!(", durable length {l}"))
                    .unwrap_or_default()
            ),
            Warrant::OnFilesystem => "on the filesystem at the ack".to_string(),
            Warrant::Assumed { reason } => format!("ASSUMED durable — {reason}"),
        }
    }
}

/// One file in a captured crash state.
///
/// ```
/// # use powercut::{CapturedFile, Warrant};
/// let f = CapturedFile {
///     path: "app.db".into(),
///     bytes: 4096,
///     warrant: Warrant::OnFilesystem,
///     truncated_from: None,
/// };
/// assert_eq!(f.bytes, 4096);
/// ```
#[derive(Clone, Debug)]
pub struct CapturedFile {
    /// Path relative to the store root.
    pub path: PathBuf,
    /// Bytes in the captured copy.
    pub bytes: u64,
    /// Why it is here.
    pub warrant: Warrant,
    /// `Some(live_len)` when the live file was longer than its durable length
    /// and the copy was cut back. The tail was written but never certified.
    pub truncated_from: Option<u64>,
}

/// A materialised directory holding exactly the state that survives at a
/// chosen [`Tier`], plus the manifest explaining every byte of it.
///
/// Reboot your store from [`CrashState::dir`] and check what came back.
///
/// ```
/// use powercut::{DurabilityWitness, ScratchDir, Scenario, SyncJournal, Tier};
///
/// let store = ScratchDir::new("doc-crashstate").unwrap();
/// let journal = SyncJournal::complete();
///
/// let acked = Scenario::new(store.path(), journal.clone())
///     .tier(Tier::PowerCut)
///     .ack(|| {
///         std::fs::write(store.path().join("durable"), b"kept").unwrap();
///         std::fs::write(store.path().join("volatile"), b"lost").unwrap();
///         journal.sync_path(store.path().join("durable")).unwrap();
///     })
///     .unwrap();
///
/// let crash = acked.crash_state();
/// assert!(crash.dir().join("durable").exists());
/// assert!(!crash.dir().join("volatile").exists(), "never fsynced, never durable");
/// assert_eq!(crash.manifest().len(), 1);
/// ```
#[derive(Debug)]
pub struct CrashState {
    dir: ScratchDir,
    tier: Tier,
    manifest: Vec<CapturedFile>,
    notes: Vec<String>,
}

impl CrashState {
    /// The directory holding the surviving bytes. Boot your store from here.
    ///
    /// It is a fresh directory, not the live store, so a recovery that
    /// rewrites files cannot corrupt what you were measuring.
    #[must_use]
    pub fn dir(&self) -> &Path {
        self.dir.path()
    }

    /// Which tier this state was captured at.
    #[must_use]
    pub fn tier(&self) -> Tier {
        self.tier
    }

    /// Every file that survived, and why.
    #[must_use]
    pub fn manifest(&self) -> &[CapturedFile] {
        &self.manifest
    }

    /// Observations the capture made that a human should read: a file that was
    /// synced and has since shrunk or vanished, a file synced without its
    /// parent directory ever being synced, a symlink that was skipped.
    ///
    /// These strings are diagnostics, not API. Do not match on them.
    #[must_use]
    pub fn notes(&self) -> &[String] {
        &self.notes
    }

    /// `true` if any surviving file leans on an [`AssumedDurable`] rather than
    /// a measured sync. A pass that used one is a weaker claim.
    #[must_use]
    pub fn used_assumption(&self) -> bool {
        self.manifest
            .iter()
            .any(|f| matches!(f.warrant, Warrant::Assumed { .. }))
    }

    /// Render the manifest as a human-readable block — files only; see
    /// [`CrashState::notes`] for the observations. This is what a failing
    /// assertion prints, and it is meant to answer *"why is that file in or
    /// out?"* without anyone having to open a debugger.
    #[must_use]
    pub fn manifest_text(&self) -> String {
        let mut out = format!(
            "surviving files at tier {} ({} file(s)):\n",
            self.tier.as_str(),
            self.manifest.len()
        );
        if self.manifest.is_empty() {
            out.push_str("  (none — nothing at all was certified durable)\n");
        }
        for f in &self.manifest {
            out.push_str(&format!(
                "  {:<28} {:>10} B  {}{}\n",
                f.path.display(),
                f.bytes,
                f.warrant.describe(),
                f.truncated_from
                    .map(|l| format!(" [truncated from {l} B live]"))
                    .unwrap_or_default()
            ));
        }
        out
    }

    /// Take a crash state from `store`, keeping only what survives at `tier`.
    ///
    /// You normally get one from [`Scenario::ack`](crate::Scenario::ack); this
    /// is here for harnesses that drive the capture themselves.
    ///
    /// # Errors
    ///
    /// Fails if the store directory cannot be read or the scratch copy cannot
    /// be written.
    ///
    /// ```
    /// use powercut::{CrashState, DurabilityWitness, ScratchDir, SyncJournal, Tier};
    ///
    /// let store = ScratchDir::new("doc-capture").unwrap();
    /// std::fs::write(store.path().join("a"), b"aaaa").unwrap();
    ///
    /// let journal = SyncJournal::complete();
    /// journal.sync_path(store.path().join("a")).unwrap();
    ///
    /// let crash = CrashState::capture(store.path(), Tier::PowerCut, &journal).unwrap();
    /// assert_eq!(crash.manifest()[0].bytes, 4);
    /// ```
    pub fn capture<W: DurabilityWitness>(
        store: impl AsRef<Path>,
        tier: Tier,
        witness: &W,
    ) -> Result<Self> {
        let store = store.as_ref();
        if !store.is_dir() {
            return Err(Error::StoreMissing(store.to_path_buf()));
        }
        let dir = ScratchDir::new(&format!("crash-{}", tier.as_str()))
            .map_err(|e| Error::io(Some(store.to_path_buf()), e))?;

        let (files, symlinks) =
            scratch::walk_files(store).map_err(|e| Error::io(Some(store.to_path_buf()), e))?;

        let mut notes: Vec<String> = symlinks
            .iter()
            .map(|p| {
                format!(
                    "skipped `{}`: powercut copies regular files only, so a store that \
                     keeps state behind a symlink is not fully captured",
                    p.display()
                )
            })
            .collect();

        let index = SyncIndex::build(witness);
        let mut manifest = Vec::new();

        for relative in &files {
            let live = store.join(relative);
            let live_len = std::fs::metadata(&live).map(|m| m.len()).unwrap_or(0);
            let key = canonical(&live);

            let (warrant, source) = match tier {
                Tier::ProcessCrash => (Warrant::OnFilesystem, Source::everything()),
                Tier::PowerCut => match index.plan_for(&key) {
                    Some(plan) => plan,
                    None => continue,
                },
            };

            if source.from.is_none() && !source.exact {
                notes.push(format!(
                    "`{}` was certified durable but no content snapshot was taken, so tier 2 \
                     used its LIVE bytes truncated to the durable length. That is optimistic: \
                     an in-place overwrite made after the sync is credited as durable when it \
                     is not. Route the sync through `SyncJournal` to get an exact capture.",
                    relative.display()
                ));
            }

            let from = source.from.as_deref().unwrap_or(live.as_path());
            let dst = dir.path().join(relative);
            let bytes = scratch::copy_file(from, &dst, source.limit)
                .map_err(|e| Error::io(Some(from.to_path_buf()), e))?;
            manifest.push(CapturedFile {
                path: relative.clone(),
                bytes,
                warrant,
                truncated_from: (bytes < live_len).then_some(live_len),
            });
        }

        if tier == Tier::PowerCut {
            notes.extend(index.integrity_notes(store, &files));
        }

        Ok(Self {
            dir,
            tier,
            manifest,
            notes,
        })
    }
}

/// Where a captured file's bytes come from.
struct Source {
    /// `None` means "the live file in the store".
    from: Option<PathBuf>,
    /// Truncate the copy to this many bytes.
    limit: Option<u64>,
    /// `true` when these bytes are exactly what was certified, rather than a
    /// truncation of whatever the live file holds now.
    exact: bool,
}

impl Source {
    /// The live file, truncated to its certified length: the optimistic
    /// fallback used when no content snapshot exists.
    fn live(limit: Option<u64>) -> Self {
        Self {
            from: None,
            limit,
            exact: false,
        }
    }

    /// The live file in full — tier 1, where every byte survives by definition.
    fn everything() -> Self {
        Self {
            from: None,
            limit: None,
            exact: true,
        }
    }
}

/// The derived durable set, keyed by canonical path.
struct SyncIndex {
    /// Last file/data sync per path.
    synced: BTreeMap<PathBuf, SyncRecord>,
    /// Directories that were fsynced at some point.
    synced_dirs: Vec<PathBuf>,
    assumed: BTreeMap<PathBuf, AssumedDurable>,
}

impl SyncIndex {
    fn build<W: DurabilityWitness>(witness: &W) -> Self {
        let mut synced: BTreeMap<PathBuf, SyncRecord> = BTreeMap::new();
        let mut synced_dirs = Vec::new();
        for record in witness.synced() {
            if record.kind == SyncKind::Directory {
                synced_dirs.push(record.path.clone());
                continue;
            }
            synced
                .entry(record.path.clone())
                .and_modify(|existing| {
                    if record.at > existing.at {
                        *existing = record.clone();
                    }
                })
                .or_insert(record);
        }
        let assumed = witness
            .assumed()
            .into_iter()
            .map(|a| (a.path.clone(), a))
            .collect();
        Self {
            synced,
            synced_dirs,
            assumed,
        }
    }

    /// What to copy for `key`, and from where, or `None` if nothing certified
    /// this path and it does not survive a power cut.
    fn plan_for(&self, key: &Path) -> Option<(Warrant, Source)> {
        if let Some(record) = self.synced.get(key) {
            let source = match record.snapshot.as_ref().filter(|p| p.is_file()) {
                // Exact: the bytes the sync actually certified.
                Some(snapshot) => Source {
                    from: Some(snapshot.clone()),
                    limit: None,
                    exact: true,
                },
                // Optimistic fallback, reported as a note by the caller.
                None => Source::live(record.durable_len),
            };
            return Some((Warrant::Synced(record.clone()), source));
        }
        self.assumed.get(key).map(|a| {
            (
                Warrant::Assumed {
                    reason: a.reason.clone(),
                },
                Source {
                    from: None,
                    limit: None,
                    exact: true,
                },
            )
        })
    }

    /// The observations that turn a confusing red into an explained one.
    fn integrity_notes(&self, store: &Path, present: &[PathBuf]) -> Vec<String> {
        let live: Vec<PathBuf> = present.iter().map(|p| canonical(&store.join(p))).collect();
        let mut notes = Vec::new();

        for (path, record) in &self.synced {
            if !live.contains(path) {
                notes.push(format!(
                    "`{}` was fsynced at barrier {} but is GONE from the store at capture \
                     time. Something unlinked or replaced a file that had already been \
                     certified durable — this is the signature of a checkpoint, a \
                     compaction, or a second connection closing.",
                    display_name(path),
                    record.at
                ));
                continue;
            }
            let live_len = std::fs::metadata(path).map(|m| m.len()).unwrap_or(0);
            if let Some(durable_len) = record.durable_len {
                if live_len < durable_len {
                    notes.push(format!(
                        "`{}` was fsynced at length {durable_len} but is only {live_len} B now. \
                         Durable bytes were REMOVED after being certified; if the data they \
                         held moved into another file, that file needs its own fsync or the \
                         data is no longer durable anywhere.",
                        display_name(path)
                    ));
                }
            }
            if let Some(snapshot) = record.snapshot.as_ref() {
                let certified = std::fs::read(snapshot).ok();
                let now = std::fs::read(path).ok();
                if certified.is_some() && now.is_some() && certified != now {
                    notes.push(format!(
                        "`{}` was REWRITTEN IN PLACE after being certified durable, and the \
                         rewrite was never synced. Tier 2 therefore captured the bytes as of \
                         the last fsync, not the bytes on disk now. A checkpoint, a compaction \
                         or a second handle closing does exactly this.",
                        display_name(path)
                    ));
                }
            }
            if let Some(parent) = path.parent() {
                if !self.synced_dirs.iter().any(|d| d == parent) {
                    notes.push(format!(
                        "`{}` was fsynced but its parent directory never was. On Unix a \
                         newly created file's NAME is only durable after the directory is \
                         fsynced, so this file may be absent after a real power cut even \
                         though its bytes were flushed.",
                        display_name(path)
                    ));
                }
            }
        }
        notes
    }
}

fn display_name(path: &Path) -> String {
    path.file_name()
        .map(|n| n.to_string_lossy().into_owned())
        .unwrap_or_else(|| path.display().to_string())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::witness::SyncJournal;

    #[test]
    fn power_cut_keeps_only_the_synced_prefix() {
        let store = ScratchDir::new("prefix").unwrap();
        let path = store.path().join("log");
        std::fs::write(&path, b"durable").unwrap();

        let journal = SyncJournal::complete();
        journal.sync_path(&path).unwrap();
        // The tail is written after the sync: never certified, never durable.
        std::fs::write(&path, b"durable+volatile").unwrap();

        let crash = CrashState::capture(store.path(), Tier::PowerCut, &journal).unwrap();
        assert_eq!(std::fs::read(crash.dir().join("log")).unwrap(), b"durable");
        assert_eq!(crash.manifest()[0].truncated_from, Some(16));
    }

    #[test]
    fn process_crash_keeps_everything() {
        let store = ScratchDir::new("everything").unwrap();
        std::fs::write(store.path().join("never-synced"), b"x").unwrap();
        let journal = SyncJournal::complete();
        let crash = CrashState::capture(store.path(), Tier::ProcessCrash, &journal).unwrap();
        assert_eq!(crash.manifest().len(), 1);
    }

    #[test]
    fn an_unsynced_in_place_rewrite_is_not_credited() {
        let store = ScratchDir::new("rewrite").unwrap();
        let path = store.path().join("db");
        std::fs::write(&path, b"AAAA").unwrap();

        let journal = SyncJournal::complete();
        journal.sync_path(&path).unwrap();
        // Same length, different bytes, never synced again: a checkpoint.
        std::fs::write(&path, b"BBBB").unwrap();

        let crash = CrashState::capture(store.path(), Tier::PowerCut, &journal).unwrap();
        assert_eq!(
            std::fs::read(crash.dir().join("db")).unwrap(),
            b"AAAA",
            "tier 2 must capture the bytes the fsync certified, not the bytes on disk now"
        );
        assert!(crash
            .notes()
            .iter()
            .any(|n| n.contains("REWRITTEN IN PLACE")));
    }

    #[test]
    fn a_vanished_durable_file_is_reported() {
        let store = ScratchDir::new("vanished").unwrap();
        let path = store.path().join("wal");
        std::fs::write(&path, b"acked").unwrap();
        let journal = SyncJournal::complete();
        journal.sync_path(&path).unwrap();
        std::fs::remove_file(&path).unwrap();

        let crash = CrashState::capture(store.path(), Tier::PowerCut, &journal).unwrap();
        assert!(crash.notes().iter().any(|n| n.contains("GONE")));
    }
}