[PATCH proxmox-backup v8 09/10] sync: move in-progress snapshot filter to helper and use log line sender

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

Currently, in-progress snapshots are being filtered out from the list
of source snapshots by pre-filtering and logging skipped snapshots
after gathering the list.

For parallel sync jobs, logging now requires however to go through
the BufferedLogger, by sending the logs via the LogLineSender. This
however requires to await inside an async context, which cannot
happen within the filter_map() closure.

Therefore, factor out the filtering to a dedicated helper in order to
avoid pollution of the SyncSource trait with a completely unrelated
parameter and refactor the filtering within that helper so the
logging can happen in async context.

Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
---
 src/server/pull.rs |  3 ++-
 src/server/push.rs |  3 ++-
 src/server/sync.rs | 39 ++++++++++++++++++++++++++-------------
 3 files changed, 30 insertions(+), 15 deletions(-)

diff --git a/src/server/pull.rs b/src/server/pull.rs
index 97def85a5..47c568376 100644
--- a/src/server/pull.rs
+++ b/src/server/pull.rs
@@ -31,7 +31,7 @@ use pbs_tools::buffered_logger::{BufferedLogger, LogLineSender};
 use pbs_tools::sha::sha256;
 
 use super::sync::{
-    check_namespace_depth_limit, exclude_not_verified_or_encrypted,
+    check_namespace_depth_limit, exclude_not_verified_or_encrypted, filter_out_in_progress,
     ignore_not_verified_or_encrypted, LocalSource, RemoteSource, RemovedVanishedStats, SkipInfo,
     SkipReason, SyncSource, SyncSourceReader, SyncStats,
 };
@@ -732,6 +732,7 @@ async fn pull_group(
         .source
         .list_backup_snapshots(source_namespace, group)
         .await?;
+    raw_list = filter_out_in_progress(raw_list, Arc::clone(&log_sender)).await?;
     raw_list.sort_unstable_by_key(|a| a.backup.time);
 
     let target_ns = source_namespace.map_prefix(&params.source.get_ns(), &params.target.ns)?;
diff --git a/src/server/push.rs b/src/server/push.rs
index 2ff46211c..1fbb82ebe 100644
--- a/src/server/push.rs
+++ b/src/server/push.rs
@@ -34,7 +34,7 @@ use pbs_tools::buffered_logger::{BufferedLogger, LogLineSender};
 use proxmox_human_byte::HumanByte;
 
 use super::sync::{
-    check_namespace_depth_limit, exclude_not_verified_or_encrypted,
+    check_namespace_depth_limit, exclude_not_verified_or_encrypted, filter_out_in_progress,
     ignore_not_verified_or_encrypted, LocalSource, RemovedVanishedStats, SkipInfo, SkipReason,
     SyncSource, SyncStats,
 };
@@ -777,6 +777,7 @@ pub(crate) async fn push_group(
         .source
         .list_backup_snapshots(namespace, group)
         .await?;
+    snapshots = filter_out_in_progress(snapshots, Arc::clone(&log_sender)).await?;
     snapshots.sort_unstable_by_key(|a| a.backup.time);
 
     if snapshots.is_empty() {
diff --git a/src/server/sync.rs b/src/server/sync.rs
index 17ed4839f..4827dc3f2 100644
--- a/src/server/sync.rs
+++ b/src/server/sync.rs
@@ -13,7 +13,7 @@ use futures::{future::FutureExt, select};
 use hyper::http::StatusCode;
 use pbs_config::BackupLockGuard;
 use serde_json::json;
-use tracing::{info, warn};
+use tracing::{info, warn, Level};
 
 use proxmox_human_byte::HumanByte;
 use proxmox_rest_server::WorkerTask;
@@ -28,6 +28,7 @@ use pbs_client::{BackupReader, BackupRepository, HttpClient, RemoteChunkReader};
 use pbs_datastore::data_blob::DataBlob;
 use pbs_datastore::read_chunk::AsyncReadChunk;
 use pbs_datastore::{BackupManifest, DataStore, ListNamespacesRecursive, LocalChunkReader};
+use pbs_tools::buffered_logger::LogLineSender;
 
 use crate::backup::ListAccessibleBackupGroups;
 use crate::server::jobstate::Job;
@@ -375,18 +376,7 @@ impl SyncSource for RemoteSource {
 
         let mut result = self.client.get(&path, Some(args)).await?;
         let snapshot_list: Vec<SnapshotListItem> = serde_json::from_value(result["data"].take())?;
-        Ok(snapshot_list
-            .into_iter()
-            .filter_map(|item: SnapshotListItem| {
-                // in-progress backups can't be synced
-                if item.size.is_none() {
-                    info!("skipping snapshot {} - in-progress backup", item.backup);
-                    return None;
-                }
-
-                Some(item)
-            })
-            .collect::<Vec<SnapshotListItem>>())
+        Ok(snapshot_list)
     }
 
     fn get_ns(&self) -> BackupNamespace {
@@ -736,6 +726,29 @@ pub fn do_sync_job(
     Ok(upid_str)
 }
 
+pub(super) async fn filter_out_in_progress(
+    snapshots: Vec<SnapshotListItem>,
+    log_sender: Arc<LogLineSender>,
+) -> Result<Vec<SnapshotListItem>, Error> {
+    let mut filtered = Vec::with_capacity(snapshots.len());
+
+    for item in snapshots {
+        // in-progress backups can't be synced
+        if item.size.is_none() {
+            log_sender
+                .log(
+                    Level::INFO,
+                    format!("skipping snapshot {} - in-progress backup", item.backup),
+                )
+                .await?;
+        } else {
+            filtered.push(item);
+        }
+    }
+
+    Ok(filtered)
+}
+
 pub(super) fn ignore_not_verified_or_encrypted(
     manifest: &BackupManifest,
     snapshot: &BackupDir,
-- 
2.47.3