powdb_backup/
manifest.rs

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

powdb_backup/
manifest.rs
