[PATCH proxmox-backup 09/20] sync: push: rewrite manifest instead of pushing pre-existing one

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

In preparation for being able to encrypt unencypted backup snapshots
during push sync jobs.

Previously the pre-existing manifest file was pushed to the remote
target since it did not require modifications and contained all the
files with the correct metadata. When encrypting, the files must
however be marked as encrypted by individually setting the crypt mode
and the manifest must be signed and the encryption key fingerprint
added to the unprotected part of the manifest.

Therefore, now recreate the manifest and update accordingly. To do
so, pushing of the index must return the full BackupStats, not just
the sync stats for accounting.

Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
---
 src/server/push.rs | 59 +++++++++++++++++++++++++++++++++-------------
 1 file changed, 43 insertions(+), 16 deletions(-)

diff --git a/src/server/push.rs b/src/server/push.rs
index 27c5b22d4..269a4c386 100644
--- a/src/server/push.rs
+++ b/src/server/push.rs
@@ -17,8 +17,8 @@ use pbs_api_types::{
     PRIV_REMOTE_DATASTORE_MODIFY, PRIV_REMOTE_DATASTORE_PRUNE,
 };
 use pbs_client::{
-    BackupRepository, BackupWriter, BackupWriterOptions, HttpClient, IndexType, MergedChunkInfo,
-    UploadOptions,
+    BackupRepository, BackupStats, BackupWriter, BackupWriterOptions, HttpClient, IndexType,
+    MergedChunkInfo, UploadOptions,
 };
 use pbs_config::CachedUserInfo;
 use pbs_datastore::data_blob::ChunkInfo;
@@ -26,7 +26,7 @@ use pbs_datastore::dynamic_index::DynamicIndexReader;
 use pbs_datastore::fixed_index::FixedIndexReader;
 use pbs_datastore::index::IndexFile;
 use pbs_datastore::read_chunk::AsyncReadChunk;
-use pbs_datastore::{DataStore, StoreProgress};
+use pbs_datastore::{BackupManifest, DataStore, StoreProgress};
 
 use super::sync::{
     check_namespace_depth_limit, exclude_not_verified_or_encrypted,
@@ -879,6 +879,7 @@ pub(crate) async fn push_snapshot(
 
     // Avoid double upload penalty by remembering already seen chunks
     let known_chunks = Arc::new(Mutex::new(HashSet::with_capacity(64 * 1024)));
+    let mut target_manifest = BackupManifest::new(snapshot.clone());
 
     for entry in source_manifest.files() {
         let mut path = backup_dir.full_path();
@@ -891,6 +892,12 @@ pub(crate) async fn push_snapshot(
                     let backup_stats = backup_writer
                         .upload_blob(file, archive_name.as_ref())
                         .await?;
+                    target_manifest.add_file(
+                        &archive_name,
+                        backup_stats.size,
+                        backup_stats.csum,
+                        entry.chunk_crypt_mode(),
+                    )?;
                     stats.add(SyncStats {
                         chunk_count: backup_stats.chunk_count as usize,
                         bytes: backup_stats.size as usize,
@@ -913,7 +920,7 @@ pub(crate) async fn push_snapshot(
                     let chunk_reader = reader
                         .chunk_reader(entry.chunk_crypt_mode())
                         .context("failed to get chunk reader")?;
-                    let sync_stats = push_index(
+                    let upload_stats = push_index(
                         &archive_name,
                         index,
                         chunk_reader,
@@ -922,7 +929,18 @@ pub(crate) async fn push_snapshot(
                         known_chunks.clone(),
                     )
                     .await?;
-                    stats.add(sync_stats);
+                    target_manifest.add_file(
+                        &archive_name,
+                        upload_stats.size,
+                        upload_stats.csum,
+                        entry.chunk_crypt_mode(),
+                    )?;
+                    stats.add(SyncStats {
+                        chunk_count: upload_stats.chunk_count as usize,
+                        bytes: upload_stats.size as usize,
+                        elapsed: upload_stats.duration,
+                        removed: None,
+                    });
                 }
                 ArchiveType::FixedIndex => {
                     if let Some(manifest) = upload_options.previous_manifest.as_ref() {
@@ -940,7 +958,7 @@ pub(crate) async fn push_snapshot(
                         .chunk_reader(entry.chunk_crypt_mode())
                         .context("failed to get chunk reader")?;
                     let size = index.index_bytes();
-                    let sync_stats = push_index(
+                    let upload_stats = push_index(
                         &archive_name,
                         index,
                         chunk_reader,
@@ -949,7 +967,18 @@ pub(crate) async fn push_snapshot(
                         known_chunks.clone(),
                     )
                     .await?;
-                    stats.add(sync_stats);
+                    target_manifest.add_file(
+                        &archive_name,
+                        upload_stats.size,
+                        upload_stats.csum,
+                        entry.chunk_crypt_mode(),
+                    )?;
+                    stats.add(SyncStats {
+                        chunk_count: upload_stats.chunk_count as usize,
+                        bytes: upload_stats.size as usize,
+                        elapsed: upload_stats.duration,
+                        removed: None,
+                    });
                 }
             }
         } else {
@@ -972,8 +1001,11 @@ pub(crate) async fn push_snapshot(
             .await?;
     }
 
-    // Rewrite manifest for pushed snapshot, recreating manifest from source on target
-    let manifest_json = serde_json::to_value(source_manifest)?;
+    // Rewrite manifest for pushed snapshot, recreating manifest from source on target,
+    // needs to update all relevant info for new manifest.
+    target_manifest.unprotected = source_manifest.unprotected;
+    target_manifest.signature = source_manifest.signature;
+    let manifest_json = serde_json::to_value(target_manifest)?;
     let manifest_string = serde_json::to_string_pretty(&manifest_json)?;
     let backup_stats = backup_writer
         .upload_blob_from_data(
@@ -1005,7 +1037,7 @@ async fn push_index(
     backup_writer: &BackupWriter,
     index_type: IndexType,
     known_chunks: Arc<Mutex<HashSet<[u8; 32]>>>,
-) -> Result<SyncStats, Error> {
+) -> Result<BackupStats, Error> {
     let (upload_channel_tx, upload_channel_rx) = mpsc::channel(20);
     let mut chunk_infos =
         stream::iter(0..index.index_count()).map(move |pos| index.chunk_info(pos).unwrap());
@@ -1057,10 +1089,5 @@ async fn push_index(
         .upload_index_chunk_info(filename, merged_chunk_info_stream, upload_options)
         .await?;
 
-    Ok(SyncStats {
-        chunk_count: upload_stats.chunk_count as usize,
-        bytes: upload_stats.size as usize,
-        elapsed: upload_stats.duration,
-        removed: None,
-    })
+    Ok(upload_stats)
 }
-- 
2.47.3