krabka_backup/
manifest.rs

1//! What a capture writes beside its artifacts, and the check that reads it
2//! back.
3//!
4//! A capture is a directory of opaque bytes: an operator cannot look at a
5//! controller checkpoint and tell whether it arrived whole. The manifest is
6//! what makes the copy checkable — it names every artifact with the size and
7//! the SHA-256 of the bytes that were uploaded, so `krabka-backup verify` can
8//! answer the only question that matters before a disaster, which is whether
9//! this copy would restore.
10
11use {
12    crate::error::BackupError,
13    serde::{Deserialize, Serialize, de::DeserializeOwned},
14    sha2::{Digest as _, Sha256},
15};
16
17/// Version of `manifest.json`, its required top-level `"version"` field.
18///
19/// Part of the 1.x on-disk contract: a 1.x `krabka-backup` reads every
20/// manifest that any earlier 1.x build wrote.
21pub const MANIFEST_VERSION: i16 = 0;
22
23/// Directory the captures live under, inside the archive.
24pub const CAPTURE_ROOT: &str = "restore-inputs";
25
26/// Object name of the manifest inside one capture.
27pub const MANIFEST: &str = "manifest.json";
28
29/// Object name of the captured RLMM snapshot.
30pub const RLMM_SNAPSHOT: &str = "rlmm-snapshot";
31
32/// Object name of the captured controller metadata checkpoint.
33pub const METADATA_CHECKPOINT: &str = "cluster-metadata.checkpoint";
34
35/// Object name of the captured committed group offsets.
36pub const GROUP_OFFSETS: &str = "group-offsets.json";
37
38/// Object name of the committed diskless-WAL projection.
39pub const DISKLESS_WAL_INDEX: &str = "diskless-wal-index.json";
40
41/// One captured artifact, as it was uploaded.
42#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
43pub struct Artifact {
44    /// Object name inside the capture, one of the four constants above.
45    pub name: String,
46    /// Where the bytes came from: a path on the node, or the cluster address
47    /// they were read from.
48    pub source: String,
49    /// Bytes uploaded.
50    pub size_bytes: u64,
51    /// Lowercase hex SHA-256 of the uploaded bytes.
52    pub sha256: String,
53}
54
55/// The record of one capture.
56#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
57pub struct Manifest {
58    /// Always [`MANIFEST_VERSION`] when this build writes it.
59    pub version: i16,
60    /// The capture's own id, which is also its directory name.
61    pub capture_id: String,
62    /// When the capture ran, in milliseconds since the Unix epoch.
63    pub captured_at_ms: u64,
64    /// The log directory the file artifacts were read from, when a capture
65    /// took any.
66    pub log_dir: Option<String>,
67    /// The cluster the group offsets were read from, when a capture took them.
68    pub bootstrap_server: Option<String>,
69    /// Every artifact this capture wrote, in the order it wrote them.
70    pub artifacts: Vec<Artifact>,
71}
72
73impl Manifest {
74    /// Decode a manifest, refusing a missing or unknown `"version"`.
75    ///
76    /// # Errors
77    ///
78    /// [`BackupError::UnsupportedVersion`] for a manifest with no version or
79    /// one other than [`MANIFEST_VERSION`], and [`BackupError::Json`] for bytes
80    /// that are not a manifest. `context` names the object in either error.
81    pub fn from_slice(bytes: &[u8], context: &str) -> Result<Self, BackupError> {
82        decode_versioned(bytes, context, MANIFEST_VERSION)
83    }
84
85    /// The artifact of this name, if the capture took one.
86    #[must_use]
87    pub fn artifact(&self, name: &str) -> Option<&Artifact> {
88        self.artifacts.iter().find(|entry| entry.name == name)
89    }
90}
91
92/// Just the version of a capture document, read before the rest of it.
93#[derive(Deserialize)]
94struct VersionProbe {
95    version: Option<serde_json::Value>,
96}
97
98/// Decode a JSON capture document whose top-level `"version"` must be
99/// `expected`.
100///
101/// The version is read on its own first, so a document of another version is
102/// reported as that, and not as whatever field its shape lacks.
103pub(crate) fn decode_versioned<T: DeserializeOwned>(
104    bytes: &[u8],
105    context: &str,
106    expected: i16,
107) -> Result<T, BackupError> {
108    let json = |source| BackupError::Json {
109        context: context.to_owned(),
110        source,
111    };
112    let probe: VersionProbe = serde_json::from_slice(bytes).map_err(json)?;
113    if !matches!(&probe.version, Some(version) if *version == expected) {
114        return Err(BackupError::UnsupportedVersion {
115            context: context.to_owned(),
116            found: probe.version.map(|version| version.to_string()),
117            expected,
118        });
119    }
120    serde_json::from_slice(bytes).map_err(json)
121}
122
123/// Lowercase hex SHA-256 of `bytes`.
124#[must_use]
125pub fn sha256_hex(bytes: &[u8]) -> String {
126    let mut hasher = Sha256::new();
127    hasher.update(bytes);
128    hex::encode(hasher.finalize())
129}
130
131/// What is wrong with one artifact's bytes, or `None` when they are the bytes
132/// the manifest recorded.
133///
134/// The size is checked as well as the digest, because a size mismatch names
135/// the failure an operator can act on — a truncated upload — while a digest
136/// mismatch alone only says the bytes differ.
137#[must_use]
138pub fn artifact_problem(expected: &Artifact, actual: &[u8]) -> Option<String> {
139    let actual_size = actual.len() as u64;
140    if actual_size != expected.size_bytes {
141        return Some(format!(
142            "{}: archive holds {actual_size} bytes, manifest recorded {}",
143            expected.name, expected.size_bytes
144        ));
145    }
146    let actual_sha = sha256_hex(actual);
147    (actual_sha != expected.sha256).then(|| {
148        format!(
149            "{}: archive holds sha256 {actual_sha}, manifest recorded {}",
150            expected.name, expected.sha256
151        )
152    })
153}
154
155#[cfg(test)]
156mod tests {
157    use {
158        super::{Artifact, MANIFEST_VERSION, Manifest, artifact_problem, sha256_hex},
159        crate::error::BackupError,
160        assert2::check,
161    };
162
163    fn artifact(bytes: &[u8]) -> Artifact {
164        Artifact {
165            name: "rlmm-snapshot".to_owned(),
166            source: "/var/lib/krabka/remote-log-metadata/snapshot".to_owned(),
167            size_bytes: bytes.len() as u64,
168            sha256: sha256_hex(bytes),
169        }
170    }
171
172    #[test]
173    fn the_empty_input_hashes_to_the_published_sha256_of_the_empty_string() {
174        check!(
175            sha256_hex(b"") == "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
176        );
177    }
178
179    #[test]
180    fn bytes_that_match_the_manifest_are_no_problem() {
181        check!(artifact_problem(&artifact(b"snapshot bytes"), b"snapshot bytes") == None);
182    }
183
184    #[test]
185    fn a_short_read_is_reported_as_a_size_mismatch() {
186        let problem = artifact_problem(&artifact(b"snapshot bytes"), b"snapshot")
187            .expect("a short read is a problem");
188        check!(problem.contains("8 bytes"), "got: {problem}");
189        check!(problem.contains("14"), "got: {problem}");
190    }
191
192    #[test]
193    fn same_length_different_bytes_are_reported_as_a_digest_mismatch() {
194        let problem = artifact_problem(&artifact(b"snapshot bytes"), b"snapshot bytez")
195            .expect("different bytes are a problem");
196        check!(problem.contains("sha256"), "got: {problem}");
197    }
198
199    fn golden_manifest() -> Manifest {
200        Manifest {
201            version: MANIFEST_VERSION,
202            capture_id: "0001700000000000".to_owned(),
203            captured_at_ms: 1_700_000_000_000,
204            log_dir: Some("/var/lib/krabka".to_owned()),
205            bootstrap_server: Some("broker-1:9092".to_owned()),
206            artifacts: vec![artifact(b"snapshot bytes")],
207        }
208    }
209
210    /// The exact bytes this build writes for [`golden_manifest`]. A change
211    /// here is a change to the 1.x capture format.
212    const GOLDEN_MANIFEST: &str = concat!(
213        r#"{"version":0,"capture_id":"0001700000000000","captured_at_ms":1700000000000,"#,
214        r#""log_dir":"/var/lib/krabka","bootstrap_server":"broker-1:9092","artifacts":"#,
215        r#"[{"name":"rlmm-snapshot","source":"/var/lib/krabka/remote-log-metadata/snapshot","#,
216        r#""size_bytes":14,"#,
217        r#""sha256":"ee36ef8194c4dd1e734e6d64f62653008e6566dbd5e2cd7f289f0f4f3e4467a4"}]}"#,
218    );
219
220    #[test]
221    fn a_manifest_encodes_to_the_golden_bytes_and_decodes_back() {
222        let manifest = golden_manifest();
223        let encoded = serde_json::to_string(&manifest).expect("encode the manifest");
224        check!(encoded == GOLDEN_MANIFEST);
225        let decoded = Manifest::from_slice(GOLDEN_MANIFEST.as_bytes(), "manifest.json")
226            .expect("decode the manifest");
227        check!(decoded == manifest);
228        check!(decoded.artifact("rlmm-snapshot").is_some());
229        check!(decoded.artifact("group-offsets.json").is_none());
230    }
231
232    #[yare::parameterized(
233        pre_1_0 = {
234            "pre-1.0 manifest",
235            r#"{"capture_id":"1","captured_at_ms":1,"log_dir":null,"bootstrap_server":null,"artifacts":[]}"#,
236            None,
237        },
238        future = {
239            "future version",
240            r#"{"version":1,"capture_id":"1","captured_at_ms":1,"log_dir":null,"bootstrap_server":null,"artifacts":[]}"#,
241            Some("1"),
242        },
243    )]
244    fn a_manifest_with_a_missing_or_unknown_version_is_refused(
245        name: &str,
246        json: &str,
247        found: Option<&str>,
248    ) {
249        let error = Manifest::from_slice(json.as_bytes(), "restore-inputs/1/manifest.json")
250            .expect_err(name);
251        assert2::assert!(
252            let BackupError::UnsupportedVersion {
253                context,
254                found: actual,
255                expected: MANIFEST_VERSION
256            } = &error,
257            "case {name}: {error}"
258        );
259        check!(context == "restore-inputs/1/manifest.json", "case {name}");
260        check!(actual.as_deref() == found, "case {name}");
261    }
262}
krabka_backup/manifest.rs

krabka_backup/
manifest.rs
