powdb_backup/
manifest.rs

1use powdb_storage::catalog::{
2    expression_index_file_name, Catalog, IndexKeySource, CATALOG_LSN_FILE, CATALOG_VERSION,
3    LEGACY_CATALOG_VERSION,
4};
5use powdb_storage::wal::WAL_FORMAT_VERSION;
6use powdb_sync::IdentitySnapshot;
7use powdb_sync::RETAINED_SEGMENT_FORMAT_VERSION;
8use serde::{Deserialize, Serialize};
9use std::collections::BTreeSet;
10use std::io;
11use std::path::Path;
12
13fn legacy_catalog_version() -> u16 {
14    LEGACY_CATALOG_VERSION
15}
16
17pub(crate) fn validate_catalog_version(version: u16) -> io::Result<()> {
18    if !(LEGACY_CATALOG_VERSION..=CATALOG_VERSION).contains(&version) {
19        return Err(io::Error::new(
20            io::ErrorKind::InvalidData,
21            format!("unsupported snapshot catalog format {version}"),
22        ));
23    }
24    Ok(())
25}
26
27pub(crate) fn validate_catalog_transition(base: u16, next: u16) -> io::Result<()> {
28    validate_catalog_version(base)?;
29    validate_catalog_version(next)?;
30    if next < base {
31        return Err(io::Error::new(
32            io::ErrorKind::InvalidData,
33            format!("catalog format cannot move backward from {base} to {next}"),
34        ));
35    }
36    Ok(())
37}
38
39/// Durable files referenced by current catalog metadata.
40///
41/// Expression-index filenames are reconstructed only from persisted table and
42/// index IDs. Stray files from a dropped or failed index are deliberately not
43/// included in a new snapshot.
44pub(crate) fn active_durable_file_names(catalog: &Catalog) -> BTreeSet<String> {
45    let mut names = BTreeSet::from(["catalog.bin".to_string(), CATALOG_LSN_FILE.to_string()]);
46    for table in catalog.list_tables() {
47        names.insert(format!("{table}.heap"));
48        if let Some(indexes) = catalog.index_metadata(table) {
49            for index in indexes {
50                match index.source {
51                    IndexKeySource::Column { column } => {
52                        names.insert(format!("{table}_{column}.idx"));
53                    }
54                    IndexKeySource::Expression { index_id, .. } => {
55                        names.insert(expression_index_file_name(table, index_id));
56                    }
57                }
58            }
59        }
60    }
61    names
62}
63
64#[derive(Debug, Clone, Serialize, Deserialize)]
65pub struct FileEntry {
66    pub name: String,
67    pub len: u64,
68    pub blake3_hex: String,
69}
70
71#[derive(Debug, Clone, Serialize, Deserialize)]
72pub struct BackupManifest {
73    pub format_version: u32,
74    pub created_unix_secs: u64,
75    /// The page-LSN high-water mark this backup is consistent at.
76    pub source_lsn: u64,
77    /// Active on-disk catalog format at the snapshot boundary.
78    #[serde(default = "legacy_catalog_version")]
79    pub catalog_version: u16,
80    #[serde(default, skip_serializing_if = "Option::is_none")]
81    pub sync: Option<SyncSnapshotMetadata>,
82    pub files: Vec<FileEntry>,
83}
84
85impl BackupManifest {
86    pub const FORMAT_VERSION: u32 = 1;
87    pub const FILE_NAME: &'static str = "manifest.json";
88
89    pub fn validate_version(&self) -> io::Result<()> {
90        if self.format_version != Self::FORMAT_VERSION {
91            return Err(io::Error::other(format!(
92                "unsupported backup format {} (this build understands {})",
93                self.format_version,
94                Self::FORMAT_VERSION
95            )));
96        }
97        validate_catalog_version(self.catalog_version)?;
98        validate_sync_metadata(&self.sync, self.source_lsn, self.catalog_version)?;
99        Ok(())
100    }
101
102    pub fn write(&self, dir: &Path) -> io::Result<()> {
103        let json = serde_json::to_vec_pretty(self).map_err(io::Error::other)?;
104        std::fs::write(dir.join(Self::FILE_NAME), json)
105    }
106
107    pub fn read(dir: &Path) -> io::Result<Self> {
108        let bytes = std::fs::read(dir.join(Self::FILE_NAME))?;
109        let m: BackupManifest = serde_json::from_slice(&bytes).map_err(io::Error::other)?;
110        m.validate_version()?;
111        Ok(m)
112    }
113}
114
115/// A file that changed (relative to a base) in an incremental backup.
116#[derive(Debug, Clone, Serialize, Deserialize)]
117pub enum ChangedFile {
118    /// A small/unpaged file copied whole (e.g. catalog.bin).
119    Whole {
120        name: String,
121        len: u64,
122        blake3_hex: String,
123    },
124    /// A paged heap file: only pages whose LSN > base.source_lsn.
125    /// The sidecar delta file `<name>.delta` holds, for each listed page index
126    /// in order, a 4-byte LE page index followed by PAGE_SIZE bytes.
127    Pages {
128        name: String,
129        /// page count of the file at increment time
130        total_pages: u32,
131        /// which pages are in the delta (ascending)
132        page_indices: Vec<u32>,
133        /// `<name>.delta`
134        delta_file: String,
135        delta_len: u64,
136        delta_blake3_hex: String,
137    },
138}
139
140#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
141pub struct SyncSnapshotMetadata {
142    pub identity: IdentitySnapshot,
143    pub source_lsn: u64,
144    pub catalog_blake3_hex: String,
145    pub wal_format_version: u16,
146    pub catalog_version: u16,
147    pub retained_segment_format_version: u16,
148}
149
150impl SyncSnapshotMetadata {
151    pub fn current(
152        identity: IdentitySnapshot,
153        source_lsn: u64,
154        catalog_blake3_hex: String,
155        catalog_version: u16,
156    ) -> Self {
157        Self {
158            identity,
159            source_lsn,
160            catalog_blake3_hex,
161            wal_format_version: WAL_FORMAT_VERSION,
162            catalog_version,
163            retained_segment_format_version: RETAINED_SEGMENT_FORMAT_VERSION,
164        }
165    }
166
167    pub fn validate_against_source_lsn(&self, source_lsn: u64) -> io::Result<()> {
168        self.identity.validate()?;
169        if self.source_lsn != source_lsn {
170            return Err(io::Error::new(
171                io::ErrorKind::InvalidData,
172                format!(
173                    "sync snapshot source_lsn {} does not match manifest source_lsn {}",
174                    self.source_lsn, source_lsn
175                ),
176            ));
177        }
178        if self.catalog_blake3_hex.is_empty() {
179            return Err(io::Error::new(
180                io::ErrorKind::InvalidData,
181                "sync snapshot catalog hash must be non-empty",
182            ));
183        }
184        if self.wal_format_version != WAL_FORMAT_VERSION {
185            return Err(io::Error::new(
186                io::ErrorKind::InvalidData,
187                format!(
188                    "unsupported sync snapshot WAL format {}",
189                    self.wal_format_version
190                ),
191            ));
192        }
193        validate_catalog_version(self.catalog_version)?;
194        if self.retained_segment_format_version != RETAINED_SEGMENT_FORMAT_VERSION {
195            return Err(io::Error::new(
196                io::ErrorKind::InvalidData,
197                format!(
198                    "unsupported retained segment format {}",
199                    self.retained_segment_format_version
200                ),
201            ));
202        }
203        Ok(())
204    }
205}
206
207/// Manifest for an incremental (page-LSN diff) backup.
208#[derive(Debug, Clone, Serialize, Deserialize)]
209pub struct IncrementManifest {
210    /// reuse FORMAT_VERSION = 1
211    pub format_version: u32,
212    pub created_unix_secs: u64,
213    /// the source_lsn of the base (full or prior increment) this builds on
214    pub base_source_lsn: u64,
215    /// high-water mark after this increment (== catalog.max_lsn())
216    pub source_lsn: u64,
217    /// Active on-disk catalog format after applying this increment.
218    #[serde(default = "legacy_catalog_version")]
219    pub catalog_version: u16,
220    #[serde(default, skip_serializing_if = "Option::is_none")]
221    pub sync: Option<SyncSnapshotMetadata>,
222    pub changed: Vec<ChangedFile>,
223}
224
225impl IncrementManifest {
226    pub const FORMAT_VERSION: u32 = 1;
227    pub const FILE_NAME: &'static str = "increment.json";
228
229    pub fn validate_version(&self) -> io::Result<()> {
230        if self.format_version != Self::FORMAT_VERSION {
231            return Err(io::Error::other(format!(
232                "unsupported increment format {} (this build understands {})",
233                self.format_version,
234                Self::FORMAT_VERSION
235            )));
236        }
237        validate_catalog_version(self.catalog_version)?;
238        validate_sync_metadata(&self.sync, self.source_lsn, self.catalog_version)?;
239        Ok(())
240    }
241
242    pub fn write(&self, dir: &Path) -> io::Result<()> {
243        let json = serde_json::to_vec_pretty(self).map_err(io::Error::other)?;
244        std::fs::write(dir.join(Self::FILE_NAME), json)
245    }
246
247    pub fn read(dir: &Path) -> io::Result<Self> {
248        let bytes = std::fs::read(dir.join(Self::FILE_NAME))?;
249        let m: IncrementManifest = serde_json::from_slice(&bytes).map_err(io::Error::other)?;
250        m.validate_version()?;
251        Ok(m)
252    }
253}
254
255pub(crate) fn current_sync_snapshot_metadata(
256    data_dir: &Path,
257    source_lsn: u64,
258    catalog_version: u16,
259) -> io::Result<Option<SyncSnapshotMetadata>> {
260    let Some(identity) = powdb_sync::read_identity_snapshot_if_exists(data_dir)? else {
261        return Ok(None);
262    };
263    let catalog_bytes = std::fs::read(data_dir.join("catalog.bin"))?;
264    let catalog_blake3_hex = blake3::hash(&catalog_bytes).to_hex().to_string();
265    Ok(Some(SyncSnapshotMetadata::current(
266        identity,
267        source_lsn,
268        catalog_blake3_hex,
269        catalog_version,
270    )))
271}
272
273fn validate_sync_metadata(
274    metadata: &Option<SyncSnapshotMetadata>,
275    source_lsn: u64,
276    catalog_version: u16,
277) -> io::Result<()> {
278    if let Some(metadata) = metadata {
279        metadata.validate_against_source_lsn(source_lsn)?;
280        if metadata.catalog_version != catalog_version {
281            return Err(io::Error::new(
282                io::ErrorKind::InvalidData,
283                "sync snapshot catalog format does not match manifest catalog format",
284            ));
285        }
286    }
287    Ok(())
288}
289
290#[cfg(test)]
291mod tests {
292    use super::*;
293    #[test]
294    fn manifest_round_trips_and_rejects_bad_version() {
295        let m = BackupManifest {
296            format_version: BackupManifest::FORMAT_VERSION,
297            created_unix_secs: 1_700_000_000,
298            source_lsn: 42,
299            catalog_version: LEGACY_CATALOG_VERSION,
300            sync: None,
301            files: vec![FileEntry {
302                name: "catalog.bin".into(),
303                len: 10,
304                blake3_hex: "ab".into(),
305            }],
306        };
307        let json = serde_json::to_string(&m).unwrap();
308        let back: BackupManifest = serde_json::from_str(&json).unwrap();
309        assert_eq!(back.source_lsn, 42);
310        assert_eq!(back.files.len(), 1);
311
312        let mut bad = m.clone();
313        bad.format_version = 999;
314        assert!(
315            bad.validate_version().is_err(),
316            "unknown format must be rejected"
317        );
318    }
319}
powdb_backup/manifest.rs

powdb_backup/
manifest.rs
