[PATCH proxmox-backup v2 28/43] sync: pull: set retention timestamp from sync job for new snapshots

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

The retention timestamp on the source is not relevant for the
retention on the target datastore. Therefore, drop any retention
timestamp present in the manifest when pulling from the source and
set the retention timestamp according to the target, if some.

The timestamp is calculated based on the configured retention
timespan relative to the start time of the sync job by pull
parameter construction.

During a resync, a corrupt snapshot is now as well not allowed
to be overwritten if it still is protected by a retention timespan.

Further the retention is only set when creating the snapshots, not
on any subsequent resync. By that also the last snapshot in the group
can only be protected by retention once.

When remove vanished is enabled, do not remove snapshots vanished
from the source if still protected by a retention locally.

Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
---
 src/api2/pull.rs   |  11 ++++-
 src/server/pull.rs | 117 ++++++++++++++++++++++++++++++++++++++++++---
 2 files changed, 120 insertions(+), 8 deletions(-)

diff --git a/src/api2/pull.rs b/src/api2/pull.rs
index 6b53c55b7..c197c161f 100644
--- a/src/api2/pull.rs
+++ b/src/api2/pull.rs
@@ -10,8 +10,8 @@ use pbs_api_types::{
     Authid, BackupNamespace, CRYPT_KEY_ID_SCHEMA, DATASTORE_SCHEMA, GROUP_FILTER_LIST_SCHEMA,
     GroupFilter, NS_MAX_DEPTH_REDUCED_SCHEMA, PRIV_DATASTORE_APPEND, PRIV_DATASTORE_BACKUP,
     PRIV_DATASTORE_PRUNE, PRIV_REMOTE_READ, REMOTE_ID_SCHEMA, REMOVE_VANISHED_BACKUPS_SCHEMA,
-    RESYNC_CORRUPT_SCHEMA, RateLimitConfig, SYNC_ENCRYPTED_ONLY_SCHEMA, SYNC_VERIFIED_ONLY_SCHEMA,
-    SYNC_WORKER_THREADS_SCHEMA, SyncJobConfig, TRANSFER_LAST_SCHEMA,
+    RESYNC_CORRUPT_SCHEMA, RateLimitConfig, RetentionTimespan, SYNC_ENCRYPTED_ONLY_SCHEMA,
+    SYNC_VERIFIED_ONLY_SCHEMA, SYNC_WORKER_THREADS_SCHEMA, SyncJobConfig, TRANSFER_LAST_SCHEMA,
 };
 use pbs_config::CachedUserInfo;
 use proxmox_rest_server::WorkerTask;
@@ -93,6 +93,7 @@ impl TryFrom<&SyncJobConfig> for PullParameters {
             sync_job.resync_corrupt,
             sync_job.worker_threads,
             sync_job.associated_key.clone(),
+            sync_job.retention_timespan.clone(),
         )
     }
 }
@@ -162,6 +163,10 @@ impl TryFrom<&SyncJobConfig> for PullParameters {
                 },
                 optional: true,
             },
+            "retention-timespan": {
+                type: RetentionTimespan,
+                optional: true,
+            },
         },
     },
     access: {
@@ -191,6 +196,7 @@ async fn pull(
     resync_corrupt: Option<bool>,
     worker_threads: Option<usize>,
     decryption_keys: Option<Vec<String>>,
+    retention_timespan: Option<RetentionTimespan>,
     rpcenv: &mut dyn RpcEnvironment,
 ) -> Result<String, Error> {
     let auth_id: Authid = rpcenv.get_auth_id().unwrap().parse()?;
@@ -233,6 +239,7 @@ async fn pull(
         resync_corrupt,
         worker_threads,
         decryption_keys,
+        retention_timespan,
     )?;
 
     // fixme: set to_stdout to false?
diff --git a/src/server/pull.rs b/src/server/pull.rs
index 33f75c9c5..c1531ebda 100644
--- a/src/server/pull.rs
+++ b/src/server/pull.rs
@@ -23,7 +23,7 @@ use pbs_api_types::{
     ArchiveType, Authid, BackupDir, BackupGroup, BackupNamespace, CLIENT_LOG_BLOB_NAME, CryptMode,
     Fingerprint, GroupFilter, MANIFEST_BLOB_NAME, MAX_NAMESPACE_DEPTH, Operation,
     PRIV_DATASTORE_APPEND, PRIV_DATASTORE_AUDIT, PRIV_DATASTORE_BACKUP, RateLimitConfig, Remote,
-    SnapshotListItem, VerifyState, print_store_and_ns,
+    RetentionTimespan, SnapshotListItem, VerifyState, print_store_and_ns,
 };
 use pbs_client::BackupRepository;
 use pbs_config::CachedUserInfo;
@@ -77,6 +77,8 @@ pub(crate) struct PullParameters {
     worker_threads: Option<usize>,
     /// Decryption key ids and configs to decrypt snapshots with matching key fingerprint
     crypt_configs: Vec<(String, Arc<CryptConfig>)>,
+    // Retention timestamp to be set for newly pulled snapshots on sync target.
+    retain_until: Option<i64>,
 }
 
 impl PullParameters {
@@ -99,6 +101,7 @@ impl PullParameters {
         resync_corrupt: Option<bool>,
         worker_threads: Option<usize>,
         decryption_keys: Option<Vec<String>>,
+        retention_timespan: Option<RetentionTimespan>,
     ) -> Result<Self, Error> {
         if let Some(max_depth) = max_depth {
             ns.check_max_depth(max_depth)?;
@@ -140,6 +143,10 @@ impl PullParameters {
 
         let group_filter = group_filter.unwrap_or_default();
 
+        let retain_until = retention_timespan
+            .map(|timespan| timespan.to_timestamp_from_systemtime())
+            .transpose()?;
+
         let crypt_configs = if let Some(key_ids) = &decryption_keys {
             let mut crypt_configs = Vec::with_capacity(key_ids.len());
             for key_id in key_ids {
@@ -167,6 +174,7 @@ impl PullParameters {
             resync_corrupt,
             worker_threads,
             crypt_configs,
+            retain_until,
         })
     }
 }
@@ -744,7 +752,8 @@ async fn pull_snapshot<'a>(
     };
 
     let mut manifest_data = tmp_manifest_blob.raw_data().to_vec();
-    let manifest = BackupManifest::try_from(tmp_manifest_blob).with_context(|| prefix.clone())?;
+    let mut manifest =
+        BackupManifest::try_from(tmp_manifest_blob).with_context(|| prefix.clone())?;
 
     if ignore_not_verified_or_encrypted(
         &manifest,
@@ -762,6 +771,27 @@ async fn pull_snapshot<'a>(
         return Ok(None);
     }
 
+    let target_retain_until = match &existing_target_manifest {
+        Some(existing_target_manifest) => existing_target_manifest.retention_timestamp()?,
+        None => params.retain_until,
+    };
+
+    let source_retain_until = manifest.retention_timestamp()?;
+    if let Some(unprotected) = manifest.unprotected.as_object_mut() {
+        unprotected.remove("retain-until");
+    } else {
+        bail!("Encountered manifest without 'unprotected' section.");
+    }
+
+    if let Some(retain_until) = target_retain_until {
+        manifest.modify_retention_timestamp(retain_until)?;
+    }
+
+    let active_retention = match &existing_target_manifest {
+        Some(existing_target_manifest) => existing_target_manifest.active_retention_lock()?,
+        None => None,
+    };
+
     let (crypt_config, new_manifest) = match optionally_use_decryption_key(
         Arc::clone(&params),
         &manifest,
@@ -785,6 +815,43 @@ async fn pull_snapshot<'a>(
         }
     };
 
+    // Detect snapshots whose manifest only differs from the source by the retention timestamp
+    // set on a previous sync, so they are not re-synced on every run. At this point the source
+    // manifest already carries the retention of the existing target manifest.
+    let mut content_unchanged = false;
+    if let Some(existing) = &existing_target_manifest {
+        let dummy = CryptConfig::new([0u8; 32])?;
+        if existing.signature(&dummy)? == manifest.signature(&dummy)? {
+            if existing.unprotected == manifest.unprotected {
+                fetch_log(crypt_config).await?;
+                cleanup().await?;
+                return Ok(Some(sync_stats));
+            }
+            // only the unprotected part changed, e.g. notes, which is fine to update even if
+            // the snapshot is retained
+            content_unchanged = true;
+        }
+    }
+
+    if let Some(retain_until) = active_retention.filter(|_| !content_unchanged) {
+        let reason = if corrupt {
+            "corrupt snapshot"
+        } else {
+            "changed snapshot"
+        };
+        log_sender
+            .log(
+                if corrupt { Level::WARN } else { Level::INFO },
+                format!(
+                    "{prefix}: skipped re-sync of {reason}, retention locked until '{}'",
+                    proxmox_time::epoch_to_rfc3339(retain_until)?,
+                ),
+            )
+            .await?;
+        let _ = std::fs::remove_file(&tmp_manifest_name);
+        return Ok(None);
+    }
+
     for item in manifest.files() {
         let path = snapshot.full_archive_path(&item.filename);
         if !corrupt && path.exists() {
@@ -852,7 +919,7 @@ async fn pull_snapshot<'a>(
         sync_stats.add(stats);
     }
 
-    if let Some(new_manifest) = new_manifest {
+    let rewritten_manifest_blob = if let Some(new_manifest) = new_manifest {
         let mut new_manifest = Arc::try_unwrap(new_manifest)
             .map_err(|_arc| {
                 format_err!("failed to take ownership of still referenced new manifest")
@@ -862,6 +929,7 @@ async fn pull_snapshot<'a>(
 
         // copy over notes ecc, but drop encryption key fingerprint and verify state, to be
         // reverified independent from the sync.
+        // Retention timestamp has already been set above, so do not drop it here.
         new_manifest.unprotected = manifest.unprotected.clone();
         if let Some(unprotected) = new_manifest.unprotected.as_object_mut() {
             // legacy field replaced by sync-source-signature
@@ -878,8 +946,19 @@ async fn pull_snapshot<'a>(
             new_manifest.set_sync_source_signature(expected.bytes())?;
         }
 
+        Some(new_manifest.to_data_blob(None)?)
+    } else if source_retain_until != target_retain_until {
+        // Only the retention timestamp in the unprotected section differs between source and
+        // target. Keep everything else as is, so the file list, the signature and the key
+        // fingerprint stay intact and restores with the encryption key keep working.
+
         // keep signature
-        let manifest_blob = new_manifest.to_data_blob(None)?;
+        Some(manifest.to_data_blob(None)?)
+    } else {
+        None
+    };
+
+    if let Some(manifest_blob) = rewritten_manifest_blob {
         // update contents to be uploaded to backend
         manifest_data = manifest_blob.raw_data().to_vec();
 
@@ -1407,12 +1486,38 @@ async fn pull_group(
             if source_snapshots.contains(&snapshot.backup_time()) {
                 continue;
             }
-            if snapshot.is_protected() {
+            let is_retained_result = snapshot
+                .load_manifest()
+                .and_then(|(manifest, _size)| manifest.active_retention_lock())
+                .map(|retain_until| retain_until.is_some());
+
+            let is_retained = match is_retained_result {
+                Ok(retained) => retained,
+                Err(err) => {
+                    log_sender
+                        .log(
+                            Level::WARN,
+                            format!(
+                                "{prefix}: failed checking retention for {}: {err}",
+                                snapshot.dir(),
+                            ),
+                        )
+                        .await?;
+                    // do not delete snapshot if unable to check its retention
+                    true
+                }
+            };
+            if snapshot.is_protected() || is_retained {
+                let reason = if snapshot.is_protected() {
+                    "protected"
+                } else {
+                    "retained"
+                };
                 log_sender
                     .log(
                         Level::INFO,
                         format!(
-                            "{prefix}: don't delete vanished snapshot {} (protected)",
+                            "{prefix}: don't delete vanished snapshot {} ({reason})",
                             snapshot.dir(),
                         ),
                     )
-- 
2.47.3