backup/
manifest.rs

1//! The manifest: one consistent restore point, and the object keys it names.
2//!
3//! Schema per W4 §3.1, with one deliberate change: each object carries a
4//! CRC-32 (`crc32`, eight lowercase hex digits) instead of a SHA-256. The
5//! checksum is there to catch corruption, not tampering, and S3 verifies every
6//! upload with its own SHA-256 checksum besides.
7//!
8//! Every object a manifest names has an immutable key: the snapshot and the
9//! archives carry the generation they were uploaded for, a side file its own
10//! checksum and size, a segment its generation and byte range. Nothing a
11//! manifest names is ever overwritten, so every manifest still in the bucket
12//! verifies — not only the newest. Each entry also says which file of the
13//! store it restores to (`name`), since the key no longer does.
14//!
15//! # Lineage
16//!
17//! Every manifest, and every entry in it, carries the `lineage` of the store
18//! it was taken from: a random id the worker gives a store the first time it
19//! backs it up (the `backup.lineage` file, which travels with every restore
20//! point). A prefix belongs to one lineage. A manifest whose entries name
21//! more than one lineage does not parse ([`Manifest::from_json`]): a restore
22//! point is never assembled from two stores' objects.
23
24use serde::{Deserialize, Serialize};
25
26/// The only manifest version this release writes or reads. Version 2 added
27/// `lineage`; version 1 was never released.
28pub const MANIFEST_VERSION: u32 = 2;
29
30/// One consistent restore point: the snapshot, the archives and side files
31/// that go with it, and the ordered WAL segments of one generation.
32///
33/// Restoring it means: download `files`, concatenate `segments` (all of
34/// `wal_generation`, in offset order, starting at offset 0) into `wal.bin`,
35/// and open the store.
36#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
37pub struct Manifest {
38    /// [`MANIFEST_VERSION`].
39    pub version: u32,
40    /// The lineage of the store this restore point was taken from.
41    pub lineage: String,
42    /// When the manifest was written, RFC 3339 in UTC.
43    pub created_at: String,
44    /// Global frame index one past the last frame this restore point holds —
45    /// the index space `open_at` and `--at <commit>` use: what a store opened
46    /// from `files` plus `segments` reports as `wal_total_commits`. Computed
47    /// from exactly those objects: `oldest_commit_seq` (the `wal.floor` it
48    /// lists) plus every listed archive's and segment's `frames`. Also the
49    /// manifest's key, so manifests sort by it.
50    pub commit_seq: u64,
51    /// The first global frame index still reachable through the archives
52    /// (the store's `wal.floor`; 0 when nothing was ever pruned). A point in
53    /// time before it cannot be restored.
54    pub oldest_commit_seq: u64,
55    /// Snapshot, archives and side files, each with its size and checksum.
56    pub files: Vec<FileEntry>,
57    /// The WAL generation `segments` belong to. A new generation starts each
58    /// time the store's `wal.bin` is replaced, so offsets never collide.
59    pub wal_generation: u64,
60    /// Segments of `wal_generation`, in offset order, contiguous from 0.
61    pub segments: Vec<SegmentEntry>,
62}
63
64/// One whole object named by a manifest.
65#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
66pub struct FileEntry {
67    /// The store file it restores to: `snapshot.bin`, `wal.N.archive`,
68    /// `commit_times.bin`, ….
69    pub name: String,
70    pub key: String,
71    pub size: u64,
72    /// CRC-32 of the object's bytes, as eight lowercase hex digits.
73    pub crc32: String,
74    /// WAL frames in the object: an archive's frame count, 0 for any other
75    /// file.
76    #[serde(default)]
77    pub frames: u64,
78    /// The lineage of the store the object was uploaded from.
79    pub lineage: String,
80}
81
82/// One WAL segment named by a manifest. Its byte range is in its key.
83#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
84pub struct SegmentEntry {
85    pub key: String,
86    /// CRC-32 of the segment's bytes, as eight lowercase hex digits.
87    pub crc32: String,
88    /// WAL frames in the segment.
89    #[serde(default)]
90    pub frames: u64,
91    /// The lineage of the store the segment was uploaded from.
92    pub lineage: String,
93}
94
95impl Manifest {
96    pub fn to_json(&self) -> Vec<u8> {
97        serde_json::to_vec_pretty(self).expect("a manifest always serialises")
98    }
99
100    /// Parse a manifest, refusing any version this release does not know and
101    /// any manifest that is not of one lineage throughout
102    /// ([`verify_lineage`](Self::verify_lineage)).
103    pub fn from_json(bytes: &[u8]) -> Result<Self, String> {
104        let m: Manifest =
105            serde_json::from_slice(bytes).map_err(|e| format!("manifest is not valid: {e}"))?;
106        if m.version != MANIFEST_VERSION {
107            return Err(format!(
108                "manifest version {} is not supported by this release (expected {MANIFEST_VERSION})",
109                m.version
110            ));
111        }
112        m.verify_lineage()?;
113        Ok(m)
114    }
115
116    /// Check that the manifest names a lineage and that every file and
117    /// segment in it was uploaded from that lineage.
118    pub fn verify_lineage(&self) -> Result<(), String> {
119        if self.lineage.is_empty() {
120            return Err("manifest names no lineage".into());
121        }
122        let foreign = self
123            .files
124            .iter()
125            .map(|f| (&f.key, &f.lineage))
126            .chain(self.segments.iter().map(|s| (&s.key, &s.lineage)))
127            .find(|(_, l)| **l != self.lineage);
128        match foreign {
129            Some((key, l)) => Err(format!(
130                "manifest of lineage {} names {key}, uploaded from lineage {l}: \
131                 a restore point mixing two stores is refused",
132                self.lineage
133            )),
134            None => Ok(()),
135        }
136    }
137}
138
139/// CRC-32 of `bytes` in the manifest's format.
140pub fn crc32_hex(bytes: &[u8]) -> String {
141    format_crc32(crc32fast::hash(bytes))
142}
143
144/// A CRC-32 value in the manifest's format.
145pub fn format_crc32(crc: u32) -> String {
146    format!("{crc:08x}")
147}
148
149/// `{prefix}/{name}`, or `name` alone when the prefix is empty.
150pub fn object_key(prefix: &str, name: &str) -> String {
151    if prefix.is_empty() {
152        name.to_string()
153    } else {
154        format!("{prefix}/{name}")
155    }
156}
157
158/// `{prefix}/wal/{generation}/{start:020}-{end:020}`.
159pub fn segment_key(prefix: &str, generation: u64, start: u64, end: u64) -> String {
160    object_key(prefix, &format!("wal/{generation}/{start:020}-{end:020}"))
161}
162
163/// The `(generation, start, end)` a segment key names, or `None` when `key`
164/// is not a segment key under `prefix`.
165pub fn parse_segment_key(prefix: &str, key: &str) -> Option<(u64, u64, u64)> {
166    let rest = key.strip_prefix(&object_key(prefix, "wal/"))?;
167    let (generation, range) = rest.split_once('/')?;
168    let (start, end) = range.split_once('-')?;
169    Some((
170        generation.parse().ok()?,
171        start.parse().ok()?,
172        end.parse().ok()?,
173    ))
174}
175
176/// `{prefix}/manifest/{commit_seq:020}.json`.
177pub fn manifest_key(prefix: &str, commit_seq: u64) -> String {
178    object_key(prefix, &format!("manifest/{commit_seq:020}.json"))
179}
180
181/// `{prefix}/LATEST`: holds the newest manifest's key as plain UTF-8.
182pub fn latest_key(prefix: &str) -> String {
183    object_key(prefix, "LATEST")
184}
185
186/// `{prefix}/snapshots/{generation}-{commit_seq}.bin`: the snapshot uploaded
187/// for `generation`, when the store's durable position was `commit_seq`.
188pub fn snapshot_key(prefix: &str, generation: u64, commit_seq: u64) -> String {
189    object_key(prefix, &format!("snapshots/{generation}-{commit_seq}.bin"))
190}
191
192/// `{prefix}/archives/{generation}-wal.{n}.archive`: archive `n`, uploaded
193/// while shipping `generation`.
194pub fn archive_key(prefix: &str, generation: u64, n: u64) -> String {
195    object_key(prefix, &format!("archives/{generation}-wal.{n}.archive"))
196}
197
198/// `{prefix}/side/{name}/{crc32}-{size}`: a side file, addressed by content.
199pub fn side_key(prefix: &str, name: &str, crc32: &str, size: u64) -> String {
200    object_key(prefix, &format!("side/{name}/{crc32}-{size}"))
201}
202
203/// The generation a snapshot or archive key was uploaded for, or `None` for
204/// any other key.
205pub fn parse_versioned_generation(prefix: &str, key: &str) -> Option<u64> {
206    let rest = key
207        .strip_prefix(&object_key(prefix, "snapshots/"))
208        .or_else(|| key.strip_prefix(&object_key(prefix, "archives/")))?;
209    rest.split_once('-')?.0.parse().ok()
210}
211
212/// The archive number in a store file name `wal.{n}.archive`.
213pub fn parse_archive_name(name: &str) -> Option<u64> {
214    name.strip_prefix("wal.")?
215        .strip_suffix(".archive")?
216        .parse()
217        .ok()
218}
219
220#[cfg(test)]
221mod tests {
222    use super::*;
223
224    #[test]
225    fn keys_are_zero_padded_and_parse_back() {
226        let k = segment_key("p", 3, 0, 65536);
227        assert_eq!(k, "p/wal/3/00000000000000000000-00000000000000065536");
228        assert_eq!(parse_segment_key("p", &k), Some((3, 0, 65536)));
229        assert_eq!(parse_segment_key("q", &k), None);
230        assert_eq!(
231            segment_key("", 0, 1, 2),
232            "wal/0/00000000000000000001-00000000000000000002"
233        );
234        assert_eq!(
235            manifest_key("p", 42),
236            "p/manifest/00000000000000000042.json"
237        );
238        assert_eq!(latest_key(""), "LATEST");
239        assert_eq!(archive_key("a/b", 3, 7), "a/b/archives/3-wal.7.archive");
240        assert_eq!(snapshot_key("p", 4, 99), "p/snapshots/4-99.bin");
241        assert_eq!(
242            side_key("p", "wal.floor", "0000abcd", 8),
243            "p/side/wal.floor/0000abcd-8"
244        );
245        assert_eq!(
246            parse_versioned_generation("p", "p/snapshots/4-99.bin"),
247            Some(4)
248        );
249        assert_eq!(
250            parse_versioned_generation("a/b", "a/b/archives/3-wal.7.archive"),
251            Some(3)
252        );
253        assert_eq!(parse_versioned_generation("p", "p/side/x/1-2"), None);
254        assert_eq!(parse_archive_name("wal.7.archive"), Some(7));
255        assert_eq!(parse_archive_name("wal.floor"), None);
256    }
257
258    #[test]
259    fn an_unknown_version_is_refused() {
260        let mut m = Manifest {
261            version: MANIFEST_VERSION + 1,
262            lineage: "l".into(),
263            created_at: "2026-10-20T00:00:00Z".into(),
264            commit_seq: 0,
265            oldest_commit_seq: 0,
266            files: vec![],
267            wal_generation: 0,
268            segments: vec![],
269        };
270        assert!(Manifest::from_json(&m.to_json()).is_err());
271        m.version = MANIFEST_VERSION;
272        assert!(Manifest::from_json(&m.to_json()).is_ok());
273    }
274
275    #[test]
276    fn a_restore_point_mixing_two_lineages_does_not_parse() {
277        let entry = |lineage: &str| FileEntry {
278            name: "wal.3.archive".into(),
279            key: "p/archives/0-wal.3.archive".into(),
280            size: 1,
281            crc32: "00000000".into(),
282            frames: 1,
283            lineage: lineage.into(),
284        };
285        let mut m = Manifest {
286            version: MANIFEST_VERSION,
287            lineage: "b".into(),
288            created_at: "2026-10-22T00:00:00Z".into(),
289            commit_seq: 1,
290            oldest_commit_seq: 0,
291            files: vec![entry("b")],
292            wal_generation: 0,
293            segments: vec![],
294        };
295        assert!(Manifest::from_json(&m.to_json()).is_ok());
296        m.files.push(entry("a"));
297        let e = Manifest::from_json(&m.to_json()).unwrap_err();
298        assert!(e.contains("lineage a"), "{e}");
299        m.files.pop();
300        m.segments.push(SegmentEntry {
301            key: "p/wal/0/x".into(),
302            crc32: "00000000".into(),
303            frames: 0,
304            lineage: "a".into(),
305        });
306        assert!(Manifest::from_json(&m.to_json()).is_err());
307        m.segments.clear();
308        m.lineage.clear();
309        assert!(Manifest::from_json(&m.to_json()).is_err());
310    }
311
312    #[test]
313    fn crc_is_eight_hex_digits() {
314        assert_eq!(crc32_hex(b""), "00000000");
315        assert_eq!(crc32_hex(b"123456789"), "cbf43926");
316    }
317}
backup/manifest.rs

backup/
manifest.rs
