oxy fsck: check a repository's integrity 387335ff7f4d

//! Repository integrity check (spec §21.1): `oxy fsck`.

//!

//! Reads everything and changes nothing. Four passes:

//!

//! 1. every stored object: header, length, hash, and that its body decodes

//!    as its type says;

//! 2. the operation log: every operation reachable from the head is

//!    present and decodes, and clocks never run backwards;

//! 3. references: the current view's bookmarks, tags, remote bookmarks,

//!    workspaces and policy, then the dependency closure of everything

//!    they name, with each change's author identity, blobs and chunks,

//!    and each change's signatures or seals;

//! 4. derived data: every bookmark's state validates, every checkpoint

//!    restores exactly the state it names, and reference and tag

//!    attestations carry good signatures.

//!

//! Missing blobs are warnings, not errors, in a partial clone

//! (`filter.max-blob` or a sparse filter), and changes beyond a shallow

//! boundary are expected to be absent.

use super::*;

use std::collections::{HashSet, VecDeque};

#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]

pub enum Severity {

    /// Worth knowing; the repository is still sound.

    Warning,

    /// Data is missing, corrupt or fails verification.

    Error,

}

#[derive(Clone, Debug, PartialEq, Eq)]

pub struct Problem {

    pub severity: Severity,

    /// The object the problem is about, when there is one.

    pub object: Option<ObjectRef>,

    pub what: String,

}

#[derive(Clone, Debug, Default, PartialEq, Eq)]

pub struct FsckReport {

    pub objects: usize,

    pub ops: usize,

    /// Changes checked, working-copy changes included.

    pub changes: usize,

    /// Committed changes authenticated by their own signature or by a seal.

    pub signed: usize,

    pub unsigned: usize,

    pub checkpoints: usize,

    pub attestations: usize,

    pub problems: Vec<Problem>,

}

impl FsckReport {

    pub fn errors(&self) -> usize {

        self.problems

            .iter()

            .filter(|p| p.severity == Severity::Error)

            .count()

    }

    pub fn warnings(&self) -> usize {

        self.problems

            .iter()

            .filter(|p| p.severity == Severity::Warning)

            .count()

    }

    fn error(&mut self, object: Option<ObjectRef>, what: impl Into<String>) {

        self.problems.push(Problem {

            severity: Severity::Error,

            object,

            what: what.into(),

        });

    }

    fn warn(&mut self, object: Option<ObjectRef>, what: impl Into<String>) {

        self.problems.push(Problem {

            severity: Severity::Warning,

            object,

            what: what.into(),

        });

    }

}

/// Whether `body` decodes as an object of type `ty`.

fn decodes(ty: ObjectType, body: &[u8]) -> std::result::Result<(), String> {

    let r = match ty {

        ObjectType::Chunk => Ok(()),

        ObjectType::Blob => Blob::decode(body).map(|_| ()),

        ObjectType::Change => Change::decode(body).map(|_| ()),

        ObjectType::Checkpoint => crate::objects::Checkpoint::decode(body).map(|_| ()),

        ObjectType::Op => match Op::decode(body) {

            Ok(_) => Ok(()),

            Err(e) => Op::decode_format1(body).map(|_| ()).map_err(|_| e),

        },

        ObjectType::Attest => Attest::decode(body).map(|_| ()),

        ObjectType::Identity => Identity::decode(body).map(|_| ()),

        ObjectType::Policy => crate::policy::Policy::decode(body).map(|_| ()),

        ObjectType::State => crate::objects::StateObject::decode(body).map(|_| ()),

    };

    r.map_err(|e| e.to_string())

}

impl Repo {

    /// Check the whole repository; see the module documentation.

    pub fn fsck(&self) -> Result<FsckReport> {

        let mut rep = FsckReport::default();

        let bad = self.fsck_objects(&mut rep)?;

        self.fsck_ops(&mut rep, &bad);

        self.fsck_references(&mut rep, &bad)?;

        self.fsck_derived(&mut rep)?;

        rep.problems.sort_by_key(|p| std::cmp::Reverse(p.severity));

        Ok(rep)

    }

    /// Pass 1. Returns the objects that failed, so later passes report

    /// them once rather than at every reference.

    fn fsck_objects(&self, rep: &mut FsckReport) -> Result<HashSet<ObjectRef>> {

        let mut bad = HashSet::new();

        for r in self.store.list()? {

            rep.objects += 1;

            match self.store.get(&r) {

                Ok((ty, body)) => {

                    if let Err(e) = decodes(ty, &body) {

                        rep.error(

                            Some(r),

                            format!("{} object does not decode: {e}", ty.name()),

                        );

                        bad.insert(r);

                    }

                }

                Err(e) => {

                    rep.error(Some(r), format!("unreadable object: {e}"));

                    bad.insert(r);

                }

            }

        }

        Ok(bad)

    }

    /// Pass 2.

    fn fsck_ops(&self, rep: &mut FsckReport, bad: &HashSet<ObjectRef>) {

        let head = match fs::read_to_string(self.oxy.join("ops").join("head")) {

            Ok(h) => h,

            Err(e) => {

                rep.error(None, format!("cannot read .oxy/ops/head: {e}"));

                return;

            }

        };

        let Some(head) = ObjectRef::from_hex(head.trim()) else {

            rep.error(None, ".oxy/ops/head does not hold an object reference");

            return;

        };

        let mut seen = HashSet::new();

        let mut queue: VecDeque<(ObjectRef, Option<(ObjectRef, u64)>)> =

            VecDeque::from([(head, None)]);

        while let Some((r, child)) = queue.pop_front() {

            if !seen.insert(r) {

                continue;

            }

            if bad.contains(&r) {

                continue;

            }

            let body = match self.store.get_typed(&r, ObjectType::Op) {

                Ok(b) => b,

                Err(StoreError::NotFound(_)) => {

                    let by = child

                        .map(|(c, _)| format!(", the parent of operation {}", c.short(12)))

                        .unwrap_or_default();

                    rep.error(

                        Some(r),

                        format!("operation missing{by}; the log is cut here"),

                    );

                    continue;

                }

                Err(e) => {

                    rep.error(Some(r), format!("operation unreadable: {e}"));

                    continue;

                }

            };

            rep.ops += 1;

            let (parents, wall_ms) = match Op::decode(&body) {

                Ok(op) => (op.parents, op.hlc.wall_ms),

                Err(_) => match Op::decode_format1(&body) {

                    Ok(op) => (op.parents, op.hlc.wall_ms),

                    Err(e) => {

                        rep.error(Some(r), format!("operation does not decode: {e}"));

                        continue;

                    }

                },

            };

            if let Some((c, child_ms)) = child {

                if wall_ms > child_ms {

                    rep.warn(

                        Some(c),

                        format!(

                            "operation is timed before its parent {} (a clock was wrong)",

                            r.short(12)

                        ),

                    );

                }

            }

            queue.extend(parents.into_iter().map(|p| (p, Some((r, wall_ms)))));

        }

    }

    /// Pass 3.

    fn fsck_references(&self, rep: &mut FsckReport, bad: &HashSet<ObjectRef>) -> Result<()> {

        let view = match self.view() {

            Ok(v) => v,

            Err(e) => {

                rep.error(None, format!("cannot read the current view: {e}"));

                return Ok(());

            }

        };

        let boundary = self.boundary();

        let partial = self.config.max_blob.is_some() || !self.config.sparse_paths.is_empty();

        let mut roots: Vec<(ChangeHash, String)> = Vec::new();

        let mut states: Vec<(ObjectRef, String)> = Vec::new();

        states.extend(

            view.bookmarks

                .iter()

                .map(|(n, s)| (*s, format!("bookmark {n}"))),

        );

        states.extend(view.tags.iter().map(|(n, s)| (*s, format!("tag {n}"))));

        for (remote, bms) in &view.remotes {

            states.extend(bms.iter().map(|(n, s)| (*s, format!("{remote}/{n}"))));

        }

        for (st, what) in states {

            match self.store.get_state(&st) {

                Ok(s) => roots.extend(s.heads.into_iter().map(|h| (h, what.clone()))),

                Err(e) if !bad.contains(&st) => {

                    rep.error(Some(st), format!("state of {what} unusable: {e}"))

                }

                Err(_) => {}

            }

        }

        for (ws, h) in &view.workspaces {

            roots.push((*h, format!("workspace {}", ws.to_hex())));

        }

        if let Some(p) = view.policy {

            match self.store.get_typed(&p, ObjectType::Policy) {

                Ok(bytes) => {

                    if let Ok(policy) = crate::policy::Policy::decode(&bytes) {

                        for id in &policy.identities {

                            if !self.store.has(&id.hash()) {

                                rep.warn(

                                    Some(id.hash()),

                                    "an identity the policy names is not stored",

                                );

                            }

                        }

                    }

                }

                Err(e) if !bad.contains(&p) => {

                    rep.error(Some(p), format!("the policy is unusable: {e}"))

                }

                Err(_) => {}

            }

        }

        // Working-copy changes are never signed; they are checked but not tallied.

        let working: HashSet<ChangeHash> = view.workspaces.values().copied().collect();

        let mut seen: HashSet<ChangeHash> = HashSet::new();

        let mut blobs_seen: HashSet<ObjectRef> = HashSet::new();

        let mut stack = roots;

        while let Some((h, needed_by)) = stack.pop() {

            if !seen.insert(h) {

                continue;

            }

            let c = match self.store.get_change(&h) {

                Ok(c) => c,

                Err(StoreError::NotFound(_)) if boundary.contains(&h) => continue,

                Err(StoreError::NotFound(_)) => {

                    rep.error(Some(h), format!("change missing, needed by {needed_by}"));

                    continue;

                }

                Err(e) => {

                    if !bad.contains(&h) {

                        rep.error(Some(h), format!("change unusable: {e}"));

                    }

                    continue;

                }

            };

            rep.changes += 1;

            let label = format!("change {}", h.short(12));

            stack.extend(c.deps.iter().map(|d| (*d, label.clone())));

            if !c.author.is_zero() && !self.store.has(&c.author) {

                rep.warn(

                    Some(h),

                    format!("author identity {} is not stored", c.author.short(12)),

                );

            }

            for e in &c.edits {

                let Edit::Replace { old, new, .. } = e else {

                    continue;

                };

                for b in [old, new].into_iter().flatten() {

                    if !blobs_seen.insert(*b) {

                        continue;

                    }

                    self.fsck_blob(rep, *b, &label, partial, bad);

                }

            }

            if working.contains(&h) {

                continue;

            }

            match self.authenticate_change(&h) {

                Ok(SignatureStatus::Valid { .. }) => rep.signed += 1,

                Ok(SignatureStatus::Unsigned) => rep.unsigned += 1,

                Ok(SignatureStatus::Invalid(why)) => {

                    rep.error(Some(h), format!("bad signature: {why}"))

                }

                Err(e) => rep.error(Some(h), format!("cannot check signatures: {e}")),

            }

        }

        Ok(())

    }

    fn fsck_blob(

        &self,

        rep: &mut FsckReport,

        b: ObjectRef,

        used_by: &str,

        partial: bool,

        bad: &HashSet<ObjectRef>,

    ) {

        match self.store.get_typed(&b, ObjectType::Blob) {

            Ok(bytes) => {

                let Ok(blob) = Blob::decode(&bytes) else {

                    return;

                };

                for ch in blob.chunks {

                    if !self.store.has(&ch) {

                        rep.error(

                            Some(ch),

                            format!("chunk missing from blob {} ({used_by})", b.short(12)),

                        );

                    }

                }

            }

            Err(StoreError::NotFound(_)) if partial => rep.warn(

                Some(b),

                format!("blob not fetched ({used_by}); expected in a partial clone"),

            ),

            Err(StoreError::NotFound(_)) => rep.error(Some(b), format!("blob missing ({used_by})")),

            Err(e) if !bad.contains(&b) => {

                rep.error(Some(b), format!("blob unusable ({used_by}): {e}"))

            }

            Err(_) => {}

        }

    }

    /// Pass 4.

    fn fsck_derived(&self, rep: &mut FsckReport) -> Result<()> {

        let Ok(view) = self.view() else { return Ok(()) };

        let shallow = self.is_shallow();

        if !shallow {

            for (name, st) in &view.bookmarks {

                let Ok(s) = self.store.get_state(st) else {

                    continue;

                };

                if let Err(e) = self.load_state_full(&s.heads) {

                    rep.error(

                        Some(*st),

                        format!("bookmark {name} does not form a valid state: {e}"),

                    );

                }

            }

        }

        for r in self.checkpoint_refs()? {

            rep.checkpoints += 1;

            let cp = match self.store.get_typed(&r, ObjectType::Checkpoint) {

                Ok(b) => match crate::objects::Checkpoint::decode(&b) {

                    Ok(cp) => cp,

                    Err(_) => continue,

                },

                Err(StoreError::NotFound(_)) => {

                    rep.warn(

                        Some(r),

                        "checkpoint pointer names a missing checkpoint; it will be rewritten",

                    );

                    continue;

                }

                Err(_) => continue,

            };

            let payload = match self.read_blob(&cp.payload) {

                Ok(p) => p,

                Err(e) => {

                    rep.error(Some(r), format!("checkpoint payload unreadable: {e}"));

                    continue;

                }

            };

            if State::from_checkpoint_payload(&payload).is_err() {

                rep.error(Some(r), "checkpoint payload does not decode");

                continue;

            }

            if !shallow {

                match self.load_state_full(&cp.heads) {

                    Ok(full) if full.checkpoint_payload() != payload => {

                        rep.error(Some(r), "checkpoint disagrees with the changes it names")

                    }

                    Ok(_) => {}

                    Err(e) => rep.error(

                        Some(r),

                        format!("checkpoint names changes that do not form a state: {e}"),

                    ),

                }

            }

        }

        let mut attests = Vec::new();

        for name in view.bookmarks.keys() {

            if let Ok(Some((r, _))) = self.latest_ref_attest(name) {

                attests.push((r, format!("bookmark {name}")));

            }

        }

        for name in view.tags.keys() {

            if let Ok(Some((r, _))) = self.tag_attest(name) {

                attests.push((r, format!("tag {name}")));

            }

        }

        for (r, what) in attests {

            rep.attestations += 1;

            let sigs = self.store.signatures(&r)?;

            if sigs.is_empty() {

                rep.warn(Some(r), format!("attestation for {what} is unsigned"));

            }

            for sig in sigs {

                match self.store.get_identity(&sig.key) {

                    Ok(id) => {

                        if let Err(e) = verify_signature(&id, &sig, Domain::Attest, &r) {

                            rep.error(

                                Some(r),

                                format!("attestation for {what} has a bad signature: {e}"),

                            );

                        }

                    }

                    Err(_) => rep.warn(

                        Some(r),

                        format!(

                            "attestation for {what} is signed by {}, whose identity is not stored",

                            sig.key.short(12)

                        ),

                    ),

                }

            }

        }

        Ok(())

    }

}

mod fsck;

pub use fsck::{FsckReport, Problem, Severity};

#[test]

fn fsck_passes_a_sound_repository_and_names_each_fault() {

    hermetic();

    let dir = tempfile::tempdir().unwrap();

    let root = dir.path();

    let origin_path = root.join("origin");

    let origin = Repo::init(&origin_path).unwrap();

    let mut a = Repo::clone(&origin_path.to_string_lossy(), &root.join("a"))

        .unwrap()

        .with_home(root.join("home-a"));

    a.create_identity("Ada", "[email protected]", IdentityScope::Repo)

        .unwrap();

    write(&root.join("a"), "f.txt", b"one\n");

    write(&root.join("a"), "bin.dat", b"\0\x01\x02binary");

    let h1 = a.commit("first", &[]).unwrap();

    write(&root.join("a"), "f.txt", b"one\ntwo\n");

    let h2 = a.commit("second", &[]).unwrap();

    a.tag_create("v1", &[h2], "first release").unwrap();

    a.checkpoint().unwrap();

    assert!(a.push("origin", false).unwrap()[0].ok);

    // Sound: signed changes, a seal, a tag, a checkpoint, a ref attestation.

    for repo in [&a, &origin] {

        let rep = repo.fsck().unwrap();

        assert_eq!(rep.problems, vec![], "{rep:?}");

        // The two commits, plus the working copy's change in `a`.

        assert!(rep.changes >= 2 && rep.signed >= 2, "{rep:?}");

        assert!(rep.ops > 1 && rep.objects > 5);

    }

    assert!(origin.fsck().unwrap().attestations >= 1);

    // A flipped byte in a stored object is corruption, named by hash.

    let c1 = a.store().path_for(&h1);

    let pristine = fs::read(&c1).unwrap();

    let mut bytes = pristine.clone();

    *bytes.last_mut().unwrap() ^= 1;

    fs::write(&c1, &bytes).unwrap();

    let rep = a.fsck().unwrap();

    assert!(rep.errors() >= 1);

    assert!(

        rep.problems

            .iter()

            .any(|p| p.object == Some(h1) && p.what.contains("unreadable")),

        "{rep:?}"

    );

    fs::write(&c1, &pristine).unwrap();

    assert_eq!(a.fsck().unwrap().errors(), 0);

    // A missing change is named with what needed it.

    fs::remove_file(&c1).unwrap();

    let rep = a.fsck().unwrap();

    assert!(

        rep.problems.iter().any(|p| p.severity == Severity::Error

            && p.object == Some(h1)

            && p.what.contains("change missing, needed by change")),

        "{rep:?}"

    );

    fs::write(&c1, &pristine).unwrap();

    // A missing blob is an error in a full clone.

    let c2 = a.store().get_change(&h1).unwrap();

    let blob = c2

        .edits

        .iter()

        .find_map(|e| match e {

            Edit::Replace { new: Some(b), .. } => Some(*b),

            _ => None,

        })

        .expect("the binary file is a blob");

    let bp = a.store().path_for(&blob);

    let blob_bytes = fs::read(&bp).unwrap();

    fs::remove_file(&bp).unwrap();

    let rep = a.fsck().unwrap();

    assert!(

        rep.problems

            .iter()

            .any(|p| p.object == Some(blob) && p.what.contains("blob missing")),

        "{rep:?}"

    );

    fs::write(&bp, &blob_bytes).unwrap();

    // A broken op log head is reported, not a crash.

    let head = root.join("a/.oxy/ops/head");

    let saved = fs::read(&head).unwrap();

    fs::write(&head, b"not a hash").unwrap();

    let rep = a.fsck().unwrap();

    assert!(

        rep.problems.iter().any(|p| p.what.contains("ops/head")),

        "{rep:?}"

    );

    fs::write(&head, saved).unwrap();

    assert_eq!(a.fsck().unwrap().problems, vec![]);

}

    pub(crate) fn path_for(&self, r: &ObjectRef) -> PathBuf {

oxy fsck            # check every object, the op log, references, signatures, checkpoints

| **Status** | Draft 0.8h |

### 21.1 Integrity check

`oxy fsck` reads everything and changes nothing. It checks (1) every stored object: header, length, ObjectRef, and that the body decodes as its type; (2) the operation log: every operation reachable from the head through all parents is present and decodes, and no operation's clock is earlier than its parent's (a warning); (3) the current view's bookmarks, tags, remote bookmarks, workspaces and policy, and the dependency closure of every change they name, with each change's author identity, blobs and chunks, and each committed change's signature or seal (§14.4); (4) that every bookmark forms a valid state, that every checkpoint restores exactly the state of the changes it names (§11), and that reference and tag attestations carry valid signatures. A change beyond the shallow boundary (§17) may be absent; a blob absent from a partial clone (`filter.max-blob` or a sparse filter) is a warning, not an error. Each problem names the object it concerns. The exit status is 1 when any error is found and 0 otherwise; `--json` gives the counts and problems.

oxy fsck [--json]                   check integrity (§21.1)

*Changelog.* Draft 0.8h (2026-09-28): `oxy fsck` (§21.1). Draft 0.8g (2026-09-27): delegates of a rule's identities may sign reference attestations (§14.5); local-path pushes bind unbound repositories like HTTP ones. Draft 0.8f (2026-09-27): challenge signatures (§14.3) for hosts. Draft 0.8e (2026-09-27): anonymous changes (zero author) are authenticated by a seal from any key, so imported and pre-identity history can be pushed under a policy (§14.4). Draft 0.8d (2026-09-27): unbound hosted repositories adopt the first pusher's id (§16.1). Draft 0.8c (2026-09-27): bare repositories (`oxy init --bare`, §12.2: no workspace, no working copy; hosting) and the reference server as an embeddable library (§15.1). Draft 0.8b (2026-09-27): the machine interface for editors (§24): `root`, `cat`, `version`, `hunks`, `discard`, and `--json` on the reporting commands.

    /// Check the repository's integrity: objects, operation log, references, signatures, checkpoints

    Fsck {

        #[arg(long)]

        json: bool,

    },

        Cmd::Fsck { json } => workspace::fsck(json),

/// `oxy fsck`: exit 0 when sound (warnings allowed), 1 on any error.

pub fn fsck(as_json: bool) -> Result<i32> {

    use oxy_core::repo::Severity;

    let repo = open()?;

    let sp = ui::spinner("Checking every object…");

    let r = repo.fsck()?;

    sp.clear();

    let code = if r.errors() > 0 { 1 } else { 0 };

    if as_json {

        let problems: Vec<String> = r

            .problems

            .iter()

            .map(|p| {

                json::obj([

                    (

                        "severity",

                        json::s(match p.severity {

                            Severity::Error => "error",

                            Severity::Warning => "warning",

                        }),

                    ),

                    (

                        "object",

                        json::opt(p.object.map(|o| o.to_hex()).as_deref(), json::s),

                    ),

                    ("what", json::s(&p.what)),

                ])

            })

            .collect();

        println!(

            "{}",

            json::obj([

                ("ok", json::b(code == 0)),

                ("objects", json::n(r.objects)),

                ("ops", json::n(r.ops)),

                ("changes", json::n(r.changes)),

                ("signed", json::n(r.signed)),

                ("unsigned", json::n(r.unsigned)),

                ("checkpoints", json::n(r.checkpoints)),

                ("attestations", json::n(r.attestations)),

                ("errors", json::n(r.errors())),

                ("warnings", json::n(r.warnings())),

                ("problems", json::arr(problems)),

            ])

        );

        return Ok(code);

    }

    println!(

        "Checked {}, {}, {} ({} signed or sealed, {} unsigned), {}, {}.",

        plural(r.objects, "object"),

        plural(r.ops, "operation"),

        plural(r.changes, "change"),

        r.signed,

        r.unsigned,

        plural(r.checkpoints, "checkpoint"),

        plural(r.attestations, "attestation"),

    );

    for p in &r.problems {

        let obj = p

            .object

            .map(|o| format!("{} ", ui::hash(format!("#{}", o.short(12)))))

            .unwrap_or_default();

        match p.severity {

            Severity::Error => println!("{} {obj}{}", ui::err(ui::glyph("✖", "x")), p.what),

            Severity::Warning => println!("{} {obj}{}", ui::caution(), p.what),

        }

    }

    if r.problems.is_empty() {

        println!("{}", ui::done("No problems found."));

    } else if code == 0 {

        println!(

            "{}",

            ui::done(format!("Sound, with {}.", plural(r.warnings(), "warning")))

        );

    } else {

        println!(

            "{} {} and {}. Fetch missing objects from a remote with `oxy pull`, or clone afresh and compare.",

            ui::err(ui::glyph("✖", "x")),

            plural(r.errors(), "error"),

            plural(r.warnings(), "warning")

        );

    }

    Ok(code)

}