[PATCH proxmox-backup v7 2/9] datastore: add move-group

[Hannes Laimer <h.laimer@proxmox.com>]

Add support for moving a single backup group to a different namespace
within the same datastore.

For the filesystem backend each snapshot directory is renamed
individually. For S3 all objects are copied to the target prefix
before deleting the source, per snapshot.

Exclusive locks on the group and all its snapshots are acquired
before the move to ensure no concurrent operations are active.
Snapshots are locked and moved in batches to avoid exhausting file
descriptors on groups with many snapshots.

If the target group already exists and merging is enabled, the
individual snapshots are moved into the existing target group
provided both groups have the same owner and the source snapshots
are strictly newer than the target snapshots.

Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
 pbs-datastore/src/backup_info.rs | 165 +++++++++++++++++++++++++++-
 pbs-datastore/src/datastore.rs   | 177 +++++++++++++++++++++++++++++++
 2 files changed, 340 insertions(+), 2 deletions(-)

diff --git a/pbs-datastore/src/backup_info.rs b/pbs-datastore/src/backup_info.rs
index c33eb307..46dd9af1 100644
--- a/pbs-datastore/src/backup_info.rs
+++ b/pbs-datastore/src/backup_info.rs
@@ -9,7 +9,7 @@ use std::time::Duration;
 use anyhow::{bail, format_err, Context, Error};
 use const_format::concatcp;
 
-use proxmox_s3_client::S3PathPrefix;
+use proxmox_s3_client::{S3ObjectKey, S3PathPrefix};
 use proxmox_sys::fs::{lock_dir_noblock, lock_dir_noblock_shared, replace_file, CreateOptions};
 use proxmox_systemd::escape_unit;
 
@@ -273,8 +273,169 @@ impl BackupGroup {
         Ok(delete_stats)
     }
 
+    /// Check merge invariants for moving this group's snapshots into an existing
+    /// target group. Returns an error if ownership differs or snapshot times overlap.
+    pub(crate) fn check_merge_invariants(&self, target_ns: &BackupNamespace) -> Result<(), Error> {
+        let target_group = BackupGroup {
+            store: self.store.clone(),
+            ns: target_ns.clone(),
+            group: self.group.clone(),
+        };
+
+        let src_owner = self.get_owner()?;
+        let tgt_owner = target_group.get_owner()?;
+        if src_owner != tgt_owner {
+            bail!(
+                "cannot merge group '{}/{}': owner mismatch (source: {src_owner}, \
+                target: {tgt_owner})",
+                self.group.ty,
+                self.group.id,
+            );
+        }
+
+        let src_oldest = self
+            .iter_snapshots()?
+            .collect::<Result<Vec<_>, _>>()?
+            .iter()
+            .map(|s| s.backup_time())
+            .min();
+        let tgt_newest = target_group
+            .iter_snapshots()?
+            .collect::<Result<Vec<_>, _>>()?
+            .iter()
+            .map(|s| s.backup_time())
+            .max();
+        if let (Some(src_oldest), Some(tgt_newest)) = (src_oldest, tgt_newest) {
+            if src_oldest <= tgt_newest {
+                bail!(
+                    "cannot merge group '{}/{}': snapshot time overlap \
+                    (oldest source: {src_oldest}, newest target: {tgt_newest})",
+                    self.group.ty,
+                    self.group.id,
+                );
+            }
+        }
+
+        Ok(())
+    }
+
+    /// Move a single snapshot to a target group path.
+    ///
+    /// For the filesystem backend, renames the snapshot directory. For S3, copies all
+    /// objects under the snapshot prefix to the target, renames the local cache directory,
+    /// then deletes the source objects. A copy failure returns an error with the snapshot
+    /// intact at source. A delete failure is logged as a warning.
+    ///
+    /// The caller must hold an exclusive lock on the snapshot being moved.
+    pub(crate) fn move_snapshot(
+        &self,
+        snap: &BackupDir,
+        target_group_path: &Path,
+        backend: &DatastoreBackend,
+    ) -> Result<(), Error> {
+        let src_snap_path = snap.full_path();
+        let dst_snap_path = target_group_path.join(snap.backup_time_string());
+
+        match backend {
+            DatastoreBackend::Filesystem => {
+                std::fs::rename(&src_snap_path, &dst_snap_path).with_context(|| {
+                    format!("failed to move snapshot {src_snap_path:?} to {dst_snap_path:?}")
+                })?;
+            }
+            DatastoreBackend::S3(s3_client) => {
+                let src_rel = snap.relative_path();
+                let src_rel_str = src_rel
+                    .to_str()
+                    .ok_or_else(|| format_err!("invalid source snapshot path"))?;
+                let src_prefix_str = format!("{S3_CONTENT_PREFIX}/{src_rel_str}/");
+
+                let dst_snap_rel = dst_snap_path
+                    .strip_prefix(self.store.base_path())
+                    .with_context(|| {
+                        format!(
+                            "target snapshot path {dst_snap_path:?} does not start with \
+                            datastore base path {:?}",
+                            self.store.base_path(),
+                        )
+                    })?;
+                let dst_rel_str = dst_snap_rel
+                    .to_str()
+                    .ok_or_else(|| format_err!("invalid target snapshot path"))?;
+                let dst_prefix_str = format!("{S3_CONTENT_PREFIX}/{dst_rel_str}/");
+
+                let store_prefix = format!("{}/", self.store.name());
+
+                // Copy all objects for this snapshot to the target prefix. On
+                // failure the source snapshot remains intact and any partial
+                // target copies stay as leftovers (visible via the API for cleanup).
+                let prefix = S3PathPrefix::Some(src_prefix_str.clone());
+                let mut token: Option<String> = None;
+                let mut src_keys = Vec::new();
+
+                loop {
+                    let result = proxmox_async::runtime::block_on(
+                        s3_client.list_objects_v2(&prefix, token.as_deref()),
+                    )
+                    .context("failed to list snapshot objects on S3 backend")?;
+
+                    for item in result.contents {
+                        let full_key_str: &str = &item.key;
+                        let rel_key =
+                            full_key_str.strip_prefix(&store_prefix).ok_or_else(|| {
+                                format_err!("unexpected key prefix in '{full_key_str}'")
+                            })?;
+                        let src_key = S3ObjectKey::try_from(rel_key)?;
+
+                        let suffix = rel_key
+                            .strip_prefix(&src_prefix_str)
+                            .ok_or_else(|| format_err!("unexpected key format '{rel_key}'"))?;
+                        let dst_key_str = format!("{dst_prefix_str}{suffix}");
+                        let dst_key = S3ObjectKey::try_from(dst_key_str.as_str())?;
+
+                        proxmox_async::runtime::block_on(
+                            s3_client.copy_object(src_key.clone(), dst_key),
+                        )
+                        .with_context(|| format!("failed to copy S3 object '{rel_key}'"))?;
+                        src_keys.push(src_key);
+                    }
+
+                    if result.is_truncated {
+                        token = result.next_continuation_token;
+                    } else {
+                        break;
+                    }
+                }
+
+                // Rename local cache directory.
+                std::fs::rename(&src_snap_path, &dst_snap_path).with_context(|| {
+                    format!("failed to move snapshot cache {src_snap_path:?} to {dst_snap_path:?}")
+                })?;
+
+                // Delete source S3 objects. Treat failures as warnings since the
+                // snapshot is already at the target.
+                for src_key in src_keys {
+                    if let Err(err) =
+                        proxmox_async::runtime::block_on(s3_client.delete_object(src_key.clone()))
+                    {
+                        log::warn!(
+                            "S3 move: failed to delete source object '{src_key:?}' \
+                            (snapshot already at target, orphaned object requires manual removal): \
+                            {err:#}"
+                        );
+                    }
+                }
+            }
+        }
+
+        // Clean up stale source lock files under /run for this snapshot.
+        let _ = std::fs::remove_file(snap.manifest_lock_path());
+        let _ = std::fs::remove_file(snap.lock_path());
+
+        Ok(())
+    }
+
     /// Helper function, assumes that no more snapshots are present in the group.
-    fn remove_group_dir(&self) -> Result<(), Error> {
+    pub(crate) fn remove_group_dir(&self) -> Result<(), Error> {
         let note_path = self.store.group_notes_path(&self.ns, &self.group);
         if let Err(err) = std::fs::remove_file(&note_path) {
             if err.kind() != std::io::ErrorKind::NotFound {
diff --git a/pbs-datastore/src/datastore.rs b/pbs-datastore/src/datastore.rs
index f1475d77..f88c356e 100644
--- a/pbs-datastore/src/datastore.rs
+++ b/pbs-datastore/src/datastore.rs
@@ -88,6 +88,15 @@ const CHUNK_LOCK_TIMEOUT: Duration = Duration::from_secs(3 * 60 * 60);
 const S3_DELETE_BATCH_LIMIT: usize = 100;
 // max defer time for s3 batch deletions
 const S3_DELETE_DEFER_LIMIT_SECONDS: Duration = Duration::from_secs(60 * 5);
+// upper bound on concurrent snapshot locks during move operations to avoid exhausting FDs
+const MAX_SNAPSHOT_LOCKS: usize = 512;
+
+/// Error from a per-group move attempt. Soft errors (e.g. lock conflicts) are
+/// retryable, hard errors are not.
+pub(crate) enum MoveGroupError {
+    Soft(Error),
+    Hard(Error),
+}
 
 /// checks if auth_id is owner, or, if owner is a token, if
 /// auth_id is the user of the token
@@ -1129,6 +1138,174 @@ impl DataStore {
         backup_group.destroy(&self.backend()?)
     }
 
+    /// Move a single backup group to a different namespace within the same datastore.
+    pub fn move_group(
+        self: &Arc<Self>,
+        source_ns: &BackupNamespace,
+        group: &pbs_api_types::BackupGroup,
+        target_ns: &BackupNamespace,
+        merge_group: bool,
+    ) -> Result<(), Error> {
+        if *OLD_LOCKING {
+            bail!("move operations require new-style file-based locking");
+        }
+        if source_ns == target_ns {
+            bail!("source and target namespace must be different");
+        }
+
+        let source_group = self.backup_group(source_ns.clone(), group.clone());
+
+        // Pre-lock existence checks to avoid unnecessary locking overhead.
+        if !self.namespace_exists(target_ns) {
+            bail!("target namespace '{target_ns}' does not exist");
+        }
+        if !source_group.exists() {
+            bail!("group '{group}' does not exist in namespace '{source_ns}'");
+        }
+
+        let backend = self.backend()?;
+
+        self.lock_and_move_group(&source_group, source_ns, target_ns, &backend, merge_group)
+            .map_err(|err| match err {
+                MoveGroupError::Soft(err) | MoveGroupError::Hard(err) => err,
+            })
+    }
+
+    /// Lock and move a single group. Acquires exclusive locks on both source and target
+    /// group paths, then moves snapshots in batches - each batch is locked, moved, and
+    /// released before the next batch starts. This ensures no concurrent readers (e.g.
+    /// verify) can operate on snapshots being moved, without exhausting file descriptors.
+    ///
+    /// Returns a `MoveGroupError` on failure, indicating whether the failure was a
+    /// transient lock conflict (retryable) or a hard error.
+    pub(crate) fn lock_and_move_group(
+        self: &Arc<Self>,
+        group: &BackupGroup,
+        source_ns: &BackupNamespace,
+        target_ns: &BackupNamespace,
+        backend: &DatastoreBackend,
+        merge_groups: bool,
+    ) -> Result<(), MoveGroupError> {
+        let target_group_ns = group
+            .backup_ns()
+            .map_prefix(source_ns, target_ns)
+            .map_err(MoveGroupError::Hard)?;
+
+        let target_group = self.backup_group(target_group_ns.clone(), group.group().clone());
+
+        // Acquire exclusive group locks - source prevents new snapshot additions/removals,
+        // target prevents concurrent creation at the destination path.
+        let _group_lock = group.lock().map_err(MoveGroupError::Soft)?;
+        let _target_group_lock = target_group.lock().map_err(MoveGroupError::Soft)?;
+
+        let merge = target_group.exists();
+
+        if merge && !merge_groups {
+            return Err(MoveGroupError::Hard(format_err!(
+                "group '{}' already exists in target namespace '{target_group_ns}' \
+                and merging is disabled",
+                group.group(),
+            )));
+        }
+
+        if merge {
+            group
+                .check_merge_invariants(&target_group_ns)
+                .map_err(MoveGroupError::Hard)?;
+        }
+
+        let mut snapshots: Vec<_> = group
+            .iter_snapshots()
+            .map_err(MoveGroupError::Hard)?
+            .collect::<Result<Vec<_>, _>>()
+            .map_err(MoveGroupError::Hard)?;
+        // Sort by time so that a partial failure (e.g. batch N succeeds but N+1
+        // fails) leaves a clean time-ordered split: all moved snapshots are older
+        // than all remaining ones. This allows a merge retry to succeed since the
+        // invariant (source oldest > target newest) still holds.
+        snapshots.sort_by_key(|s| s.backup_time());
+
+        // Ensure the target type and group directories exist.
+        std::fs::create_dir_all(self.type_path(&target_group_ns, group.group().ty))
+            .map_err(|err| MoveGroupError::Hard(err.into()))?;
+        let target_group_path = self.group_path(&target_group_ns, group.group());
+        if !merge {
+            std::fs::create_dir_all(&target_group_path)
+                .map_err(|err| MoveGroupError::Hard(err.into()))?;
+            // Copy owner so the group is visible in the UI during the move (especially
+            // relevant for S3 where the copy can take a while).
+            if let Ok(owner) = group.get_owner() {
+                if let Err(err) = target_group.set_owner(&owner, true) {
+                    warn!(
+                        "failed to set owner for target group '{}' in '{target_group_ns}': {err:#}",
+                        group.group(),
+                    );
+                }
+            }
+        }
+
+        log::info!(
+            "{} group '{}/{}' from '{}' to '{target_group_ns}'",
+            if merge { "merging" } else { "moving" },
+            group.group().ty,
+            group.group().id,
+            group.backup_ns(),
+        );
+
+        // Lock and move snapshots in batches. Each batch is locked, moved, and released
+        // before the next batch starts. This ensures no concurrent reader (e.g. verify,
+        // which only takes a snapshot-level shared lock) can operate on a snapshot while
+        // it is being moved.
+        for chunk in snapshots.chunks(MAX_SNAPSHOT_LOCKS) {
+            let _batch_locks: Vec<_> = chunk
+                .iter()
+                .map(|snap| snap.lock().map_err(MoveGroupError::Soft))
+                .collect::<Result<Vec<_>, _>>()?;
+
+            for snap in chunk {
+                group
+                    .move_snapshot(snap, &target_group_path, backend)
+                    .map_err(MoveGroupError::Hard)?;
+            }
+            // batch locks released here
+        }
+
+        // Copy group notes to the target (unless merging into a group that already has notes).
+        let src_notes_path = self.group_notes_path(group.backup_ns(), group.group());
+        let dst_notes_path = self.group_notes_path(&target_group_ns, group.group());
+        if src_notes_path.exists() && !dst_notes_path.exists() {
+            if let Err(err) = std::fs::copy(&src_notes_path, &dst_notes_path) {
+                warn!(
+                    "failed to copy group notes from {src_notes_path:?} to {dst_notes_path:?}: {err}"
+                );
+            }
+        }
+
+        // Remove the now-empty source group directory (owner, notes, dir, group lock).
+        group.remove_group_dir().map_err(MoveGroupError::Hard)?;
+
+        // For S3: delete orphaned source group-level objects (owner, notes).
+        if let DatastoreBackend::S3(s3_client) = backend {
+            let src_group_rel = group.relative_group_path();
+            let src_group_rel_str = src_group_rel
+                .to_str()
+                .ok_or_else(|| MoveGroupError::Hard(format_err!("invalid source group path")))?;
+            for filename in [GROUP_OWNER_FILE_NAME, GROUP_NOTES_FILE_NAME] {
+                let key_str = format!("{S3_CONTENT_PREFIX}/{src_group_rel_str}/{filename}");
+                if let Ok(key) = S3ObjectKey::try_from(key_str.as_str()) {
+                    if let Err(err) = proxmox_async::runtime::block_on(s3_client.delete_object(key))
+                    {
+                        log::warn!(
+                            "S3 move: failed to delete source group object '{key_str}': {err:#}"
+                        );
+                    }
+                }
+            }
+        }
+
+        Ok(())
+    }
+
     /// Remove a backup directory including all content
     pub fn remove_backup_dir(
         self: &Arc<Self>,
-- 
2.47.3