[PATCH proxmox-backup] sync: pull: use LogLineSender in d SyncSourceReader

[PATCH proxmox-backup] sync: pull: use LogLineSender in d SyncSourceReader

[Dominik Csapak]

otherwise some log messages are not rendered with the correct prefix
when using multiple worker threads.

For example:

Snapshot ct/122/2026-04-24T12:56:17Z: got backup log file client.log.blob

vs

[ct/122]: Snapshot ct/122/2026-04-24T13:00:02Z: got backup log file client.log.blob

Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
---
there may be other places where we don't use the LogLineSender here,
but could not find any on first glance.

 src/server/pull.rs | 10 ++++---
 src/server/sync.rs | 67 +++++++++++++++++++++++++++++++++++-----------
 2 files changed, 59 insertions(+), 18 deletions(-)

diff --git a/src/server/pull.rs b/src/server/pull.rs
index 7fa273edb..2dff2a9f8 100644
--- a/src/server/pull.rs
+++ b/src/server/pull.rs
@@ -450,7 +450,7 @@ async fn pull_single_archive<'a>(
         .await?;
 
     reader
-        .load_file_into(archive_name, &tmp_path)
+        .load_file_into(archive_name, &tmp_path, Arc::clone(&log_sender))
         .await
         .with_context(|| archive_prefix.clone())?;
 
@@ -664,7 +664,11 @@ async fn pull_snapshot<'a>(
     let mut tmp_manifest_name = manifest_name.clone();
     tmp_manifest_name.set_extension("tmp");
     let Some(tmp_manifest_blob) = reader
-        .load_file_into(MANIFEST_BLOB_NAME.as_ref(), &tmp_manifest_name)
+        .load_file_into(
+            MANIFEST_BLOB_NAME.as_ref(),
+            &tmp_manifest_name,
+            Arc::clone(&log_sender),
+        )
         .await
         .with_context(|| prefix.clone())?
     else {
@@ -676,7 +680,7 @@ async fn pull_snapshot<'a>(
     let fetch_log = async || {
         if !client_log_name.exists() {
             reader
-                .try_download_client_log(&client_log_name)
+                .try_download_client_log(&client_log_name, Arc::clone(&log_sender))
                 .await
                 .with_context(|| prefix.clone())?;
             if client_log_name.exists() {
diff --git a/src/server/sync.rs b/src/server/sync.rs
index d25180341..8acbaf673 100644
--- a/src/server/sync.rs
+++ b/src/server/sync.rs
@@ -101,10 +101,19 @@ pub(crate) trait SyncSourceReader: Send + Sync {
     /// Asynchronously loads a file from the source into a local file.
     /// `filename` is the name of the file to load from the source.
     /// `into` is the path of the local file to load the source file into.
-    async fn load_file_into(&self, filename: &str, into: &Path) -> Result<Option<DataBlob>, Error>;
+    async fn load_file_into(
+        &self,
+        filename: &str,
+        into: &Path,
+        log_sender: Arc<LogLineSender>,
+    ) -> Result<Option<DataBlob>, Error>;
 
     /// Tries to download the client log from the source and save it into a local file.
-    async fn try_download_client_log(&self, to_path: &Path) -> Result<(), Error>;
+    async fn try_download_client_log(
+        &self,
+        to_path: &Path,
+        log_sender: Arc<LogLineSender>,
+    ) -> Result<(), Error>;
 
     fn skip_chunk_sync(&self, target_store_name: &str) -> bool;
 }
@@ -137,7 +146,12 @@ impl SyncSourceReader for RemoteSourceReader {
         Ok(Arc::new(chunk_reader))
     }
 
-    async fn load_file_into(&self, filename: &str, into: &Path) -> Result<Option<DataBlob>, Error> {
+    async fn load_file_into(
+        &self,
+        filename: &str,
+        into: &Path,
+        log_sender: Arc<LogLineSender>,
+    ) -> Result<Option<DataBlob>, Error> {
         let mut tmp_file = std::fs::OpenOptions::new()
             .write(true)
             .create(true)
@@ -149,10 +163,15 @@ impl SyncSourceReader for RemoteSourceReader {
             match err.downcast_ref::<HttpError>() {
                 Some(HttpError { code, message }) => match *code {
                     StatusCode::NOT_FOUND => {
-                        info!(
-                            "Snapshot {}: skipped because vanished since start of sync",
-                            &self.dir
-                        );
+                        log_sender
+                            .log(
+                                Level::INFO,
+                                format!(
+                                    "Snapshot {}: skipped because vanished since start of sync",
+                                    &self.dir
+                                ),
+                            )
+                            .await?;
                         return Ok(None);
                     }
                     _ => {
@@ -168,7 +187,11 @@ impl SyncSourceReader for RemoteSourceReader {
         Ok(DataBlob::load_from_reader(&mut tmp_file).ok())
     }
 
-    async fn try_download_client_log(&self, to_path: &Path) -> Result<(), Error> {
+    async fn try_download_client_log(
+        &self,
+        to_path: &Path,
+        log_sender: Arc<LogLineSender>,
+    ) -> Result<(), Error> {
         let mut tmp_path = to_path.to_owned();
         tmp_path.set_extension("tmp");
 
@@ -189,11 +212,16 @@ impl SyncSourceReader for RemoteSourceReader {
             if let Err(err) = std::fs::rename(&tmp_path, to_path) {
                 bail!("Atomic rename file {to_path:?} failed - {err}");
             }
-            info!(
-                "Snapshot {snapshot}: got backup log file {client_log_name}",
-                snapshot = &self.dir,
-                client_log_name = client_log_name.deref()
-            );
+            log_sender
+                .log(
+                    Level::INFO,
+                    format!(
+                        "Snapshot {snapshot}: got backup log file {client_log_name}",
+                        snapshot = &self.dir,
+                        client_log_name = client_log_name.deref()
+                    ),
+                )
+                .await?;
         }
 
         Ok(())
@@ -215,7 +243,12 @@ impl SyncSourceReader for LocalSourceReader {
         Ok(Arc::new(chunk_reader))
     }
 
-    async fn load_file_into(&self, filename: &str, into: &Path) -> Result<Option<DataBlob>, Error> {
+    async fn load_file_into(
+        &self,
+        filename: &str,
+        into: &Path,
+        _log_sender: Arc<LogLineSender>,
+    ) -> Result<Option<DataBlob>, Error> {
         let mut tmp_file = std::fs::OpenOptions::new()
             .write(true)
             .create(true)
@@ -229,7 +262,11 @@ impl SyncSourceReader for LocalSourceReader {
         Ok(DataBlob::load_from_reader(&mut tmp_file).ok())
     }
 
-    async fn try_download_client_log(&self, _to_path: &Path) -> Result<(), Error> {
+    async fn try_download_client_log(
+        &self,
+        _to_path: &Path,
+        _log_sender: Arc<LogLineSender>,
+    ) -> Result<(), Error> {
         Ok(())
     }
 
-- 
2.47.3

Re: [PATCH proxmox-backup] sync: pull: use LogLineSender in d SyncSourceReader

[Dominik Csapak]

the 'd' in the commit subject is a typo. can resend if necessary or 
please just fix up on applying (if it gets applied), thanks!