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