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