fsck.rs 71.6%

265 of 370 lines run, 34 of 64 functions · run #420 on main · run missed

//! 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(|_| ()),
        // Structure only: the text blobs are checked with the change graph.
        ObjectType::Change => {
            Change::decode_with(body, &mut |_, _, _| Ok(crate::lines::Lines::default())).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::Receipt => crate::receipt::Receipt::decode(body).map(|_| ()),
        ObjectType::Certificate => crate::cert::Certificate::decode(body).map(|_| ()),
        ObjectType::Grant => crate::cert::Grant::decode(body).map(|_| ()),
        ObjectType::Artifact => crate::artifact::Artifact::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> {
        // The head as it is now, as the operations pass reads it.
        self.forget_head();
        let mut rep = FsckReport::default();
        let bad = {
            let _t = crate::timing::phase("fsck: objects");
            self.fsck_objects(&mut rep)?
        };
        {
            let _t = crate::timing::phase("fsck: operations");
            self.fsck_ops(&mut rep, &bad);
        }
        {
            let _t = crate::timing::phase("fsck: references");
            self.fsck_references(&mut rep, &bad)?;
        }
        {
            let _t = crate::timing::phase("fsck: states and checkpoints");
            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>> {
        use rayon::prelude::*;
        let mut bad = HashSet::new();
        // A chunk at a time, read (and verified) and decoded on every core.
        for refs in self.store.list()?.chunks(8192) {
            let checked: Vec<std::result::Result<(), String>> = self
                .store
                .get_many(refs)
                .into_par_iter()
                .map(|got| match got {
                    Ok((ty, body)) => decodes(ty, &body)
                        .map_err(|e| format!("{} object does not decode: {e}", ty.name())),
                    Err(e) => Err(format!("unreadable object: {e}")),
                })
                .collect();
            for (r, c) in refs.iter().zip(checked) {
                rep.objects += 1;
                if let Err(e) = c {
                    rep.error(Some(*r), e);
                    bad.insert(*r);
                }
            }
        }
        Ok(bad)
    }

    /// Pass 2.
    fn fsck_ops(&self, rep: &mut FsckReport, bad: &HashSet<ObjectRef>) {
        let head = match self.read_head() {
            Ok((h, _)) => h,
            Err(RepoError::Msg(m)) if m.starts_with("corrupt") => {
                rep.error(None, ".oxy/ops/head does not hold an object reference");
                return;
            }
            Err(e) => {
                rep.error(None, format!("cannot read .oxy/ops/head: {e}"));
                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;
            }
            // The skeleton: the text is checked as the blobs it is stored
            // in, not decoded (the objects pass checked every object).
            let (c, texts) = match self.store.get_change_lazy(&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 t in texts {
                if let crate::objects::TextRef::Blob(b, _, _) = t {
                    if blobs_seen.insert(b) {
                        // A change cannot be read without its text: never partial.
                        self.fsck_blob(rep, b, &label, false, bad);
                    }
                }
            }
            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(_) => {}
        }
    }

    /// Whether `heads` form a valid state: its history loads and every file
    /// graph builds. Graphs are built a batch at a time and dropped, so the
    /// check needs the memory of a batch, not of the whole tree.
    fn check_state(&self, heads: &[ChangeHash]) -> Result<()> {
        let mut st = self.load_state_history(heads)?;
        let nodes = st.unbuilt_nodes();
        for batch in nodes.chunks(2048) {
            self.load_graphs_of(&mut st, batch)?;
            st.drop_graphs(batch);
        }
        Ok(())
    }

    /// 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 {
            let mut checked: HashSet<ObjectRef> = HashSet::new();
            for (name, st) in &view.bookmarks {
                if !checked.insert(*st) {
                    continue;
                }
                let Ok(s) = self.store.get_state(st) else {
                    continue;
                };
                if let Err(e) = self.check_state(&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 let Err(e) = self.state_from_checkpoint(&payload) {
                rep.error(Some(r), format!("checkpoint payload does not decode: {e}"));
                continue;
            }
            // The payload must be what the named changes produce, in its own version.
            let version = crate::cbor::decode(&payload).ok().and_then(|v| {
                crate::objects::map(&v, "payload").ok().and_then(|m| {
                    m.get(&0)
                        .and_then(|x| crate::objects::uint(x, "version").ok())
                })
            });
            let expected = |full: &State| {
                if version == Some(crate::state::CHECKPOINT_INDEX_VERSION) {
                    full.checkpoint_index(&full.checkpoint_file_graphs(), Repo::blob_ref)
                } else {
                    full.checkpoint_payload()
                }
            };
            if !shallow {
                match self.load_state_full(&cp.heads) {
                    Ok(full) if expected(&full) != 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(())
    }
}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

1		//! Repository integrity check (spec §21.1): `oxy fsck`.
2		//!
3		//! Reads everything and changes nothing. Four passes:
4		//!
5		//! 1. every stored object: header, length, hash, and that its body decodes
6		//! as its type says;
7		//! 2. the operation log: every operation reachable from the head is
8		//! present and decodes, and clocks never run backwards;
9		//! 3. references: the current view's bookmarks, tags, remote bookmarks,
10		//! workspaces and policy, then the dependency closure of everything
11		//! they name, with each change's author identity, blobs and chunks,
12		//! and each change's signatures or seals;
13		//! 4. derived data: every bookmark's state validates, every checkpoint
14		//! restores exactly the state it names, and reference and tag
15		//! attestations carry good signatures.
16		//!
17		//! Missing blobs are warnings, not errors, in a partial clone
18		//! (`filter.max-blob` or a sparse filter), and changes beyond a shallow
19		//! boundary are expected to be absent.
20
21		use super::*;
22		use std::collections::{HashSet, VecDeque};
23
24		#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
25		pub enum Severity {
26		/// Worth knowing; the repository is still sound.
27		Warning,
28		/// Data is missing, corrupt or fails verification.
29		Error,
30		}
31
32		#[derive(Clone, Debug, PartialEq, Eq)]
33		pub struct Problem {
34		pub severity: Severity,
35		/// The object the problem is about, when there is one.
36		pub object: Option<ObjectRef>,
37		pub what: String,
38		}
39
40		#[derive(Clone, Debug, Default, PartialEq, Eq)]
41		pub struct FsckReport {
42		pub objects: usize,
43		pub ops: usize,
44		/// Changes checked, working-copy changes included.
45		pub changes: usize,
46		/// Committed changes authenticated by their own signature or by a seal.
47		pub signed: usize,
48		pub unsigned: usize,
49		pub checkpoints: usize,
50		pub attestations: usize,
51		pub problems: Vec<Problem>,
52		}
53
54		impl FsckReport {
55	68×	pub fn errors(&self) -> usize {
56	68×	self.problems
57	68×	.iter()
58	68×	.filter(\|p\| p.severity == Severity::Error)
59	68×	.count()
60	68×	}
61
62	0×	pub fn warnings(&self) -> usize {
63	0×	self.problems
64	0×	.iter()
65	0×	.filter(\|p\| p.severity == Severity::Warning)
66	0×	.count()
67	0×	}
68
69	11×	fn error(&mut self, object: Option<ObjectRef>, what: impl Into<String>) {
70	11×	self.problems.push(Problem {
71	11×	severity: Severity::Error,
72	11×	object,
73	11×	what: what.into(),
74	11×	});
75	11×	}
76
77	0×	fn warn(&mut self, object: Option<ObjectRef>, what: impl Into<String>) {
78	0×	self.problems.push(Problem {
79	0×	severity: Severity::Warning,
80	0×	object,
81	0×	what: what.into(),
82	0×	});
83	0×	}
84		}
85
86		/// Whether `body` decodes as an object of type `ty`.
87	1730×	fn decodes(ty: ObjectType, body: &[u8]) -> std::result::Result<(), String> {
88	1730×	let r = match ty {
89	137×	ObjectType::Chunk => Ok(()),
90	86×	ObjectType::Blob => Blob::decode(body).map(\|_\| ()),
91		// Structure only: the text blobs are checked with the change graph.
92		ObjectType::Change => {
93	432×	Change::decode_with(body, &mut \|_, _, _\| Ok(crate::lines::Lines::default())).map(\|_\| ())
94		}
95	12×	ObjectType::Checkpoint => crate::objects::Checkpoint::decode(body).map(\|_\| ()),
96	492×	ObjectType::Op => match Op::decode(body) {
97	492×	Ok(_) => Ok(()),
98	0×	Err(e) => Op::decode_format1(body).map(\|_\| ()).map_err(\|_\| e),
99		},
100	230×	ObjectType::Attest => Attest::decode(body).map(\|_\| ()),
101	70×	ObjectType::Identity => Identity::decode(body).map(\|_\| ()),
102	0×	ObjectType::Policy => crate::policy::Policy::decode(body).map(\|_\| ()),
103	2×	ObjectType::Receipt => crate::receipt::Receipt::decode(body).map(\|_\| ()),
104	1×	ObjectType::Certificate => crate::cert::Certificate::decode(body).map(\|_\| ()),
105	0×	ObjectType::Grant => crate::cert::Grant::decode(body).map(\|_\| ()),
106	1×	ObjectType::Artifact => crate::artifact::Artifact::decode(body).map(\|_\| ()),
107	267×	ObjectType::State => crate::objects::StateObject::decode(body).map(\|_\| ()),
108		};
109	1730×	r.map_err(\|e\| e.to_string())
110	1730×	}
111
112		impl Repo {
113		/// Check the whole repository; see the module documentation.
114	80×	pub fn fsck(&self) -> Result<FsckReport> {
115		// The head as it is now, as the operations pass reads it.
116	80×	self.forget_head();
117	80×	let mut rep = FsckReport::default();
118	80×	let bad = {
119	80×	let _t = crate::timing::phase("fsck: objects");
120	80×	self.fsck_objects(&mut rep)?
121		};
122	80×	{
123	80×	let _t = crate::timing::phase("fsck: operations");
124	80×	self.fsck_ops(&mut rep, &bad);
125	80×	}
126		{
127	80×	let _t = crate::timing::phase("fsck: references");
128	80×	self.fsck_references(&mut rep, &bad)?;
129		}
130		{
131	80×	let _t = crate::timing::phase("fsck: states and checkpoints");
132	80×	self.fsck_derived(&mut rep)?;
133		}
134	80×	rep.problems.sort_by_key(\|p\| std::cmp::Reverse(p.severity));
135	80×	Ok(rep)
136	80×	}
137
138		/// Pass 1. Returns the objects that failed, so later passes report
139		/// them once rather than at every reference.
140	80×	fn fsck_objects(&self, rep: &mut FsckReport) -> Result<HashSet<ObjectRef>> {
141		use rayon::prelude::*;
142	80×	let mut bad = HashSet::new();
143		// A chunk at a time, read (and verified) and decoded on every core.
144	80×	for refs in self.store.list()?.chunks(8192) {
145	80×	let checked: Vec<std::result::Result<(), String>> = self
146	80×	.store
147	80×	.get_many(refs)
148	80×	.into_par_iter()
149	1731×	.map(\|got\| match got {
150	1730×	Ok((ty, body)) => decodes(ty, &body)
151	1730×	.map_err(\|e\| format!("{} object does not decode: {e}", ty.name())),
152	1×	Err(e) => Err(format!("unreadable object: {e}")),
153	1731×	})
154	80×	.collect();
155	1731×	for (r, c) in refs.iter().zip(checked) {
156	1731×	rep.objects += 1;
157	1731×	if let Err(e) = c {
158	1×	rep.error(Some(*r), e);
159	1×	bad.insert(*r);
160	1730×	}
161		}
162		}
163	80×	Ok(bad)
164	80×	}
165
166		/// Pass 2.
167	80×	fn fsck_ops(&self, rep: &mut FsckReport, bad: &HashSet<ObjectRef>) {
168	80×	let head = match self.read_head() {
169	79×	Ok((h, _)) => h,
170	1×	Err(RepoError::Msg(m)) if m.starts_with("corrupt") => {
171	1×	rep.error(None, ".oxy/ops/head does not hold an object reference");
172	1×	return;
173		}
174	0×	Err(e) => {
175	0×	rep.error(None, format!("cannot read .oxy/ops/head: {e}"));
176	0×	return;
177		}
178		};
179	79×	let mut seen = HashSet::new();
180	79×	let mut queue: VecDeque<(ObjectRef, Option<(ObjectRef, u64)>)> =
181	79×	VecDeque::from([(head, None)]);
182	573×	while let Some((r, child)) = queue.pop_front() {
183	494×	if !seen.insert(r) {
184	20×	continue;
185	474×	}
186	474×	if bad.contains(&r) {
187	0×	continue;
188	474×	}
189	474×	let body = match self.store.get_typed(&r, ObjectType::Op) {
190	474×	Ok(b) => b,
191		Err(StoreError::NotFound(_)) => {
192	0×	let by = child
193	0×	.map(\|(c, _)\| format!(", the parent of operation {}", c.short(12)))
194	0×	.unwrap_or_default();
195	0×	rep.error(
196	0×	Some(r),
197	0×	format!("operation missing{by}; the log is cut here"),
198		);
199	0×	continue;
200		}
201	0×	Err(e) => {
202	0×	rep.error(Some(r), format!("operation unreadable: {e}"));
203	0×	continue;
204		}
205		};
206	474×	rep.ops += 1;
207	474×	let (parents, wall_ms) = match Op::decode(&body) {
208	474×	Ok(op) => (op.parents, op.hlc.wall_ms),
209	0×	Err(_) => match Op::decode_format1(&body) {
210	0×	Ok(op) => (op.parents, op.hlc.wall_ms),
211	0×	Err(e) => {
212	0×	rep.error(Some(r), format!("operation does not decode: {e}"));
213	0×	continue;
214		}
215		},
216		};
217	474×	if let Some((c, child_ms)) = child {
218	395×	if wall_ms > child_ms {
219	0×	rep.warn(
220	0×	Some(c),
221	0×	format!(
222	0×	"operation is timed before its parent {} (a clock was wrong)",
223	0×	r.short(12)
224	0×	),
225	0×	);
226	395×	}
227	79×	}
228	474×	queue.extend(parents.into_iter().map(\|p\| (p, Some((r, wall_ms)))));
229		}
230	80×	}
231
232		/// Pass 3.
233	80×	fn fsck_references(&self, rep: &mut FsckReport, bad: &HashSet<ObjectRef>) -> Result<()> {
234	80×	let view = match self.view() {
235	79×	Ok(v) => v,
236	1×	Err(e) => {
237	1×	rep.error(None, format!("cannot read the current view: {e}"));
238	1×	return Ok(());
239		}
240		};
241	79×	let boundary = self.boundary();
242	79×	let partial = self.config.max_blob.is_some() \|\| !self.config.sparse_paths.is_empty();
243
244	79×	let mut roots: Vec<(ChangeHash, String)> = Vec::new();
245	79×	let mut states: Vec<(ObjectRef, String)> = Vec::new();
246	79×	states.extend(
247	79×	view.bookmarks
248	79×	.iter()
249	80×	.map(\|(n, s)\| (*s, format!("bookmark {n}"))),
250		);
251	79×	states.extend(view.tags.iter().map(\|(n, s)\| (*s, format!("tag {n}"))));
252	79×	for (remote, bms) in &view.remotes {
253	48×	states.extend(bms.iter().map(\|(n, s)\| (*s, format!("{remote}/{n}"))));
254		}
255	136×	for (st, what) in states {
256	136×	match self.store.get_state(&st) {
257	196×	Ok(s) => roots.extend(s.heads.into_iter().map(\|h\| (h, what.clone()))),
258	0×	Err(e) if !bad.contains(&st) => {
259	0×	rep.error(Some(st), format!("state of {what} unusable: {e}"))
260		}
261	0×	Err(_) => {}
262		}
263		}
264	79×	for (ws, h) in &view.workspaces {
265	79×	roots.push((*h, format!("workspace {}", ws.to_hex())));
266	79×	}
267	79×	if let Some(p) = view.policy {
268	0×	match self.store.get_typed(&p, ObjectType::Policy) {
269	0×	Ok(bytes) => {
270	0×	if let Ok(policy) = crate::policy::Policy::decode(&bytes) {
271	0×	for id in &policy.identities {
272	0×	if !self.store.has(&id.hash()) {
273	0×	rep.warn(
274	0×	Some(id.hash()),
275	0×	"an identity the policy names is not stored",
276	0×	);
277	0×	}
278		}
279	0×	}
280		}
281	0×	Err(e) if !bad.contains(&p) => {
282	0×	rep.error(Some(p), format!("the policy is unusable: {e}"))
283		}
284	0×	Err(_) => {}
285		}
286	79×	}
287
288		// Working-copy changes are never signed; they are checked but not tallied.
289	79×	let working: HashSet<ChangeHash> = view.workspaces.values().copied().collect();
290	79×	let mut seen: HashSet<ChangeHash> = HashSet::new();
291	79×	let mut blobs_seen: HashSet<ObjectRef> = HashSet::new();
292	79×	let mut stack = roots;
293	494×	while let Some((h, needed_by)) = stack.pop() {
294	415×	if !seen.insert(h) {
295	175×	continue;
296	240×	}
297		// The skeleton: the text is checked as the blobs it is stored
298		// in, not decoded (the objects pass checked every object).
299	240×	let (c, texts) = match self.store.get_change_lazy(&h) {
300	238×	Ok(c) => c,
301	1×	Err(StoreError::NotFound(_)) if boundary.contains(&h) => continue,
302		Err(StoreError::NotFound(_)) => {
303	1×	rep.error(Some(h), format!("change missing, needed by {needed_by}"));
304	1×	continue;
305		}
306	1×	Err(e) => {
307	1×	if !bad.contains(&h) {
308	0×	rep.error(Some(h), format!("change unusable: {e}"));
309	1×	}
310	1×	continue;
311		}
312		};
313	238×	rep.changes += 1;
314	238×	let label = format!("change {}", h.short(12));
315	238×	stack.extend(c.deps.iter().map(\|d\| (*d, label.clone())));
316	238×	if !c.author.is_zero() && !self.store.has(&c.author) {
317	0×	rep.warn(
318	0×	Some(h),
319	0×	format!("author identity {} is not stored", c.author.short(12)),
320	0×	);
321	238×	}
322	238×	for t in texts {
323	198×	if let crate::objects::TextRef::Blob(b, _, _) = t {
324	46×	if blobs_seen.insert(b) {
325	46×	// A change cannot be read without its text: never partial.
326	46×	self.fsck_blob(rep, b, &label, false, bad);
327	46×	}
328	152×	}
329		}
330	238×	for e in &c.edits {
331	213×	let Edit::Replace { old, new, .. } = e else {
332	201×	continue;
333		};
334	14×	for b in [old, new].into_iter().flatten() {
335	14×	if !blobs_seen.insert(*b) {
336	2×	continue;
337	12×	}
338	12×	self.fsck_blob(rep, *b, &label, partial, bad);
339		}
340		}
341	238×	if working.contains(&h) {
342	79×	continue;
343	159×	}
344	159×	match self.authenticate_change(&h) {
345	105×	Ok(SignatureStatus::Valid { .. }) => rep.signed += 1,
346	54×	Ok(SignatureStatus::Unsigned) => rep.unsigned += 1,
347	0×	Ok(SignatureStatus::Invalid(why)) => {
348	0×	rep.error(Some(h), format!("bad signature: {why}"))
349		}
350	0×	Err(e) => rep.error(Some(h), format!("cannot check signatures: {e}")),
351		}
352		}
353	79×	Ok(())
354	80×	}
355
356	58×	fn fsck_blob(
357	58×	&self,
358	58×	rep: &mut FsckReport,
359	58×	b: ObjectRef,
360	58×	used_by: &str,
361	58×	partial: bool,
362	58×	bad: &HashSet<ObjectRef>,
363	58×	) {
364	58×	match self.store.get_typed(&b, ObjectType::Blob) {
365	56×	Ok(bytes) => {
366	56×	let Ok(blob) = Blob::decode(&bytes) else {
367	0×	return;
368		};
369	95×	for ch in blob.chunks {
370	95×	if !self.store.has(&ch) {
371	0×	rep.error(
372	0×	Some(ch),
373	0×	format!("chunk missing from blob {} ({used_by})", b.short(12)),
374	0×	);
375	95×	}
376		}
377		}
378	0×	Err(StoreError::NotFound(_)) if partial => rep.warn(
379	0×	Some(b),
380	0×	format!("blob not fetched ({used_by}); expected in a partial clone"),
381		),
382	2×	Err(StoreError::NotFound(_)) => rep.error(Some(b), format!("blob missing ({used_by})")),
383	0×	Err(e) if !bad.contains(&b) => {
384	0×	rep.error(Some(b), format!("blob unusable ({used_by}): {e}"))
385		}
386	0×	Err(_) => {}
387		}
388	58×	}
389
390		/// Whether `heads` form a valid state: its history loads and every file
391		/// graph builds. Graphs are built a batch at a time and dropped, so the
392		/// check needs the memory of a batch, not of the whole tree.
393	80×	fn check_state(&self, heads: &[ChangeHash]) -> Result<()> {
394	80×	let mut st = self.load_state_history(heads)?;
395	80×	let nodes = st.unbuilt_nodes();
396	80×	for batch in nodes.chunks(2048) {
397	76×	self.load_graphs_of(&mut st, batch)?;
398	74×	st.drop_graphs(batch);
399		}
400	78×	Ok(())
401	80×	}
402
403		/// Pass 4.
404	80×	fn fsck_derived(&self, rep: &mut FsckReport) -> Result<()> {
405	80×	let Ok(view) = self.view() else { return Ok(()) };
406	79×	let shallow = self.is_shallow();
407	79×	if !shallow {
408	79×	let mut checked: HashSet<ObjectRef> = HashSet::new();
409	80×	for (name, st) in &view.bookmarks {
410	80×	if !checked.insert(*st) {
411	0×	continue;
412	80×	}
413	80×	let Ok(s) = self.store.get_state(st) else {
414	0×	continue;
415		};
416	80×	if let Err(e) = self.check_state(&s.heads) {
417	2×	rep.error(
418	2×	Some(*st),
419	2×	format!("bookmark {name} does not form a valid state: {e}"),
420	2×	);
421	78×	}
422		}
423	0×	}
424	79×	for r in self.checkpoint_refs()? {
425	9×	rep.checkpoints += 1;
426	9×	let cp = match self.store.get_typed(&r, ObjectType::Checkpoint) {
427	9×	Ok(b) => match crate::objects::Checkpoint::decode(&b) {
428	9×	Ok(cp) => cp,
429	0×	Err(_) => continue,
430		},
431		Err(StoreError::NotFound(_)) => {
432	0×	rep.warn(
433	0×	Some(r),
434		"checkpoint pointer names a missing checkpoint; it will be rewritten",
435		);
436	0×	continue;
437		}
438	0×	Err(_) => continue,
439		};
440	9×	let payload = match self.read_blob(&cp.payload) {
441	9×	Ok(p) => p,
442	0×	Err(e) => {
443	0×	rep.error(Some(r), format!("checkpoint payload unreadable: {e}"));
444	0×	continue;
445		}
446		};
447	9×	if let Err(e) = self.state_from_checkpoint(&payload) {
448	1×	rep.error(Some(r), format!("checkpoint payload does not decode: {e}"));
449	1×	continue;
450	8×	}
451		// The payload must be what the named changes produce, in its own version.
452	8×	let version = crate::cbor::decode(&payload).ok().and_then(\|v\| {
453	8×	crate::objects::map(&v, "payload").ok().and_then(\|m\| {
454	8×	m.get(&0)
455	8×	.and_then(\|x\| crate::objects::uint(x, "version").ok())
456	8×	})
457	8×	});
458	8×	let expected = \|full: &State\| {
459	6×	if version == Some(crate::state::CHECKPOINT_INDEX_VERSION) {
460	5×	full.checkpoint_index(&full.checkpoint_file_graphs(), Repo::blob_ref)
461		} else {
462	1×	full.checkpoint_payload()
463		}
464	6×	};
465	8×	if !shallow {
466	8×	match self.load_state_full(&cp.heads) {
467	6×	Ok(full) if expected(&full) != payload => {
468	0×	rep.error(Some(r), "checkpoint disagrees with the changes it names")
469		}
470	6×	Ok(_) => {}
471	2×	Err(e) => rep.error(
472	2×	Some(r),
473	2×	format!("checkpoint names changes that do not form a state: {e}"),
474		),
475		}
476	0×	}
477		}
478	79×	let mut attests = Vec::new();
479	80×	for name in view.bookmarks.keys() {
480	80×	if let Ok(Some((r, _))) = self.latest_ref_attest(name) {
481	21×	attests.push((r, format!("bookmark {name}")));
482	59×	}
483		}
484	79×	for name in view.tags.keys() {
485	8×	if let Ok(Some((r, _))) = self.tag_attest(name) {
486	8×	attests.push((r, format!("tag {name}")));
487	8×	}
488		}
489	79×	for (r, what) in attests {
490	29×	rep.attestations += 1;
491	29×	let sigs = self.store.signatures(&r)?;
492	29×	if sigs.is_empty() {
493	0×	rep.warn(Some(r), format!("attestation for {what} is unsigned"));
494	29×	}
495	29×	for sig in sigs {
496	29×	match self.store.get_identity(&sig.key) {
497	29×	Ok(id) => {
498	29×	if let Err(e) = verify_signature(&id, &sig, Domain::Attest, &r) {
499	0×	rep.error(
500	0×	Some(r),
501	0×	format!("attestation for {what} has a bad signature: {e}"),
502	0×	);
503	29×	}
504		}
505	0×	Err(_) => rep.warn(
506	0×	Some(r),
507	0×	format!(
508		"attestation for {what} is signed by {}, whose identity is not stored",
509	0×	sig.key.short(12)
510		),
511		),
512		}
513		}
514		}
515	79×	Ok(())
516	80×	}
517		}

Coverage / crates/oxy-core/src/repo/fsck.rs 71.6%

Keyboard