krabka_remote_storage/worm/verify/
report.rs

1//! The values a verification run hands back, and the steps that assemble them.
2//!
3//! A run describes what it found rather than what it did: one report per
4//! partition directory, the chain runs and the offset holes inside it, and the
5//! archive-wide roll-up an auditor grades. The assembly helpers here turn the
6//! walk's accumulated state into those values.
7
8use {
9    crate::worm::manifest::{ChainHead, EpochId, ManifestBody, ManifestSeq, ObjectEntry},
10    std::collections::BTreeMap,
11};
12
13const PROTECTION_SAMPLE_LIMIT: usize = 10;
14
15/// Count and bounded key sample for one object-protection mechanism.
16#[derive(Debug, Clone, Default, PartialEq, Eq)]
17pub struct ObjectProtectionReport {
18    /// Number of verified objects in this class.
19    pub count: u64,
20    /// Up to ten representative object keys.
21    pub sample: Vec<String>,
22}
23
24impl ObjectProtectionReport {
25    pub(super) fn record(&mut self, key: &str) {
26        self.count = self.count.saturating_add(1);
27        if self.sample.len() < PROTECTION_SAMPLE_LIMIT {
28            self.sample.push(key.to_owned());
29        }
30    }
31}
32
33/// One unbroken run of a partition's chain, as the archive holds it.
34#[derive(Debug, Clone, PartialEq, Eq)]
35pub struct EpochSpan {
36    /// The run's identifier.
37    pub epoch_id: EpochId,
38    /// Sequence of the run's first manifest.
39    pub first_seq: ManifestSeq,
40    /// Sequence of the run's last verified manifest.
41    pub last_seq: ManifestSeq,
42    /// Manifests verified in this run.
43    pub manifests: u64,
44    /// Lowest segment start offset in the run.
45    pub start_offset: i64,
46    /// Highest segment end offset in the run.
47    pub end_offset: i64,
48    /// Chain head after the run's last verified manifest.
49    pub head: ChainHead,
50}
51
52/// A hole between two consecutive archived segments.
53#[derive(Debug, Clone, Copy, PartialEq, Eq)]
54pub struct OffsetGap {
55    /// Last offset the archive holds before the hole.
56    pub after: i64,
57    /// First offset the archive holds after the hole.
58    pub before: i64,
59}
60
61/// The first detected break in one partition.
62#[derive(Debug, Clone, PartialEq, Eq)]
63pub struct VerifyBreak {
64    /// Object-store key of the manifest the walk stopped at.
65    pub manifest_key: String,
66    /// Chain position of that manifest, when it decoded far enough to have one.
67    pub seq: Option<ManifestSeq>,
68    /// What is wrong, in one sentence.
69    pub reason: String,
70}
71
72/// Verification result for one partition directory.
73#[derive(Debug, Clone, PartialEq, Eq)]
74pub struct PartitionVerifyReport {
75    /// The directory this report covers, as an object-store key prefix.
76    pub partition_dir: String,
77    /// Manifests verified before the walk stopped.
78    pub manifests: u64,
79    /// Object entries checked before the walk stopped.
80    pub objects_checked: u64,
81    /// Objects written with an atomic create precondition.
82    pub create_precondition_objects: ObjectProtectionReport,
83    /// Objects whose multipart write relied on bucket retention.
84    pub bucket_retention_objects: ObjectProtectionReport,
85    /// Chain runs found, ordered by their lowest segment start offset.
86    pub epochs: Vec<EpochSpan>,
87    /// Manifests that carry no signature at all.
88    pub unsigned_manifests: u64,
89    /// Manifests signed by a `key_id` the run does not trust.
90    pub untrusted_manifests: u64,
91    /// Objects in the directory that no manifest names, sorted.
92    pub orphan_objects: Vec<String>,
93    /// Holes between consecutive archived segments, sorted.
94    pub offset_gaps: Vec<OffsetGap>,
95    /// Chain head after the last verified manifest.
96    pub head: Option<ChainHead>,
97    /// `true` only for an unbroken, fully attested chain with no orphan object.
98    pub ok: bool,
99    /// The break, when there is one.
100    pub first_break: Option<VerifyBreak>,
101}
102
103/// Verification result for a whole archive.
104#[derive(Debug, Clone, PartialEq, Eq)]
105pub struct ArchiveVerifyReport {
106    /// One report per partition directory, sorted by directory.
107    pub partitions: Vec<PartitionVerifyReport>,
108    /// Unreferenced objects outside canonical partition directories.
109    pub global_orphan_objects: Vec<String>,
110}
111
112impl ArchiveVerifyReport {
113    /// `true` when every partition is unbroken, fully attested, and complete.
114    #[must_use]
115    pub fn ok(&self) -> bool {
116        self.global_orphan_objects.is_empty()
117            && self.partitions.iter().all(|partition| partition.ok)
118    }
119
120    /// Manifests verified across every partition.
121    #[must_use]
122    pub fn manifests(&self) -> u64 {
123        self.partitions.iter().fold(0u64, |total, partition| {
124            total.saturating_add(partition.manifests)
125        })
126    }
127
128    /// The break in the first broken partition, in partition order.
129    #[must_use]
130    pub fn first_break(&self) -> Option<&VerifyBreak> {
131        self.partitions
132            .iter()
133            .find_map(|partition| partition.first_break.as_ref())
134    }
135
136    /// Every manifest verified against a trusted key.
137    #[must_use]
138    pub fn fully_attested(&self) -> bool {
139        self.partitions.iter().all(|partition| {
140            partition.unsigned_manifests == 0 && partition.untrusted_manifests == 0
141        })
142    }
143
144    /// Any partition holds more than one chain run.
145    #[must_use]
146    pub fn has_epoch_restarts(&self) -> bool {
147        self.partitions
148            .iter()
149            .any(|partition| partition.epochs.len() > 1)
150    }
151}
152
153/// A partition's report and the signed claims of the objects its walk accepted.
154pub(super) struct VerifiedPartition {
155    pub report: PartitionVerifyReport,
156    pub objects: BTreeMap<String, ObjectEntry>,
157}
158
159/// A partition whose manifests could not all be decoded, so no walk ran.
160pub(super) fn broken_before_walk(
161    dir: &str,
162    orphan_objects: Vec<String>,
163    first_break: VerifyBreak,
164) -> PartitionVerifyReport {
165    PartitionVerifyReport {
166        partition_dir: dir.to_owned(),
167        manifests: 0,
168        objects_checked: 0,
169        create_precondition_objects: ObjectProtectionReport::default(),
170        bucket_retention_objects: ObjectProtectionReport::default(),
171        epochs: Vec::new(),
172        unsigned_manifests: 0,
173        untrusted_manifests: 0,
174        orphan_objects,
175        offset_gaps: Vec::new(),
176        head: None,
177        ok: false,
178        first_break: Some(first_break),
179    }
180}
181
182/// Opens or widens the span that describes the run being walked.
183pub(super) fn extend_span(span: &mut Option<EpochSpan>, body: &ManifestBody, head: ChainHead) {
184    match span {
185        None => {
186            *span = Some(EpochSpan {
187                epoch_id: body.chain.epoch_id,
188                first_seq: body.chain.seq,
189                last_seq: body.chain.seq,
190                manifests: 1,
191                start_offset: body.segment.start_offset,
192                end_offset: body.segment.end_offset,
193                head,
194            });
195        }
196        Some(span) => {
197            span.last_seq = body.chain.seq;
198            span.manifests = span.manifests.saturating_add(1);
199            span.start_offset = span.start_offset.min(body.segment.start_offset);
200            span.end_offset = span.end_offset.max(body.segment.end_offset);
201            span.head = head;
202        }
203    }
204}
205
206/// The offsets one verified segment covers.
207#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
208pub(super) struct SegmentSpan {
209    pub start: i64,
210    pub end: i64,
211}
212
213/// Holes between the offset ranges the verified segments cover.
214pub(super) fn offset_gaps(segments: &mut [SegmentSpan]) -> Vec<OffsetGap> {
215    segments.sort_unstable();
216    let mut gaps = Vec::new();
217    let mut covered: Option<i64> = None;
218    for &SegmentSpan { start, end } in &*segments {
219        if let Some(after) = covered
220            && start > after.saturating_add(1)
221        {
222            gaps.push(OffsetGap {
223                after,
224                before: start,
225            });
226        }
227        covered = Some(covered.map_or(end, |after| after.max(end)));
228    }
229    gaps
230}
231
232#[cfg(test)]
233mod tests {
234    use {
235        super::*,
236        crate::worm::{
237            archiver::WormArchiver,
238            chain::WormChainRecord,
239            manifest::{ChainStamp, manifest_head},
240            verify::test_support::{Archive, SEGMENT_SPAN, put_raw},
241        },
242        assert2::check,
243        std::sync::Arc,
244    };
245
246    #[test]
247    fn object_protection_samples_are_bounded() {
248        let mut report = ObjectProtectionReport::default();
249        for index in 0..25 {
250            report.record(&format!("object-{index}"));
251        }
252
253        check!(report.count == 25);
254        check!(report.sample.len() == PROTECTION_SAMPLE_LIMIT);
255        check!(report.sample[0] == "object-0");
256        check!(report.sample[9] == "object-9");
257    }
258
259    #[tokio::test]
260    async fn a_hole_between_segments_is_reported_as_an_offset_gap() {
261        // Two runs, and the fixture numbers the second run's segments after a
262        // deleted middle segment would sit, so the archive covers 0..99 and
263        // 200..399 with nothing between.
264        let archive = Archive::build(&[3]).await;
265        archive.delete(&archive.segments[1].manifest_key).await;
266        for entry in &archive.segments[1].entries {
267            archive.delete(&entry.key).await;
268        }
269        // Re-chain the survivors so only the offsets are wrong: seq 2 now
270        // follows seq 0, so re-stamp it at seq 1 on the first manifest's head.
271        let head = manifest_head(&archive.segments[0].manifest.body);
272        let segment = &archive.segments[2];
273        let stamped = segment.metadata.clone().with_custom_metadata(
274            WormChainRecord::request(ChainStamp {
275                epoch_id: segment.manifest.body.chain.epoch_id,
276                seq: ManifestSeq(1),
277                prev_head: head,
278            })
279            .to_custom_metadata()
280            .unwrap(),
281        );
282        let sealed = WormArchiver::new(Some(Arc::clone(&archive.signer)))
283            .seal(&stamped, segment.entries.clone())
284            .unwrap();
285        put_raw(&archive.ops, &segment.manifest_key, sealed.bytes).await;
286
287        let report = archive.verify().await;
288
289        check!(report.ok());
290        check!(
291            report.partitions[0].offset_gaps
292                == vec![OffsetGap {
293                    after: SEGMENT_SPAN - 1,
294                    before: 2 * SEGMENT_SPAN,
295                }]
296        );
297    }
298}
krabka_remote_storage/worm/verify/report.rs

krabka_remote_storage/worm/verify/
report.rs
