haematite-cli 0.21.3

//! `haematite history-check`: a read-only probe of legacy history files against
//! their service rows (docs/design/HAEM-HISTORY-ROW-PROBE-2026-10-07.md).
//!
//! References are read with `read_refs_dir` and rows with
//! `ReadOnlyDatabase`, never through `Storage::open`, which moves a `KeepAll`
//! store to reclaim and rewrites format-1 records. A store whose writer lock
//! is held is refused: the probe is for copies. Each file is judged by
//! `history::reconcile`, the rule the importer's history phase uses.

use std::fmt::{self, Write as _};
use std::path::{Path, PathBuf};
use std::process::ExitCode;

use haematite::branch::history::{self, Difference, Reconciled, Row, SERVICE_BRANCH, ShardRoot};
use haematite::branch::{KeptHead, RefRecord};
use haematite::db::vacuum::{CANONICAL_REFS_DIR, read_refs_dir};
use haematite::db::{DataDirLock, ReadOnlyDatabase};

const USAGE: &str = "usage: haematite history-check <data_dir> [--json] [--show N]";
const DEFAULT_SHOW: usize = 50;

/// Why the probe ran no comparison.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Refusal {
    pub code: &'static str,
    pub message: String,
}

impl Refusal {
    fn new(code: &'static str, message: impl Into<String>) -> Self {
        Self {
            code,
            message: message.into(),
        }
    }
}

/// What one history file was found to be against its row.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Finding {
    Reconciled(Reconciled),
    RowMissing,
    NameCorrupt,
    RowCorrupt(String),
}

impl Finding {
    /// The name the probe prints for this finding.
    #[must_use]
    pub const fn kind(&self) -> &'static str {
        match self {
            Self::Reconciled(Reconciled::Equal) => "equal",
            Self::Reconciled(Reconciled::ShardSetDiffers) => "shard_set_differs",
            Self::Reconciled(Reconciled::RowIsAKeptHead { .. }) => "row_is_a_kept_head",
            Self::Reconciled(Reconciled::RootsDiffer(_)) => "roots_differ",
            Self::RowMissing => "row_missing",
            Self::NameCorrupt => "name_corrupt",
            Self::RowCorrupt(_) => "row_corrupt",
        }
    }

    /// The fields printed after the name, in order.
    #[must_use]
    pub fn fields(&self) -> Vec<(&'static str, String)> {
        let shard = |difference: &Difference| {
            vec![
                ("shard", difference.shard_id.to_string()),
                ("file_head", difference.file_head.to_string()),
                ("row_root", difference.row_root.to_string()),
            ]
        };
        match self {
            Self::Reconciled(Reconciled::RootsDiffer(difference)) => shard(difference),
            Self::Reconciled(Reconciled::RowIsAKeptHead {
                difference,
                kept_index,
            }) => {
                let mut fields = shard(difference);
                fields.push(("kept_index", kept_index.to_string()));
                fields
            }
            Self::RowCorrupt(error) => vec![("error", error.clone())],
            Self::Reconciled(Reconciled::Equal | Reconciled::ShardSetDiffers)
            | Self::RowMissing
            | Self::NameCorrupt => Vec::new(),
        }
    }
}

/// One history file that is not equal to its row.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Mismatch {
    pub name: String,
    pub finding: Finding,
}

/// How many history files were found to be each kind.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct Counts {
    pub files: usize,
    pub equal: usize,
    pub row_missing: usize,
    pub roots_differ: usize,
    pub shard_set_differs: usize,
    pub row_is_a_kept_head: usize,
    pub name_corrupt: usize,
    pub row_corrupt: usize,
}

impl Counts {
    /// Every count, named as printed, in printed order.
    #[must_use]
    pub const fn fields(&self) -> [(&'static str, usize); 8] {
        [
            ("files", self.files),
            ("equal", self.equal),
            ("row_missing", self.row_missing),
            ("roots_differ", self.roots_differ),
            ("shard_set_differs", self.shard_set_differs),
            ("row_is_a_kept_head", self.row_is_a_kept_head),
            ("name_corrupt", self.name_corrupt),
            ("row_corrupt", self.row_corrupt),
        ]
    }
}

/// The probe's whole answer for one store.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Report {
    pub store: PathBuf,
    pub service_heads: Vec<ShardRoot>,
    pub counts: Counts,
    pub mismatches: Vec<Mismatch>,
}

impl Report {
    fn note(&mut self, name: String, finding: Finding) {
        self.counts.files += 1;
        let count = match &finding {
            Finding::Reconciled(Reconciled::Equal) => {
                self.counts.equal += 1;
                return;
            }
            Finding::Reconciled(Reconciled::ShardSetDiffers) => &mut self.counts.shard_set_differs,
            Finding::Reconciled(Reconciled::RowIsAKeptHead { .. }) => {
                &mut self.counts.row_is_a_kept_head
            }
            Finding::Reconciled(Reconciled::RootsDiffer(_)) => &mut self.counts.roots_differ,
            Finding::RowMissing => &mut self.counts.row_missing,
            Finding::NameCorrupt => &mut self.counts.name_corrupt,
            Finding::RowCorrupt(_) => &mut self.counts.row_corrupt,
        };
        *count += 1;
        self.mismatches.push(Mismatch { name, finding });
    }

    fn service_heads_text(&self) -> String {
        self.service_heads
            .iter()
            .map(|shard| format!("{}:{}", shard.shard_id, shard.root))
            .collect::<Vec<_>>()
            .join(",")
    }

    /// The summary line, at most `show` mismatch lines, and how many more
    /// there are when any are not shown.
    ///
    /// # Errors
    ///
    /// Writing into the text fails.
    pub fn text(&self, show: usize) -> Result<String, fmt::Error> {
        let mut text = String::new();
        write!(
            text,
            "history_check store={} service_heads={}",
            self.store.display(),
            self.service_heads_text()
        )?;
        for (name, count) in self.counts.fields() {
            write!(text, " {name}={count}")?;
        }
        writeln!(text)?;
        for mismatch in self.mismatches.iter().take(show) {
            write!(
                text,
                "mismatch kind={} name={}",
                mismatch.finding.kind(),
                mismatch.name
            )?;
            for (name, value) in mismatch.finding.fields() {
                write!(text, " {name}={value}")?;
            }
            writeln!(text)?;
        }
        let hidden = self.mismatches.len().saturating_sub(show);
        if hidden > 0 {
            writeln!(
                text,
                "\u{2026} and {hidden} more mismatches not shown (use --show)"
            )?;
        }
        Ok(text)
    }

    /// The same fields as [`Self::text`], as one JSON object.
    #[must_use]
    pub fn json(&self, show: usize) -> serde_json::Value {
        let mut value = serde_json::json!({
            "store": self.store.display().to_string(),
            "service_heads": self.service_heads_text(),
        });
        for (name, count) in self.counts.fields() {
            value[name] = count.into();
        }
        value["mismatches"] = self
            .mismatches
            .iter()
            .take(show)
            .map(|mismatch| {
                let mut entry = serde_json::json!({
                    "kind": mismatch.finding.kind(),
                    "name": mismatch.name,
                });
                for (name, field) in mismatch.finding.fields() {
                    entry[name] = field.into();
                }
                entry
            })
            .collect();
        value["not_shown"] = self.mismatches.len().saturating_sub(show).into();
        value
    }

    /// Whether every count but `equal` is zero.
    #[must_use]
    pub const fn all_equal(&self) -> bool {
        self.counts.files == self.counts.equal
    }
}

struct Options {
    data_dir: PathBuf,
    json: bool,
    show: usize,
}

/// Runs `haematite history-check`: exit 0 when every history file equals its
/// row, 1 when any does not, 2 when the probe refused.
pub fn run(args: impl Iterator<Item = String>) -> ExitCode {
    let outcome = parse(args).and_then(|options| {
        let report = check(&options.data_dir)?;
        let rendered = if options.json {
            format!("{:#}\n", report.json(options.show))
        } else {
            report
                .text(options.show)
                .map_err(|error| Refusal::new("history_check_render", error.to_string()))?
        };
        Ok((report.all_equal(), rendered))
    });
    match outcome {
        Ok((all_equal, rendered)) => {
            print!("{rendered}");
            if all_equal {
                ExitCode::SUCCESS
            } else {
                ExitCode::FAILURE
            }
        }
        Err(refusal) => {
            eprintln!("refused: {}: {}", refusal.code, refusal.message);
            ExitCode::from(2)
        }
    }
}

fn parse(mut args: impl Iterator<Item = String>) -> Result<Options, Refusal> {
    let usage =
        |message: String| Refusal::new("history_check_usage", format!("{message}\n{USAGE}"));
    let mut data_dir = None;
    let mut json = false;
    let mut show = DEFAULT_SHOW;
    while let Some(argument) = args.next() {
        match argument.as_str() {
            "--json" => json = true,
            "--show" => {
                let value = args
                    .next()
                    .ok_or_else(|| usage("--show needs a count".to_owned()))?;
                show = value
                    .parse()
                    .map_err(|_| usage(format!("--show {value:?} is not a count")))?;
            }
            flag if flag.starts_with("--") => {
                return Err(usage(format!("unknown flag {flag:?}")));
            }
            path if data_dir.is_none() => data_dir = Some(PathBuf::from(path)),
            other => return Err(usage(format!("unexpected argument {other:?}"))),
        }
    }
    Ok(Options {
        data_dir: data_dir.ok_or_else(|| usage("a data directory is needed".to_owned()))?,
        json,
        show,
    })
}

/// Compares every legacy history file in `data_dir` with its service row,
/// writing nothing.
///
/// # Errors
///
/// A live writer holds the store, or its references, service record or
/// rows cannot be read.
pub fn check(data_dir: &Path) -> Result<Report, Refusal> {
    match DataDirLock::is_held(data_dir) {
        Ok(false) => {}
        Ok(true) => {
            return Err(Refusal::new(
                "history_check_store_in_use",
                format!(
                    "{} has a live writer; run the check on a copy",
                    data_dir.display()
                ),
            ));
        }
        Err(error) => {
            return Err(Refusal::new(
                "history_check_lock_unreadable",
                error.to_string(),
            ));
        }
    }
    let scan = read_refs_dir(&data_dir.join(CANONICAL_REFS_DIR))
        .map_err(|error| Refusal::new("history_check_refs_unreadable", error.to_string()))?;
    let service = scan
        .records
        .iter()
        .find(|record| record.name == SERVICE_BRANCH)
        .ok_or_else(|| Refusal::new("history_check_not_served", "no service record"))?;
    let reader = ReadOnlyDatabase::open(data_dir)
        .map_err(|error| Refusal::new("history_check_store_unreadable", error.to_string()))?;
    let mut files = scan
        .records
        .iter()
        .filter(|record| record.name.starts_with(history::FILE_PREFIX))
        .collect::<Vec<_>>();
    files.sort_by(|left, right| left.name.cmp(&right.name));
    let mut report = Report {
        store: data_dir.to_path_buf(),
        service_heads: heads(service),
        counts: Counts::default(),
        mismatches: Vec::new(),
    };
    for record in files {
        let kept = scan
            .kept
            .get(&record.name)
            .map(Vec::as_slice)
            .unwrap_or_default();
        let finding = judge(&reader, &report.service_heads, record, kept)?;
        report.note(record.name.clone(), finding);
    }
    Ok(report)
}

fn judge(
    reader: &ReadOnlyDatabase,
    service: &[ShardRoot],
    record: &RefRecord,
    kept: &[KeptHead],
) -> Result<Finding, Refusal> {
    let Some(key) = history::file_key(&record.name) else {
        return Ok(Finding::NameCorrupt);
    };
    let encoded = key.encode();
    let shard_id = reader.shard_for(&encoded);
    let root = service
        .iter()
        .find(|shard| shard.shard_id == shard_id)
        .ok_or_else(|| {
            Refusal::new(
                "history_check_service_shard_missing",
                format!("the service record has no root for shard {shard_id}"),
            )
        })?
        .root;
    let value = reader.get_at(shard_id, root, &encoded).map_err(|error| {
        Refusal::new(
            "history_check_row_unreadable",
            format!("{}: {error}", record.name),
        )
    })?;
    let Some(value) = value else {
        return Ok(Finding::RowMissing);
    };
    let row = match Row::decode(&encoded, &value) {
        Ok(row) => row,
        Err(error) => return Ok(Finding::RowCorrupt(error.to_string())),
    };
    let versions = kept
        .iter()
        .map(|version| {
            record
                .shards
                .iter()
                .zip(&version.heads)
                .map(|(shard, root)| ShardRoot {
                    shard_id: shard.shard_id,
                    root: *root,
                })
                .collect::<Vec<_>>()
        })
        .collect::<Vec<_>>();
    Ok(Finding::Reconciled(history::reconcile(
        &row,
        &heads(record),
        &versions,
    )))
}

fn heads(record: &RefRecord) -> Vec<ShardRoot> {
    record
        .shards
        .iter()
        .map(|shard| ShardRoot {
            shard_id: shard.shard_id,
            root: shard.head,
        })
        .collect()
}