lfsx_server/storage/
verify.rs

1use std::path::Path;
2
3use futures_util::StreamExt;
4use serde::Serialize;
5use sha2::{Digest, Sha256};
6use tokio::fs;
7
8use super::LocalStore;
9use crate::error::Error;
10use crate::namespace::Namespace;
11
12#[derive(Debug, Default, Serialize, PartialEq, Eq)]
13pub struct VerifyReport {
14    pub checked: u64,
15    pub bytes: u64,
16    pub corrupt: Vec<String>,
17    pub unreadable: Vec<String>,
18    // An audit that could not see the whole repository must not read like one
19    // that found nothing wrong. Silence is the result here, so anything that
20    // makes the silence partial has to be said out loud.
21    pub incomplete: bool,
22}
23
24impl LocalStore {
25    // Every object is named after the digest of its own contents, so the store
26    // checks itself without a manifest — the property a restore is confirmed
27    // with. Compression at rest is what took it away from `sha256sum`: the file
28    // is no longer the bytes it is named after, so reading it back has to go
29    // through the same path a download does.
30    pub async fn verify(&self, ns: &Namespace) -> Result<VerifyReport, Error> {
31        let mut report = VerifyReport::default();
32
33        let Ok(mut prefixes) = fs::read_dir(self.root.join(ns.org()).join(ns.repo())).await else {
34            return Ok(report);
35        };
36
37        while let Some(prefix) = prefixes.next_entry().await? {
38            let mut fanouts = match fs::read_dir(prefix.path()).await {
39                Ok(fanouts) => fanouts,
40                Err(error) => {
41                    tracing::warn!(path = ?prefix.path(), %error, "could not be listed");
42                    report.incomplete = true;
43                    continue;
44                }
45            };
46
47            loop {
48                match fanouts.next_entry().await {
49                    Ok(Some(fanout)) => {
50                        self.verify_directory(&fanout.path(), ns, &mut report).await;
51                    }
52                    Ok(None) => break,
53                    Err(error) => {
54                        tracing::warn!(path = ?prefix.path(), %error, "listing stopped early");
55                        report.incomplete = true;
56                        break;
57                    }
58                }
59            }
60        }
61
62        Ok(report)
63    }
64
65    async fn verify_directory(&self, directory: &Path, ns: &Namespace, report: &mut VerifyReport) {
66        let mut entries = match fs::read_dir(directory).await {
67            Ok(entries) => entries,
68            Err(error) => {
69                tracing::warn!(?directory, %error, "could not be listed");
70                report.incomplete = true;
71                return;
72            }
73        };
74
75        loop {
76            let entry = match entries.next_entry().await {
77                Ok(Some(entry)) => entry,
78                Ok(None) => return,
79                Err(error) => {
80                    tracing::warn!(?directory, %error, "listing stopped early");
81                    report.incomplete = true;
82                    return;
83                }
84            };
85
86            let oid = entry.file_name().to_string_lossy().into_owned();
87            if Self::validate_oid(&oid).is_err() {
88                continue;
89            }
90
91            report.checked += 1;
92
93            match self.digest_of(ns, &oid).await {
94                Ok((digest, read)) => {
95                    report.bytes += read;
96
97                    if digest != oid {
98                        report.corrupt.push(oid);
99                    }
100                }
101                // A file that cannot be read is its own kind of answer, and the
102                // one a failing disk gives first. Reporting it as corrupt would
103                // send an operator looking for the wrong problem.
104                Err(error) => {
105                    tracing::warn!(oid, %error, "object could not be read");
106                    report.unreadable.push(oid);
107                }
108            }
109        }
110    }
111
112    async fn digest_of(&self, ns: &Namespace, oid: &str) -> Result<(String, u64), Error> {
113        let object = self.open(ns, oid).await?;
114        let size = object.size();
115
116        let mut hasher = Sha256::new();
117        let mut read = 0u64;
118        let mut chunks = object.stream(0, size).await?;
119
120        while let Some(chunk) = chunks.next().await {
121            let chunk = chunk?;
122            read += chunk.len() as u64;
123            hasher.update(&chunk);
124        }
125
126        Ok((hex::encode(hasher.finalize()), read))
127    }
128}
129
130#[cfg(test)]
131mod tests;
lfsx_server/storage/verify.rs

lfsx_server/storage/
verify.rs
