[pbs-devel] [PATCH v6 proxmox-backup 23/29] catalog: use format version 2 conditionally

[Christian Ebner <c.ebner@proxmox.com>]

Only use the catalog format version 2 when performing backups with the
change detection mode set to metadata.

This makes sure to remain compatible with older clients which do not
support the format version 2 at least for backups created using the
regular mode.

Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
---
Changes since v5:
- fix formatting using `cargo fmt`
- changed patch ordering for more logical flow

 pbs-datastore/src/catalog.rs      | 69 +++++++++++++++++++++++++------
 proxmox-backup-client/src/main.rs | 30 ++++++++++----
 2 files changed, 80 insertions(+), 19 deletions(-)

diff --git a/pbs-datastore/src/catalog.rs b/pbs-datastore/src/catalog.rs
index 220313c6..0ca2ce23 100644
--- a/pbs-datastore/src/catalog.rs
+++ b/pbs-datastore/src/catalog.rs
@@ -64,6 +64,12 @@ pub enum CatalogEntryType {
     Socket = b's',
 }
 
+#[derive(PartialEq)]
+pub enum CatalogFormatVersion {
+    V1,
+    V2,
+}
+
 impl TryFrom<u8> for CatalogEntryType {
     type Error = Error;
 
@@ -555,17 +561,22 @@ pub struct CatalogWriter<W> {
     writer: W,
     dirstack: Vec<DirInfo>,
     pos: u64,
+    version: CatalogFormatVersion,
 }
 
 impl<W: Write> CatalogWriter<W> {
     /// Create a new  CatalogWriter instance
-    pub fn new(writer: W) -> Result<Self, Error> {
+    pub fn new(writer: W, version: CatalogFormatVersion) -> Result<Self, Error> {
         let mut me = Self {
             writer,
             dirstack: vec![DirInfo::new_rootdir()],
             pos: 0,
+            version,
         };
-        me.write_all(&PROXMOX_CATALOG_FILE_MAGIC_2_0)?;
+        match me.version {
+            CatalogFormatVersion::V1 => me.write_all(&PROXMOX_CATALOG_FILE_MAGIC_1_0)?,
+            CatalogFormatVersion::V2 => me.write_all(&PROXMOX_CATALOG_FILE_MAGIC_2_0)?,
+        }
         Ok(me)
     }
 
@@ -599,6 +610,10 @@ impl<W: Write> CatalogWriter<W> {
 
 impl<W: Write> BackupCatalogWriter for CatalogWriter<W> {
     fn start_archive(&mut self, name: &CStr) -> Result<(), Error> {
+        if self.version == CatalogFormatVersion::V1 {
+            return self.start_directory(name);
+        }
+
         let new = DirInfo::new(name.to_owned());
         self.dirstack.push(new);
         Ok(())
@@ -608,6 +623,9 @@ impl<W: Write> BackupCatalogWriter for CatalogWriter<W> {
         &mut self,
         appendix: Option<pxar::encoder::AppendixStartOffset>,
     ) -> Result<(), Error> {
+        if self.version == CatalogFormatVersion::V1 {
+            return self.end_directory();
+        }
         let (start, name) = match self.dirstack.pop() {
             Some(dir) => {
                 let start = self.pos;
@@ -688,17 +706,30 @@ impl<W: Write> BackupCatalogWriter for CatalogWriter<W> {
             .last_mut()
             .ok_or_else(|| format_err!("outside root"))?;
         let name = name.to_bytes().to_vec();
-        let file_offset = FileOffset {
-            offset: file_offset.raw(),
-        };
-        dir.entries.push(DirEntry {
-            name,
-            attr: DirEntryAttribute::File {
-                size,
-                mtime,
-                extension: Some(CatalogV2Extension { ctime, file_offset }),
+        let entry = match self.version {
+            CatalogFormatVersion::V1 => DirEntry {
+                name,
+                attr: DirEntryAttribute::File {
+                    size,
+                    mtime,
+                    extension: None,
+                },
             },
-        });
+            CatalogFormatVersion::V2 => {
+                let file_offset = FileOffset {
+                    offset: file_offset.raw(),
+                };
+                DirEntry {
+                    name,
+                    attr: DirEntryAttribute::File {
+                        size,
+                        mtime,
+                        extension: Some(CatalogV2Extension { ctime, file_offset }),
+                    },
+                }
+            }
+        };
+        dir.entries.push(entry);
         Ok(())
     }
 
@@ -710,6 +741,9 @@ impl<W: Write> BackupCatalogWriter for CatalogWriter<W> {
         ctime: i64,
         appendix_ref_offset: pxar::encoder::AppendixRefOffset,
     ) -> Result<(), Error> {
+        if self.version == CatalogFormatVersion::V1 {
+            bail!("unsupported by catalog format version 1");
+        }
         let dir = self
             .dirstack
             .last_mut()
@@ -856,6 +890,17 @@ impl<R: Read + Seek> CatalogReader<R> {
         })
     }
 
+    pub fn format_version(&mut self) -> Result<CatalogFormatVersion, Error> {
+        self.reader.seek(SeekFrom::Start(0))?;
+        let mut magic = [0u8; 8];
+        self.reader.read_exact(&mut magic)?;
+        match magic {
+            PROXMOX_CATALOG_FILE_MAGIC_1_0 => Ok(CatalogFormatVersion::V1),
+            PROXMOX_CATALOG_FILE_MAGIC_2_0 => Ok(CatalogFormatVersion::V2),
+            _ => bail!("got unexpected magic number for catalog"),
+        }
+    }
+
     pub fn appendix_offset(
         &mut self,
         archive_name: &[u8],
diff --git a/proxmox-backup-client/src/main.rs b/proxmox-backup-client/src/main.rs
index 470399e8..0a3f24c0 100644
--- a/proxmox-backup-client/src/main.rs
+++ b/proxmox-backup-client/src/main.rs
@@ -49,7 +49,7 @@ use pbs_client::{
     BackupStats, BackupWriter, ChunkStream, FixedChunkStream, HttpClient, PxarBackupStream,
     RemoteChunkReader, UploadOptions, BACKUP_DETECTION_MODE_SPEC, BACKUP_SOURCE_SCHEMA,
 };
-use pbs_datastore::catalog::{CatalogReader, CatalogWriter};
+use pbs_datastore::catalog::{CatalogFormatVersion, CatalogReader, CatalogWriter};
 use pbs_datastore::chunk_store::verify_chunk_size;
 use pbs_datastore::dynamic_index::{BufferedDynamicReader, DynamicIndexReader};
 use pbs_datastore::fixed_index::FixedIndexReader;
@@ -536,6 +536,7 @@ struct CatalogUploadResult {
 fn spawn_catalog_upload(
     client: Arc<BackupWriter>,
     encrypt: bool,
+    catalog_format_version: CatalogFormatVersion,
 ) -> Result<CatalogUploadResult, Error> {
     let (catalog_tx, catalog_rx) = std::sync::mpsc::sync_channel(10); // allow to buffer 10 writes
     let catalog_stream = proxmox_async::blocking::StdChannelStream(catalog_rx);
@@ -549,9 +550,10 @@ fn spawn_catalog_upload(
         injections.clone(),
     );
 
-    let catalog_writer = Arc::new(Mutex::new(CatalogWriter::new(TokioWriterAdapter::new(
-        StdChannelWriter::new(catalog_tx),
-    ))?));
+    let catalog_writer = Arc::new(Mutex::new(CatalogWriter::new(
+        TokioWriterAdapter::new(StdChannelWriter::new(catalog_tx)),
+        catalog_format_version,
+    )?));
 
     let (catalog_result_tx, catalog_result_rx) = tokio::sync::oneshot::channel();
 
@@ -1015,8 +1017,17 @@ async fn create_backup(
             (BackupSpecificationType::PXAR, false) => {
                 // start catalog upload on first use
                 if catalog.is_none() {
-                    let catalog_upload_res =
-                        spawn_catalog_upload(client.clone(), crypto.mode == CryptMode::Encrypt)?;
+                    let catalog_format_version =
+                        if let BackupDetectionMode::Metadata(_) = detection_mode {
+                            CatalogFormatVersion::V2
+                        } else {
+                            CatalogFormatVersion::V1
+                        };
+                    let catalog_upload_res = spawn_catalog_upload(
+                        client.clone(),
+                        crypto.mode == CryptMode::Encrypt,
+                        catalog_format_version,
+                    )?;
                     catalog = Some(catalog_upload_res.catalog_writer);
                     catalog_result_rx = Some(catalog_upload_res.result);
                 }
@@ -1215,8 +1226,13 @@ async fn download_reference_catalog(
         .map_err(|err| format_err!("failed to download reference catalog - {}", err))?;
 
     catalogfile.seek(SeekFrom::Start(0))?;
+    let mut reader = CatalogReader::new(catalogfile);
 
-    Ok(CatalogReader::new(catalogfile))
+    match reader.format_version() {
+        Ok(CatalogFormatVersion::V2) => Ok(reader),
+        Ok(CatalogFormatVersion::V1) => Err(format_err!("unsupported catalog format version")),
+        Err(err) => Err(err),
+    }
 }
 
 async fn dump_image<W: Write>(
-- 
2.39.2