[pbs-devel] [PATCH proxmox-backup 00/17] fix chunk upload/insert, rename corrupt chunks and GC race conditions for s3 backend

[pbs-devel] [PATCH proxmox-backup 00/17] fix chunk upload/insert, rename corrupt chunks and GC race conditions for s3 backend

[Christian Ebner]

These patches fix possible race conditions on datastores with s3 backend for
chunk insert, renaming of corrupt chunks during verification and cleanup during
garbage collection. Further, the patches assure consistency between the chunk
marker file of the local datastore cache, the s3 object store and the in-memory
LRU cache during state changes occurring by one of the above mentioned operations.

Consistency is achieved by using a per-chunk file locking mechanism. File locks
are stored on the predefined location for datastore file locks, using the same
.chunks/prefix/digest folder layout for consistency and to keep readdir and other
fs operations performant.

Before introducing the file locking mechanism, the patches refactor pre-existing
code to move most of the backend related logic away from the api code to the
datastore implementation, in order to have a common interface especially for
chunk insert.

As part of the series it is now also assured that chunks which are removed from
the local datastore cache, are also dropped from it's in-memory LRU cache and'
therefore a consistent state is achieved.

This patch series obsoletes two previous patch series with unfortunately
incomplete bugfix attempts found at:
- https://lore.proxmox.com/pbs-devel/8d711a20-b193-47a9-8f38-6ce800e6d0e8@proxmox.com/T/
- https://lore.proxmox.com/pbs-devel/20251015164008.975591-1-c.ebner@proxmox.com/T/

proxmox-backup:

Christian Ebner (17):
  sync: pull: instantiate backend only once per sync job
  api/datastore: move group notes setting to the datastore
  api/datastore: move snapshot deletion into dedicated datastore helper
  api/datastore: move backup log upload by implementing datastore helper
  api/datastore: add dedicated datastore helper to set snapshot notes
  datastore: refactor chunk insert based on backend
  verify: rename corrupted to corrupt in log output and function names
  verify/datastore: make rename corrupt chunk a datastore helper method
  datastore: refactor rename_corrupt_chunk error handling
  datastore: implement per-chunk file locking helper for s3 backend
  datastore: acquire chunk store mutex lock when renaming corrupt chunk
  datastore: get per-chunk file lock for chunk rename on s3 backend
  fix #6961: datastore: verify: evict corrupt chunks from in-memory LRU
    cache
  datastore: add locking to protect against races on chunk insert for s3
  GC: fix race with chunk upload/insert on s3 backends
  GC: lock chunk marker before cleanup in phase 3 on s3 backends
  datastore: GC: drop overly verbose info message during s3 chunk sweep

 pbs-datastore/src/backup_info.rs |   2 +-
 pbs-datastore/src/chunk_store.rs |  54 ++++++-
 pbs-datastore/src/datastore.rs   | 262 +++++++++++++++++++++++++++++--
 src/api2/admin/datastore.rs      |  74 +++------
 src/api2/backup/upload_chunk.rs  |  64 ++------
 src/api2/tape/restore.rs         |   6 +-
 src/backup/verify.rs             |  83 ++--------
 src/server/pull.rs               |  47 +++---
 8 files changed, 369 insertions(+), 223 deletions(-)


Summary over all repositories:
  8 files changed, 369 insertions(+), 223 deletions(-)

-- 
Generated by git-murpp 0.8.1


_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

[pbs-devel] [PATCH proxmox-backup 01/17] sync: pull: instantiate backend only once per sync job

[Christian Ebner]

Currently the target datastores' backend is instatziated for each
chunk to be inserted, which on s3 backed datastores leads to the
s3-client being re-instantiated and a new connection being
established.

Optimize this by only creating the backend once and sharing it for
all the chunk inserts to be performed.

Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
---
 src/server/pull.rs | 30 +++++++++++++++++++++---------
 1 file changed, 21 insertions(+), 9 deletions(-)

diff --git a/src/server/pull.rs b/src/server/pull.rs
index 817b57ac5..de8b140bc 100644
--- a/src/server/pull.rs
+++ b/src/server/pull.rs
@@ -38,6 +38,8 @@ use crate::tools::parallel_handler::ParallelHandler;
 pub(crate) struct PullTarget {
     store: Arc<DataStore>,
     ns: BackupNamespace,
+    // Contains the active S3Client in case of S3 backend
+    backend: DatastoreBackend,
 }
 
 /// Parameters for a pull operation.
@@ -114,10 +116,9 @@ impl PullParameters {
                 ns: remote_ns,
             })
         };
-        let target = PullTarget {
-            store: DataStore::lookup_datastore(store, Some(Operation::Write))?,
-            ns,
-        };
+        let store = DataStore::lookup_datastore(store, Some(Operation::Write))?;
+        let backend = store.backend()?;
+        let target = PullTarget { store, ns, backend };
 
         let group_filter = group_filter.unwrap_or_default();
 
@@ -141,6 +142,7 @@ async fn pull_index_chunks<I: IndexFile>(
     target: Arc<DataStore>,
     index: I,
     downloaded_chunks: Arc<Mutex<HashSet<[u8; 32]>>>,
+    backend: &DatastoreBackend,
 ) -> Result<SyncStats, Error> {
     use futures::stream::{self, StreamExt, TryStreamExt};
 
@@ -162,13 +164,14 @@ async fn pull_index_chunks<I: IndexFile>(
     );
 
     let target2 = target.clone();
+    let backend = backend.clone();
     let verify_pool = ParallelHandler::new(
         "sync chunk writer",
         4,
         move |(chunk, digest, size): (DataBlob, [u8; 32], u64)| {
             // println!("verify and write {}", hex::encode(&digest));
             chunk.verify_unencrypted(size as usize, &digest)?;
-            match target2.backend()? {
+            match &backend {
                 DatastoreBackend::Filesystem => {
                     target2.insert_chunk(&chunk, &digest)?;
                 }
@@ -283,6 +286,7 @@ async fn pull_single_archive<'a>(
     snapshot: &'a pbs_datastore::BackupDir,
     archive_info: &'a FileInfo,
     downloaded_chunks: Arc<Mutex<HashSet<[u8; 32]>>>,
+    backend: &DatastoreBackend,
 ) -> Result<SyncStats, Error> {
     let archive_name = &archive_info.filename;
     let mut path = snapshot.full_path();
@@ -317,6 +321,7 @@ async fn pull_single_archive<'a>(
                     snapshot.datastore().clone(),
                     index,
                     downloaded_chunks,
+                    backend,
                 )
                 .await?;
                 sync_stats.add(stats);
@@ -339,6 +344,7 @@ async fn pull_single_archive<'a>(
                     snapshot.datastore().clone(),
                     index,
                     downloaded_chunks,
+                    backend,
                 )
                 .await?;
                 sync_stats.add(stats);
@@ -495,15 +501,21 @@ async fn pull_snapshot<'a>(
             }
         }
 
-        let stats =
-            pull_single_archive(reader.clone(), snapshot, item, downloaded_chunks.clone()).await?;
+        let stats = pull_single_archive(
+            reader.clone(),
+            snapshot,
+            item,
+            downloaded_chunks.clone(),
+            &params.target.backend,
+        )
+        .await?;
         sync_stats.add(stats);
     }
 
     if let Err(err) = std::fs::rename(&tmp_manifest_name, &manifest_name) {
         bail!("Atomic rename file {:?} failed - {}", manifest_name, err);
     }
-    if let DatastoreBackend::S3(s3_client) = snapshot.datastore().backend()? {
+    if let DatastoreBackend::S3(s3_client) = &params.target.backend {
         let object_key = pbs_datastore::s3::object_key_from_path(
             &snapshot.relative_path(),
             MANIFEST_BLOB_NAME.as_ref(),
@@ -520,7 +532,7 @@ async fn pull_snapshot<'a>(
     if !client_log_name.exists() {
         reader.try_download_client_log(&client_log_name).await?;
         if client_log_name.exists() {
-            if let DatastoreBackend::S3(s3_client) = snapshot.datastore().backend()? {
+            if let DatastoreBackend::S3(s3_client) = &params.target.backend {
                 let object_key = pbs_datastore::s3::object_key_from_path(
                     &snapshot.relative_path(),
                     CLIENT_LOG_BLOB_NAME.as_ref(),
-- 
2.47.3



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

[pbs-devel] [PATCH proxmox-backup 02/17] api/datastore: move group notes setting to the datastore

[Christian Ebner]

In an effort to abstract away the datastore backend related logic
from the api, move the set_group_notes to a helper method on the
datastore.

The new helper method now also assures that the exclusive lock on
the backup group is acquired before updating the notes, in order to
avoid possible race conditions, e.g. with backup group destruction.

Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
---
 pbs-datastore/src/datastore.rs | 26 ++++++++++++++++++++++++++
 src/api2/admin/datastore.rs    | 17 ++++-------------
 2 files changed, 30 insertions(+), 13 deletions(-)

diff --git a/pbs-datastore/src/datastore.rs b/pbs-datastore/src/datastore.rs
index 127ba1c81..45f315aeb 100644
--- a/pbs-datastore/src/datastore.rs
+++ b/pbs-datastore/src/datastore.rs
@@ -2418,4 +2418,30 @@ impl DataStore {
             .map_err(|err| format_err!("{err:#}"))?;
         Ok((backend_type, Some(s3_client)))
     }
+
+    /// Creates or updates the notes associated with a backup group.
+    /// Acquires exclusive lock on the backup group.
+    pub fn set_group_notes(
+        self: &Arc<Self>,
+        notes: String,
+        backup_group: BackupGroup,
+    ) -> Result<(), Error> {
+        let _lock = backup_group.lock().context("failed to lock backup group")?;
+
+        if let DatastoreBackend::S3(s3_client) = self.backend()? {
+            let mut path = backup_group.backup_ns().path();
+            path.push(backup_group.group().to_string());
+            let object_key = crate::s3::object_key_from_path(&path, "notes")
+                .context("invalid owner file object key")?;
+            let data = hyper::body::Bytes::copy_from_slice(notes.as_bytes());
+            let _is_duplicate = proxmox_async::runtime::block_on(
+                s3_client.upload_replace_with_retry(object_key, data),
+            )
+            .context("failed to set notes on s3 backend")?;
+        }
+        let notes_path = self.group_notes_path(backup_group.backup_ns(), backup_group.group());
+        replace_file(notes_path, notes.as_bytes(), CreateOptions::new(), false)
+            .context("failed to replace group notes file")?;
+        Ok(())
+    }
 }
diff --git a/src/api2/admin/datastore.rs b/src/api2/admin/datastore.rs
index d192ee390..131cdae51 100644
--- a/src/api2/admin/datastore.rs
+++ b/src/api2/admin/datastore.rs
@@ -2009,19 +2009,10 @@ pub fn set_group_notes(
         &backup_group,
     )?;
 
-    if let DatastoreBackend::S3(s3_client) = datastore.backend()? {
-        let mut path = ns.path();
-        path.push(backup_group.to_string());
-        let object_key = pbs_datastore::s3::object_key_from_path(&path, "notes")
-            .context("invalid owner file object key")?;
-        let data = hyper::body::Bytes::copy_from_slice(notes.as_bytes());
-        let _is_duplicate =
-            proxmox_async::runtime::block_on(s3_client.upload_replace_with_retry(object_key, data))
-                .context("failed to set notes on s3 backend")?;
-    }
-    let notes_path = datastore.group_notes_path(&ns, &backup_group);
-    replace_file(notes_path, notes.as_bytes(), CreateOptions::new(), false)?;
-
+    let backup_group = datastore.backup_group(ns, backup_group);
+    datastore
+        .set_group_notes(notes, backup_group)
+        .map_err(|err| format_err!("failed to set group notes - {err:#?}"))?;
     Ok(())
 }
 
-- 
2.47.3



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

[pbs-devel] [PATCH proxmox-backup 03/17] api/datastore: move snapshot deletion into dedicated datastore helper

[Christian Ebner]

In an effort to move all datastore backend related logic to the
datastore itself.

Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
---
 pbs-datastore/src/datastore.rs |  9 +++++++++
 src/api2/admin/datastore.rs    | 25 ++++++++-----------------
 2 files changed, 17 insertions(+), 17 deletions(-)

diff --git a/pbs-datastore/src/datastore.rs b/pbs-datastore/src/datastore.rs
index 45f315aeb..46600a88c 100644
--- a/pbs-datastore/src/datastore.rs
+++ b/pbs-datastore/src/datastore.rs
@@ -2444,4 +2444,13 @@ impl DataStore {
             .context("failed to replace group notes file")?;
         Ok(())
     }
+
+    /// Delete a backup snapshot from the datastore.
+    /// Acquires exclusive lock on the backup snapshot.
+    pub fn delete_snapshot(self: &Arc<Self>, snapshot: &BackupDir) -> Result<(), Error> {
+        let backend = self.backend().context("failed to get backend")?;
+        // acquires exclusive lock on snapshot before removal
+        snapshot.destroy(false, &backend)?;
+        Ok(())
+    }
 }
diff --git a/src/api2/admin/datastore.rs b/src/api2/admin/datastore.rs
index 131cdae51..763440df9 100644
--- a/src/api2/admin/datastore.rs
+++ b/src/api2/admin/datastore.rs
@@ -378,9 +378,9 @@ pub async fn delete_snapshot(
         )?;
 
         let snapshot = datastore.backup_dir(ns, backup_dir)?;
-
-        snapshot.destroy(false, &datastore.backend()?)?;
-
+        datastore
+            .delete_snapshot(&snapshot)
+            .map_err(|err| format_err!("failed to delete snapshot - {err:#?}"))?;
         Ok(Value::Null)
     })
     .await?
@@ -986,21 +986,12 @@ pub fn prune(
             });
 
             if !keep {
-                match datastore.backend() {
-                    Ok(backend) => {
-                        if let Err(err) = backup_dir.destroy(false, &backend) {
-                            warn!(
-                                "failed to remove dir {:?}: {}",
-                                backup_dir.relative_path(),
-                                err,
-                            );
-                        }
-                    }
-                    Err(err) => warn!(
-                        "failed to remove dir {:?}: {err}",
+                if let Err(err) = datastore.delete_snapshot(backup_dir) {
+                    warn!(
+                        "failed to remove dir {:?}: {err:#?}",
                         backup_dir.relative_path()
-                    ),
-                };
+                    );
+                }
             }
         }
         prune_result
-- 
2.47.3



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

[pbs-devel] [PATCH proxmox-backup 04/17] api/datastore: move backup log upload by implementing datastore helper

[Christian Ebner]

In an effort to decouple the api from the datastore backend, move the
backup task log upload to use a new add blob helper method of the
datastore.

The new helper is fully sync and called on a blocking task, thereby
now also solving the previously incorrectly blocking rename_file() in
async context.

Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
---
 pbs-datastore/src/datastore.rs | 22 ++++++++++++++++++++++
 src/api2/admin/datastore.rs    | 25 +++++++------------------
 2 files changed, 29 insertions(+), 18 deletions(-)

diff --git a/pbs-datastore/src/datastore.rs b/pbs-datastore/src/datastore.rs
index 46600a88c..cc1267d78 100644
--- a/pbs-datastore/src/datastore.rs
+++ b/pbs-datastore/src/datastore.rs
@@ -2453,4 +2453,26 @@ impl DataStore {
         snapshot.destroy(false, &backend)?;
         Ok(())
     }
+
+    /// Adds the blob to the given snapshot.
+    /// Requires the caller to hold the exclusive lock.
+    pub fn add_blob(
+        self: &Arc<Self>,
+        filename: &str,
+        snapshot: BackupDir,
+        blob: DataBlob,
+    ) -> Result<(), Error> {
+        if let DatastoreBackend::S3(s3_client) = self.backend()? {
+            let object_key = crate::s3::object_key_from_path(&snapshot.relative_path(), filename)
+                .context("invalid client log object key")?;
+            let data = hyper::body::Bytes::copy_from_slice(blob.raw_data());
+            proxmox_async::runtime::block_on(s3_client.upload_replace_with_retry(object_key, data))
+                .context("failed to upload client log to s3 backend")?;
+        };
+
+        let mut path = snapshot.full_path();
+        path.push(filename);
+        replace_file(&path, blob.raw_data(), CreateOptions::new(), false)?;
+        Ok(())
+    }
 }
diff --git a/src/api2/admin/datastore.rs b/src/api2/admin/datastore.rs
index 763440df9..6881b4093 100644
--- a/src/api2/admin/datastore.rs
+++ b/src/api2/admin/datastore.rs
@@ -28,9 +28,7 @@ use proxmox_router::{
 use proxmox_rrd_api_types::{RrdMode, RrdTimeframe};
 use proxmox_schema::*;
 use proxmox_sortable_macro::sortable;
-use proxmox_sys::fs::{
-    file_read_firstline, file_read_optional_string, replace_file, CreateOptions,
-};
+use proxmox_sys::fs::{file_read_firstline, file_read_optional_string, CreateOptions};
 use proxmox_time::CalendarEvent;
 use proxmox_worker_task::WorkerTaskContext;
 
@@ -63,7 +61,7 @@ use pbs_datastore::manifest::BackupManifest;
 use pbs_datastore::prune::compute_prune_info;
 use pbs_datastore::{
     check_backup_owner, ensure_datastore_is_mounted, task_tracking, BackupDir, DataStore,
-    DatastoreBackend, LocalChunkReader, StoreProgress,
+    LocalChunkReader, StoreProgress,
 };
 use pbs_tools::json::required_string_param;
 use proxmox_rest_server::{formatter, worker_is_active, WorkerTask};
@@ -1536,20 +1534,11 @@ pub fn upload_backup_log(
         // always verify blob/CRC at server side
         let blob = DataBlob::load_from_reader(&mut &data[..])?;
 
-        if let DatastoreBackend::S3(s3_client) = datastore.backend()? {
-            let object_key = pbs_datastore::s3::object_key_from_path(
-                &backup_dir.relative_path(),
-                file_name.as_ref(),
-            )
-            .context("invalid client log object key")?;
-            let data = hyper::body::Bytes::copy_from_slice(blob.raw_data());
-            s3_client
-                .upload_replace_with_retry(object_key, data)
-                .await
-                .context("failed to upload client log to s3 backend")?;
-        };
-
-        replace_file(&path, blob.raw_data(), CreateOptions::new(), false)?;
+        tokio::task::spawn_blocking(move || {
+            datastore.add_blob(file_name.as_ref(), backup_dir, blob)
+        })
+        .await
+        .map_err(|err| format_err!("{err:#?}"))??;
 
         // fixme: use correct formatter
         Ok(formatter::JSON_FORMATTER.format_data(Value::Null, &*rpcenv))
-- 
2.47.3



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

[pbs-devel] [PATCH proxmox-backup 05/17] api/datastore: add dedicated datastore helper to set snapshot notes

[Christian Ebner]

In an effort to decouple the api from the datastore backend.

Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
---
 pbs-datastore/src/datastore.rs | 11 +++++++++++
 src/api2/admin/datastore.rs    |  7 +------
 2 files changed, 12 insertions(+), 6 deletions(-)

diff --git a/pbs-datastore/src/datastore.rs b/pbs-datastore/src/datastore.rs
index cc1267d78..ae85be76d 100644
--- a/pbs-datastore/src/datastore.rs
+++ b/pbs-datastore/src/datastore.rs
@@ -2475,4 +2475,15 @@ impl DataStore {
         replace_file(&path, blob.raw_data(), CreateOptions::new(), false)?;
         Ok(())
     }
+
+    /// Set the notes for given snapshots.
+    pub fn set_notes(self: &Arc<Self>, notes: String, snapshot: BackupDir) -> Result<(), Error> {
+        let backend = self.backend().context("failed to get backend")?;
+        snapshot
+            .update_manifest(&backend, |manifest| {
+                manifest.unprotected["notes"] = notes.into();
+            })
+            .map_err(|err| format_err!("unable to update manifest blob - {err}"))?;
+        Ok(())
+    }
 }
diff --git a/src/api2/admin/datastore.rs b/src/api2/admin/datastore.rs
index 6881b4093..000cf80c3 100644
--- a/src/api2/admin/datastore.rs
+++ b/src/api2/admin/datastore.rs
@@ -2090,12 +2090,7 @@ pub fn set_notes(
     )?;
 
     let backup_dir = datastore.backup_dir(ns, backup_dir)?;
-
-    backup_dir
-        .update_manifest(&datastore.backend()?, |manifest| {
-            manifest.unprotected["notes"] = notes.into();
-        })
-        .map_err(|err| format_err!("unable to update manifest blob - {}", err))?;
+    datastore.set_notes(notes, backup_dir)?;
 
     Ok(())
 }
-- 
2.47.3



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

[pbs-devel] [PATCH proxmox-backup 06/17] datastore: refactor chunk insert based on backend

[Christian Ebner]

With the goal to move all the backend related implementation details
into the datastore's method and deduplicate code as much as possible.

The backend has to be passed in from the caller on chunk insert, as
this stores the s3-client which lives for the whole backup session
lifetime. Same is true for the pull sync job and the tape restore,
although for latter the s3 functionality is not exposed yet as still
incomplete.

To distinguish between inserts in datastors with cache, for cases
where chunks should bypass the cache, add a dedicated method for that.

This will facilitate the implementation of the per-chunk file locking
for cache and backend consistency on s3 backed stores.

Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
---
 pbs-datastore/src/datastore.rs  | 85 ++++++++++++++++++++++++++++++++-
 src/api2/backup/upload_chunk.rs | 64 ++++---------------------
 src/api2/tape/restore.rs        |  6 ++-
 src/server/pull.rs              | 19 +-------
 4 files changed, 98 insertions(+), 76 deletions(-)

diff --git a/pbs-datastore/src/datastore.rs b/pbs-datastore/src/datastore.rs
index ae85be76d..9008d8fc6 100644
--- a/pbs-datastore/src/datastore.rs
+++ b/pbs-datastore/src/datastore.rs
@@ -8,6 +8,7 @@ use std::time::{Duration, SystemTime};
 
 use anyhow::{bail, format_err, Context, Error};
 use http_body_util::BodyExt;
+use hyper::body::Bytes;
 use nix::unistd::{unlinkat, UnlinkatFlags};
 use pbs_tools::lru_cache::LruCache;
 use tokio::io::AsyncWriteExt;
@@ -1855,8 +1856,88 @@ impl DataStore {
             .cond_touch_chunk(digest, assert_exists)
     }
 
-    pub fn insert_chunk(&self, chunk: &DataBlob, digest: &[u8; 32]) -> Result<(bool, u64), Error> {
-        self.inner.chunk_store.insert_chunk(chunk, digest)
+    /// Inserts the chunk with given digest in the chunk store based on the given backend.
+    ///
+    /// The backend is passed along in order to reuse an active connection to the backend, created
+    /// on environment instantiation, e.g. an s3 client which lives for the whole backup session.
+    /// This calls into async code, so callers must assure to never hold a Mutex lock.
+    pub fn insert_chunk(
+        &self,
+        chunk: &DataBlob,
+        digest: &[u8; 32],
+        backend: &DatastoreBackend,
+    ) -> Result<(bool, u64), Error> {
+        match backend {
+            DatastoreBackend::Filesystem => self.inner.chunk_store.insert_chunk(chunk, digest),
+            DatastoreBackend::S3(s3_client) => self.insert_chunk_cached(chunk, digest, &s3_client),
+        }
+    }
+
+    /// Inserts the chunk with given digest in the chunk store based on the given backend, but
+    /// always bypasses the local datastore cache.
+    ///
+    /// The backend is passed along in order to reuse an active connection to the backend, created
+    /// on environment instantiation, e.g. an s3 client which lives for the whole backup session.
+    /// This calls into async code, so callers must assure to never hold a Mutex lock.
+    ///
+    /// FIXME: refactor into insert_chunk() once the backend instance is cacheable on the datastore
+    /// itself.
+    pub fn insert_chunk_no_cache(
+        &self,
+        chunk: &DataBlob,
+        digest: &[u8; 32],
+        backend: &DatastoreBackend,
+    ) -> Result<(bool, u64), Error> {
+        match backend {
+            DatastoreBackend::Filesystem => self.inner.chunk_store.insert_chunk(chunk, digest),
+            DatastoreBackend::S3(s3_client) => {
+                let chunk_data: Bytes = chunk.raw_data().to_vec().into();
+                let chunk_size = chunk_data.len() as u64;
+                let object_key = crate::s3::object_key_from_digest(digest)?;
+                let is_duplicate = proxmox_async::runtime::block_on(
+                    s3_client.upload_no_replace_with_retry(object_key, chunk_data),
+                )
+                .map_err(|err| format_err!("failed to upload chunk to s3 backend - {err:#}"))?;
+                Ok((is_duplicate, chunk_size))
+            }
+        }
+    }
+
+    fn insert_chunk_cached(
+        &self,
+        chunk: &DataBlob,
+        digest: &[u8; 32],
+        s3_client: &Arc<S3Client>,
+    ) -> Result<(bool, u64), Error> {
+        let chunk_data = chunk.raw_data();
+        let chunk_size = chunk_data.len() as u64;
+
+        // Avoid re-upload to S3 if the chunk is either present in the in-memory LRU cache
+        // or the chunk marker file exists on filesystem. The latter means the chunk has
+        // been uploaded in the past, but was evicted from the LRU cache since but was not
+        // cleaned up by garbage collection, so contained in the S3 object store.
+        if self.cache_contains(&digest) {
+            tracing::info!("Skip upload of cached chunk {}", hex::encode(digest));
+            return Ok((true, chunk_size));
+        }
+        if let Ok(true) = self.cond_touch_chunk(digest, false) {
+            tracing::info!(
+                "Skip upload of already encountered chunk {}",
+                hex::encode(digest),
+            );
+            return Ok((true, chunk_size));
+        }
+
+        tracing::info!("Upload new chunk {}", hex::encode(digest));
+        let object_key = crate::s3::object_key_from_digest(digest)?;
+        let chunk_data: Bytes = chunk_data.to_vec().into();
+        let is_duplicate = proxmox_async::runtime::block_on(
+            s3_client.upload_no_replace_with_retry(object_key, chunk_data),
+        )
+        .map_err(|err| format_err!("failed to upload chunk to s3 backend - {err:#}"))?;
+        tracing::info!("Caching of chunk {}", hex::encode(digest));
+        self.cache_insert(&digest, &chunk)?;
+        Ok((is_duplicate, chunk_size))
     }
 
     pub fn stat_chunk(&self, digest: &[u8; 32]) -> Result<std::fs::Metadata, Error> {
diff --git a/src/api2/backup/upload_chunk.rs b/src/api2/backup/upload_chunk.rs
index 297f23b12..f4a50002d 100644
--- a/src/api2/backup/upload_chunk.rs
+++ b/src/api2/backup/upload_chunk.rs
@@ -5,7 +5,7 @@ use anyhow::{bail, format_err, Error};
 use futures::*;
 use hex::FromHex;
 use http_body_util::{BodyDataStream, BodyExt};
-use hyper::body::{Bytes, Incoming};
+use hyper::body::Incoming;
 use hyper::http::request::Parts;
 use serde_json::{json, Value};
 
@@ -15,7 +15,7 @@ use proxmox_sortable_macro::sortable;
 
 use pbs_api_types::{BACKUP_ARCHIVE_NAME_SCHEMA, CHUNK_DIGEST_SCHEMA};
 use pbs_datastore::file_formats::{DataBlobHeader, EncryptedDataBlobHeader};
-use pbs_datastore::{DataBlob, DatastoreBackend};
+use pbs_datastore::DataBlob;
 use pbs_tools::json::{required_integer_param, required_string_param};
 
 use super::environment::*;
@@ -232,59 +232,15 @@ async fn upload_to_backend(
     let (digest, size, chunk) =
         UploadChunk::new(BodyDataStream::new(req_body), digest, size, encoded_size).await?;
 
-    match &env.backend {
-        DatastoreBackend::Filesystem => {
-            let (is_duplicate, compressed_size) = proxmox_async::runtime::block_in_place(|| {
-                env.datastore.insert_chunk(&chunk, &digest)
-            })?;
-            Ok((digest, size, compressed_size as u32, is_duplicate))
-        }
-        DatastoreBackend::S3(s3_client) => {
-            let chunk_data: Bytes = chunk.raw_data().to_vec().into();
-
-            if env.no_cache {
-                let object_key = pbs_datastore::s3::object_key_from_digest(&digest)?;
-                let is_duplicate = s3_client
-                    .upload_no_replace_with_retry(object_key, chunk_data)
-                    .await
-                    .map_err(|err| format_err!("failed to upload chunk to s3 backend - {err:#}"))?;
-                return Ok((digest, size, encoded_size, is_duplicate));
-            }
-
-            // Avoid re-upload to S3 if the chunk is either present in the LRU cache or the chunk
-            // file exists on filesystem. The latter means that the chunk has been present in the
-            // past an was not cleaned up by garbage collection, so contained in the S3 object store.
-            if env.datastore.cache_contains(&digest) {
-                tracing::info!("Skip upload of cached chunk {}", hex::encode(digest));
-                return Ok((digest, size, encoded_size, true));
-            }
-            if let Ok(true) = env.datastore.cond_touch_chunk(&digest, false) {
-                tracing::info!(
-                    "Skip upload of already encountered chunk {}",
-                    hex::encode(digest)
-                );
-                return Ok((digest, size, encoded_size, true));
-            }
-
-            tracing::info!("Upload of new chunk {}", hex::encode(digest));
-            let object_key = pbs_datastore::s3::object_key_from_digest(&digest)?;
-            let is_duplicate = s3_client
-                .upload_no_replace_with_retry(object_key, chunk_data)
-                .await
-                .map_err(|err| format_err!("failed to upload chunk to s3 backend - {err:#}"))?;
-
-            // Only insert the chunk into the cache after it has been successufuly uploaded.
-            // Although less performant than doing this in parallel, it is required for consisency
-            // since chunks are considered as present on the backend if the file exists in the local
-            // cache store.
-            let datastore = env.datastore.clone();
-            tracing::info!("Caching of chunk {}", hex::encode(digest));
-            let _ = tokio::task::spawn_blocking(move || datastore.cache_insert(&digest, &chunk))
-                .await?;
-
-            Ok((digest, size, encoded_size, is_duplicate))
-        }
+    if env.no_cache {
+        let (is_duplicate, chunk_size) =
+            env.datastore
+                .insert_chunk_no_cache(&chunk, &digest, &env.backend)?;
+        return Ok((digest, size, chunk_size as u32, is_duplicate));
     }
+
+    let (is_duplicate, chunk_size) = env.datastore.insert_chunk(&chunk, &digest, &env.backend)?;
+    Ok((digest, size, chunk_size as u32, is_duplicate))
 }
 
 pub const API_METHOD_UPLOAD_SPEEDTEST: ApiMethod = ApiMethod::new(
diff --git a/src/api2/tape/restore.rs b/src/api2/tape/restore.rs
index 8b6979017..4f2ee3db6 100644
--- a/src/api2/tape/restore.rs
+++ b/src/api2/tape/restore.rs
@@ -1151,6 +1151,7 @@ fn restore_partial_chunk_archive<'a>(
     let bytes = Arc::new(std::sync::atomic::AtomicU64::new(0));
     let bytes2 = bytes.clone();
 
+    let backend = datastore.backend()?;
     let writer_pool = ParallelHandler::new(
         "tape restore chunk writer",
         4,
@@ -1162,7 +1163,7 @@ fn restore_partial_chunk_archive<'a>(
                     chunk.decode(None, Some(&digest))?; // verify digest
                 }
 
-                datastore.insert_chunk(&chunk, &digest)?;
+                datastore.insert_chunk(&chunk, &digest, &backend)?;
             }
             Ok(())
         },
@@ -1544,6 +1545,7 @@ fn restore_chunk_archive<'a>(
     let bytes = Arc::new(std::sync::atomic::AtomicU64::new(0));
     let bytes2 = bytes.clone();
 
+    let backend = datastore.backend()?;
     let writer_pool = ParallelHandler::new(
         "tape restore chunk writer",
         4,
@@ -1560,7 +1562,7 @@ fn restore_chunk_archive<'a>(
                     chunk.decode(None, Some(&digest))?; // verify digest
                 }
 
-                datastore.insert_chunk(&chunk, &digest)?;
+                datastore.insert_chunk(&chunk, &digest, &backend)?;
             } else if verbose {
                 info!("Found existing chunk: {}", hex::encode(digest));
             }
diff --git a/src/server/pull.rs b/src/server/pull.rs
index de8b140bc..db7084f3e 100644
--- a/src/server/pull.rs
+++ b/src/server/pull.rs
@@ -171,24 +171,7 @@ async fn pull_index_chunks<I: IndexFile>(
         move |(chunk, digest, size): (DataBlob, [u8; 32], u64)| {
             // println!("verify and write {}", hex::encode(&digest));
             chunk.verify_unencrypted(size as usize, &digest)?;
-            match &backend {
-                DatastoreBackend::Filesystem => {
-                    target2.insert_chunk(&chunk, &digest)?;
-                }
-                DatastoreBackend::S3(s3_client) => {
-                    if target2.cache_contains(&digest) {
-                        return Ok(());
-                    }
-                    target2.cache_insert(&digest, &chunk)?;
-                    let data = chunk.raw_data().to_vec();
-                    let upload_data = hyper::body::Bytes::from(data);
-                    let object_key = pbs_datastore::s3::object_key_from_digest(&digest)?;
-                    let _is_duplicate = proxmox_async::runtime::block_on(
-                        s3_client.upload_no_replace_with_retry(object_key, upload_data),
-                    )
-                    .context("failed to upload chunk to s3 backend")?;
-                }
-            }
+            target2.insert_chunk(&chunk, &digest, &backend)?;
             Ok(())
         },
     );
-- 
2.47.3



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

[pbs-devel] [PATCH proxmox-backup 07/17] verify: rename corrupted to corrupt in log output and function names

[Christian Ebner]

Throughout the codebase the use of corrupt chunk is more frequent, so
stick to that for the log outputs and function names as well.

Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
---
 src/backup/verify.rs | 14 +++++++-------
 1 file changed, 7 insertions(+), 7 deletions(-)

diff --git a/src/backup/verify.rs b/src/backup/verify.rs
index e33fdf50f..df77d177b 100644
--- a/src/backup/verify.rs
+++ b/src/backup/verify.rs
@@ -76,7 +76,7 @@ impl VerifyWorker {
         }
     }
 
-    fn rename_corrupted_chunk(datastore: Arc<DataStore>, digest: &[u8; 32]) {
+    fn rename_corrupt_chunk(datastore: Arc<DataStore>, digest: &[u8; 32]) {
         let (path, digest_str) = datastore.chunk_path(digest);
 
         let mut counter = 0;
@@ -106,14 +106,14 @@ impl VerifyWorker {
                 match pbs_datastore::s3::object_key_from_digest_with_suffix(digest, &suffix) {
                     Ok(target_key) => target_key,
                     Err(err) => {
-                        info!("could not generate target key for corrupted chunk {path:?} - {err}");
+                        info!("could not generate target key for corrupt chunk {path:?} - {err}");
                         return;
                     }
                 };
             let object_key = match pbs_datastore::s3::object_key_from_digest(digest) {
                 Ok(object_key) => object_key,
                 Err(err) => {
-                    info!("could not generate object key for corrupted chunk {path:?} - {err}");
+                    info!("could not generate object key for corrupt chunk {path:?} - {err}");
                     return;
                 }
             };
@@ -136,12 +136,12 @@ impl VerifyWorker {
 
         match std::fs::rename(&path, &new_path) {
             Ok(_) => {
-                info!("corrupted chunk renamed to {:?}", &new_path);
+                info!("corrupt chunk renamed to {:?}", &new_path);
             }
             Err(err) => {
                 match err.kind() {
                     std::io::ErrorKind::NotFound => { /* ignored */ }
-                    _ => info!("could not rename corrupted chunk {:?} - {err}", &path),
+                    _ => info!("could not rename corrupt chunk {:?} - {err}", &path),
                 }
             }
         };
@@ -189,7 +189,7 @@ impl VerifyWorker {
                     corrupt_chunks2.lock().unwrap().insert(digest);
                     info!("{err}");
                     errors2.fetch_add(1, Ordering::SeqCst);
-                    Self::rename_corrupted_chunk(datastore2.clone(), &digest);
+                    Self::rename_corrupt_chunk(datastore2.clone(), &digest);
                 } else {
                     verified_chunks2.lock().unwrap().insert(digest);
                 }
@@ -336,7 +336,7 @@ impl VerifyWorker {
         corrupt_chunks.insert(digest);
         error!(message);
         errors.fetch_add(1, Ordering::SeqCst);
-        Self::rename_corrupted_chunk(self.datastore.clone(), &digest);
+        Self::rename_corrupt_chunk(self.datastore.clone(), &digest);
     }
 
     fn verify_fixed_index(&self, backup_dir: &BackupDir, info: &FileInfo) -> Result<(), Error> {
-- 
2.47.3



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

[pbs-devel] [PATCH proxmox-backup 08/17] verify/datastore: make rename corrupt chunk a datastore helper method

[Christian Ebner]

By making the rename a helper of the datastore, within this method it
will become possible to access the inner chunk store for locking.

That will be required to correctly lock the store to avoid concurrent
chunk inserts and garbage collection operations during the rename, to
guarantee consistency on datastores with s3 backend.

This is a preparatory patch, no functional changes intended.

Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
---
 pbs-datastore/src/datastore.rs | 70 +++++++++++++++++++++++++++++++
 src/backup/verify.rs           | 75 +---------------------------------
 2 files changed, 72 insertions(+), 73 deletions(-)

diff --git a/pbs-datastore/src/datastore.rs b/pbs-datastore/src/datastore.rs
index 9008d8fc6..ebfbf5229 100644
--- a/pbs-datastore/src/datastore.rs
+++ b/pbs-datastore/src/datastore.rs
@@ -2567,4 +2567,74 @@ impl DataStore {
             .map_err(|err| format_err!("unable to update manifest blob - {err}"))?;
         Ok(())
     }
+
+    pub fn rename_corrupt_chunk(&self, digest: &[u8; 32]) {
+        let (path, digest_str) = self.chunk_path(digest);
+
+        let mut counter = 0;
+        let mut new_path = path.clone();
+        loop {
+            new_path.set_file_name(format!("{}.{}.bad", digest_str, counter));
+            if new_path.exists() && counter < 9 {
+                counter += 1;
+            } else {
+                break;
+            }
+        }
+
+        let backend = match self.backend() {
+            Ok(backend) => backend,
+            Err(err) => {
+                info!(
+                    "failed to get backend while trying to rename bad chunk: {digest_str} - {err}"
+                );
+                return;
+            }
+        };
+
+        if let DatastoreBackend::S3(s3_client) = backend {
+            let suffix = format!(".{}.bad", counter);
+            let target_key = match crate::s3::object_key_from_digest_with_suffix(digest, &suffix) {
+                Ok(target_key) => target_key,
+                Err(err) => {
+                    info!("could not generate target key for corrupt chunk {path:?} - {err}");
+                    return;
+                }
+            };
+            let object_key = match crate::s3::object_key_from_digest(digest) {
+                Ok(object_key) => object_key,
+                Err(err) => {
+                    info!("could not generate object key for corrupt chunk {path:?} - {err}");
+                    return;
+                }
+            };
+            if proxmox_async::runtime::block_on(
+                s3_client.copy_object(object_key.clone(), target_key),
+            )
+            .is_ok()
+            {
+                if proxmox_async::runtime::block_on(s3_client.delete_object(object_key)).is_err() {
+                    info!("failed to delete corrupt chunk on s3 backend: {digest_str}");
+                }
+            } else {
+                info!("failed to copy corrupt chunk on s3 backend: {digest_str}");
+                // Early return to leave the potentially locally cached chunk in the same state as
+                // on the object store. Verification might have failed because of connection issue
+                // after all.
+                return;
+            }
+        }
+
+        match std::fs::rename(&path, &new_path) {
+            Ok(_) => {
+                info!("corrupt chunk renamed to {:?}", &new_path);
+            }
+            Err(err) => {
+                match err.kind() {
+                    std::io::ErrorKind::NotFound => { /* ignored */ }
+                    _ => info!("could not rename corrupt chunk {:?} - {err}", &path),
+                }
+            }
+        };
+    }
 }
diff --git a/src/backup/verify.rs b/src/backup/verify.rs
index df77d177b..7fac46e18 100644
--- a/src/backup/verify.rs
+++ b/src/backup/verify.rs
@@ -76,77 +76,6 @@ impl VerifyWorker {
         }
     }
 
-    fn rename_corrupt_chunk(datastore: Arc<DataStore>, digest: &[u8; 32]) {
-        let (path, digest_str) = datastore.chunk_path(digest);
-
-        let mut counter = 0;
-        let mut new_path = path.clone();
-        loop {
-            new_path.set_file_name(format!("{digest_str}.{counter}.bad"));
-            if new_path.exists() && counter < 9 {
-                counter += 1;
-            } else {
-                break;
-            }
-        }
-
-        let backend = match datastore.backend() {
-            Ok(backend) => backend,
-            Err(err) => {
-                info!(
-                    "failed to get backend while trying to rename bad chunk: {digest_str} - {err}"
-                );
-                return;
-            }
-        };
-
-        if let DatastoreBackend::S3(s3_client) = backend {
-            let suffix = format!(".{counter}.bad");
-            let target_key =
-                match pbs_datastore::s3::object_key_from_digest_with_suffix(digest, &suffix) {
-                    Ok(target_key) => target_key,
-                    Err(err) => {
-                        info!("could not generate target key for corrupt chunk {path:?} - {err}");
-                        return;
-                    }
-                };
-            let object_key = match pbs_datastore::s3::object_key_from_digest(digest) {
-                Ok(object_key) => object_key,
-                Err(err) => {
-                    info!("could not generate object key for corrupt chunk {path:?} - {err}");
-                    return;
-                }
-            };
-            if proxmox_async::runtime::block_on(
-                s3_client.copy_object(object_key.clone(), target_key),
-            )
-            .is_ok()
-            {
-                if proxmox_async::runtime::block_on(s3_client.delete_object(object_key)).is_err() {
-                    info!("failed to delete corrupt chunk on s3 backend: {digest_str}");
-                }
-            } else {
-                info!("failed to copy corrupt chunk on s3 backend: {digest_str}");
-                // Early return to leave the potentially locally cached chunk in the same state as
-                // on the object store. Verification might have failed because of connection issue
-                // after all.
-                return;
-            }
-        }
-
-        match std::fs::rename(&path, &new_path) {
-            Ok(_) => {
-                info!("corrupt chunk renamed to {:?}", &new_path);
-            }
-            Err(err) => {
-                match err.kind() {
-                    std::io::ErrorKind::NotFound => { /* ignored */ }
-                    _ => info!("could not rename corrupt chunk {:?} - {err}", &path),
-                }
-            }
-        };
-    }
-
     fn verify_index_chunks(
         &self,
         index: Box<dyn IndexFile + Send>,
@@ -189,7 +118,7 @@ impl VerifyWorker {
                     corrupt_chunks2.lock().unwrap().insert(digest);
                     info!("{err}");
                     errors2.fetch_add(1, Ordering::SeqCst);
-                    Self::rename_corrupt_chunk(datastore2.clone(), &digest);
+                    datastore2.rename_corrupt_chunk(&digest);
                 } else {
                     verified_chunks2.lock().unwrap().insert(digest);
                 }
@@ -336,7 +265,7 @@ impl VerifyWorker {
         corrupt_chunks.insert(digest);
         error!(message);
         errors.fetch_add(1, Ordering::SeqCst);
-        Self::rename_corrupt_chunk(self.datastore.clone(), &digest);
+        self.datastore.rename_corrupt_chunk(&digest);
     }
 
     fn verify_fixed_index(&self, backup_dir: &BackupDir, info: &FileInfo) -> Result<(), Error> {
-- 
2.47.3



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

[pbs-devel] [PATCH proxmox-backup 09/17] datastore: refactor rename_corrupt_chunk error handling

[Christian Ebner]

As part of the verification process, the helper was not intended to
return errors on failure but rather just log information and errors.

Refactoring the code so that the helper method optionally returns the
new chunk path after it being renamed, None if the source path could
not be found and error otherwise.

However, keep the logging as info at the callsite for both error and
success message logging to not interfere with the task log.

Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
---
 pbs-datastore/src/datastore.rs | 85 ++++++++++++++--------------------
 src/backup/verify.rs           | 12 ++++-
 2 files changed, 44 insertions(+), 53 deletions(-)

diff --git a/pbs-datastore/src/datastore.rs b/pbs-datastore/src/datastore.rs
index ebfbf5229..397c37e56 100644
--- a/pbs-datastore/src/datastore.rs
+++ b/pbs-datastore/src/datastore.rs
@@ -2568,13 +2568,15 @@ impl DataStore {
         Ok(())
     }
 
-    pub fn rename_corrupt_chunk(&self, digest: &[u8; 32]) {
+    /// Renames a corrupt chunk, returning the new path if the chunk was renamed successfully.
+    /// Returns with `Ok(None)` if the chunk source was not found.
+    pub fn rename_corrupt_chunk(&self, digest: &[u8; 32]) -> Result<Option<PathBuf>, Error> {
         let (path, digest_str) = self.chunk_path(digest);
 
         let mut counter = 0;
         let mut new_path = path.clone();
         loop {
-            new_path.set_file_name(format!("{}.{}.bad", digest_str, counter));
+            new_path.set_file_name(format!("{digest_str}.{counter}.bad"));
             if new_path.exists() && counter < 9 {
                 counter += 1;
             } else {
@@ -2582,59 +2584,40 @@ impl DataStore {
             }
         }
 
-        let backend = match self.backend() {
-            Ok(backend) => backend,
-            Err(err) => {
-                info!(
-                    "failed to get backend while trying to rename bad chunk: {digest_str} - {err}"
-                );
-                return;
-            }
-        };
+        let backend = self.backend().map_err(|err| {
+            format_err!(
+                "failed to get backend while trying to rename bad chunk: {digest_str} - {err}"
+            )
+        })?;
 
         if let DatastoreBackend::S3(s3_client) = backend {
-            let suffix = format!(".{}.bad", counter);
-            let target_key = match crate::s3::object_key_from_digest_with_suffix(digest, &suffix) {
-                Ok(target_key) => target_key,
-                Err(err) => {
-                    info!("could not generate target key for corrupt chunk {path:?} - {err}");
-                    return;
-                }
-            };
-            let object_key = match crate::s3::object_key_from_digest(digest) {
-                Ok(object_key) => object_key,
-                Err(err) => {
-                    info!("could not generate object key for corrupt chunk {path:?} - {err}");
-                    return;
-                }
-            };
-            if proxmox_async::runtime::block_on(
-                s3_client.copy_object(object_key.clone(), target_key),
-            )
-            .is_ok()
-            {
-                if proxmox_async::runtime::block_on(s3_client.delete_object(object_key)).is_err() {
-                    info!("failed to delete corrupt chunk on s3 backend: {digest_str}");
-                }
-            } else {
-                info!("failed to copy corrupt chunk on s3 backend: {digest_str}");
-                // Early return to leave the potentially locally cached chunk in the same state as
-                // on the object store. Verification might have failed because of connection issue
-                // after all.
-                return;
-            }
+            let suffix = format!(".{counter}.bad");
+            let target_key = crate::s3::object_key_from_digest_with_suffix(digest, &suffix)
+                .map_err(|err| {
+                    format_err!("could not generate target key for corrupt chunk {path:?} - {err}")
+                })?;
+            let object_key = crate::s3::object_key_from_digest(digest).map_err(|err| {
+                format_err!("could not generate object key for corrupt chunk {path:?} - {err}")
+            })?;
+
+            proxmox_async::runtime::block_on(s3_client.copy_object(object_key.clone(), target_key))
+                .map_err(|err| {
+                    format_err!("failed to copy corrupt chunk on s3 backend: {digest_str} - {err}")
+                })?;
+
+            proxmox_async::runtime::block_on(s3_client.delete_object(object_key)).map_err(
+                |err| {
+                    format_err!(
+                        "failed to delete corrupt chunk on s3 backend: {digest_str} - {err}"
+                    )
+                },
+            )?;
         }
 
         match std::fs::rename(&path, &new_path) {
-            Ok(_) => {
-                info!("corrupt chunk renamed to {:?}", &new_path);
-            }
-            Err(err) => {
-                match err.kind() {
-                    std::io::ErrorKind::NotFound => { /* ignored */ }
-                    _ => info!("could not rename corrupt chunk {:?} - {err}", &path),
-                }
-            }
-        };
+            Ok(_) => Ok(Some(new_path)),
+            Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(None),
+            Err(err) => bail!("could not rename corrupt chunk {path:?} - {err}"),
+        }
     }
 }
diff --git a/src/backup/verify.rs b/src/backup/verify.rs
index 7fac46e18..31c03891a 100644
--- a/src/backup/verify.rs
+++ b/src/backup/verify.rs
@@ -118,7 +118,11 @@ impl VerifyWorker {
                     corrupt_chunks2.lock().unwrap().insert(digest);
                     info!("{err}");
                     errors2.fetch_add(1, Ordering::SeqCst);
-                    datastore2.rename_corrupt_chunk(&digest);
+                    match datastore2.rename_corrupt_chunk(&digest) {
+                        Ok(Some(new_path)) => info!("corrupt chunk renamed to {new_path:?}"),
+                        Err(err) => info!("{err}"),
+                        _ => (),
+                    }
                 } else {
                     verified_chunks2.lock().unwrap().insert(digest);
                 }
@@ -265,7 +269,11 @@ impl VerifyWorker {
         corrupt_chunks.insert(digest);
         error!(message);
         errors.fetch_add(1, Ordering::SeqCst);
-        self.datastore.rename_corrupt_chunk(&digest);
+        match self.datastore.rename_corrupt_chunk(&digest) {
+            Ok(Some(new_path)) => info!("corrupt chunk renamed to {new_path:?}"),
+            Err(err) => info!("{err}"),
+            _ => (),
+        }
     }
 
     fn verify_fixed_index(&self, backup_dir: &BackupDir, info: &FileInfo) -> Result<(), Error> {
-- 
2.47.3



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

[pbs-devel] [PATCH proxmox-backup 10/17] datastore: implement per-chunk file locking helper for s3 backend

[Christian Ebner]

Adds a datastore helper method to create per-chunk file locks. These
will be used to guard chunk operations on s3 backends to guarantee
exclusive access when performing cache and backend operations.

Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
---
 pbs-datastore/src/backup_info.rs |  2 +-
 pbs-datastore/src/chunk_store.rs | 26 ++++++++++++++++++++++++++
 pbs-datastore/src/datastore.rs   | 12 ++++++++++++
 3 files changed, 39 insertions(+), 1 deletion(-)

diff --git a/pbs-datastore/src/backup_info.rs b/pbs-datastore/src/backup_info.rs
index 4b10b6435..70c0fbe8a 100644
--- a/pbs-datastore/src/backup_info.rs
+++ b/pbs-datastore/src/backup_info.rs
@@ -936,7 +936,7 @@ fn lock_file_path_helper(ns: &BackupNamespace, path: PathBuf) -> PathBuf {
 /// deletion.
 ///
 /// It also creates the base directory for lock files.
-fn lock_helper<F>(
+pub(crate) fn lock_helper<F>(
     store_name: &str,
     path: &std::path::Path,
     lock_fn: F,
diff --git a/pbs-datastore/src/chunk_store.rs b/pbs-datastore/src/chunk_store.rs
index ba7618e40..49687b2fa 100644
--- a/pbs-datastore/src/chunk_store.rs
+++ b/pbs-datastore/src/chunk_store.rs
@@ -8,6 +8,7 @@ use anyhow::{bail, format_err, Context, Error};
 use tracing::{info, warn};
 
 use pbs_api_types::{DatastoreFSyncLevel, GarbageCollectionStatus};
+use pbs_config::BackupLockGuard;
 use proxmox_io::ReadExt;
 use proxmox_s3_client::S3Client;
 use proxmox_sys::fs::{create_dir, create_path, file_type_from_file_stat, CreateOptions};
@@ -16,6 +17,7 @@ use proxmox_sys::process_locker::{
 };
 use proxmox_worker_task::WorkerTaskContext;
 
+use crate::backup_info::DATASTORE_LOCKS_DIR;
 use crate::data_blob::DataChunkBuilder;
 use crate::file_formats::{
     COMPRESSED_BLOB_MAGIC_1_0, ENCRYPTED_BLOB_MAGIC_1_0, UNCOMPRESSED_BLOB_MAGIC_1_0,
@@ -759,6 +761,30 @@ impl ChunkStore {
         ChunkStore::check_permissions(lockfile_path, 0o644)?;
         Ok(())
     }
+
+    /// Generates the path to the chunks lock file
+    pub(crate) fn chunk_lock_path(&self, digest: &[u8]) -> PathBuf {
+        let mut lock_path = Path::new(DATASTORE_LOCKS_DIR).join(self.name.clone());
+        let digest_str = hex::encode(digest);
+        lock_path.push(".chunks");
+        let prefix = digest_to_prefix(digest);
+        lock_path.push(&prefix);
+        lock_path.push(&digest_str);
+        lock_path
+    }
+
+    /// Get an exclusive lock on the chunks lock file
+    pub(crate) fn lock_chunk(
+        &self,
+        digest: &[u8],
+        timeout: Duration,
+    ) -> Result<BackupLockGuard, Error> {
+        let lock_path = self.chunk_lock_path(digest);
+        let guard = crate::backup_info::lock_helper(self.name(), &lock_path, |path| {
+            pbs_config::open_backup_lockfile(path, Some(timeout), true)
+        })?;
+        Ok(guard)
+    }
 }
 
 #[test]
diff --git a/pbs-datastore/src/datastore.rs b/pbs-datastore/src/datastore.rs
index 397c37e56..32f3562b3 100644
--- a/pbs-datastore/src/datastore.rs
+++ b/pbs-datastore/src/datastore.rs
@@ -2568,6 +2568,18 @@ impl DataStore {
         Ok(())
     }
 
+    /// Locks the per chunk lock file if the backend requires it
+    fn lock_chunk_for_backend(&self, digest: &[u8; 32]) -> Result<Option<BackupLockGuard>, Error> {
+        // s3 put request times out after upload_size / 1 Kib/s, so about 2.3 hours for 8 MiB
+        let timeout = Duration::from_secs(3 * 60 * 60);
+        match self.inner.backend_config.ty.unwrap_or_default() {
+            DatastoreBackendType::Filesystem => Ok(None),
+            DatastoreBackendType::S3 => {
+                self.inner.chunk_store.lock_chunk(digest, timeout).map(Some)
+            }
+        }
+    }
+
     /// Renames a corrupt chunk, returning the new path if the chunk was renamed successfully.
     /// Returns with `Ok(None)` if the chunk source was not found.
     pub fn rename_corrupt_chunk(&self, digest: &[u8; 32]) -> Result<Option<PathBuf>, Error> {
-- 
2.47.3



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

[pbs-devel] [PATCH proxmox-backup 11/17] datastore: acquire chunk store mutex lock when renaming corrupt chunk

[Christian Ebner]

When renaming a corrupt chunk in the chunk store, currently the chunk
store mutex lock is not held, leading to possible races with other
operations which hold the lock and therefore assume exclusive access
such as garbage collection or backup chunk inserts. This affects
both, filesystem and S3 backends.

To fix the possible race, get the lock and rearrange the code such
that it is never held when entering async code for the s3 backend.
This does not yet solve the race and cache consistency for s3
backends, which is addressed by introducing per-chunk file locking
in subsequent patches.

Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
---
 pbs-datastore/src/datastore.rs | 16 +++++++++++-----
 1 file changed, 11 insertions(+), 5 deletions(-)

diff --git a/pbs-datastore/src/datastore.rs b/pbs-datastore/src/datastore.rs
index 32f3562b3..e7ec87a7f 100644
--- a/pbs-datastore/src/datastore.rs
+++ b/pbs-datastore/src/datastore.rs
@@ -2585,6 +2585,8 @@ impl DataStore {
     pub fn rename_corrupt_chunk(&self, digest: &[u8; 32]) -> Result<Option<PathBuf>, Error> {
         let (path, digest_str) = self.chunk_path(digest);
 
+        let _lock = self.inner.chunk_store.mutex().lock().unwrap();
+
         let mut counter = 0;
         let mut new_path = path.clone();
         loop {
@@ -2596,6 +2598,14 @@ impl DataStore {
             }
         }
 
+        let result = match std::fs::rename(&path, &new_path) {
+            Ok(_) => Ok(Some(new_path)),
+            Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(None),
+            Err(err) => bail!("could not rename corrupt chunk {path:?} - {err}"),
+        };
+
+        drop(_lock);
+
         let backend = self.backend().map_err(|err| {
             format_err!(
                 "failed to get backend while trying to rename bad chunk: {digest_str} - {err}"
@@ -2626,10 +2636,6 @@ impl DataStore {
             )?;
         }
 
-        match std::fs::rename(&path, &new_path) {
-            Ok(_) => Ok(Some(new_path)),
-            Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(None),
-            Err(err) => bail!("could not rename corrupt chunk {path:?} - {err}"),
-        }
+        result
     }
 }
-- 
2.47.3



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

[pbs-devel] [PATCH proxmox-backup 12/17] datastore: get per-chunk file lock for chunk rename on s3 backend

[Christian Ebner]

To guarantee exclusive access during s3 object store and cache
operations, acquire the per-chunk file before renaming a chunk when
the datastore is backed by s3.

This does not yet cover locking for the GC and chunk insert, part
of subsequent changes.

Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
---
 pbs-datastore/src/datastore.rs | 1 +
 1 file changed, 1 insertion(+)

diff --git a/pbs-datastore/src/datastore.rs b/pbs-datastore/src/datastore.rs
index e7ec87a7f..e65f3a60c 100644
--- a/pbs-datastore/src/datastore.rs
+++ b/pbs-datastore/src/datastore.rs
@@ -2586,6 +2586,7 @@ impl DataStore {
         let (path, digest_str) = self.chunk_path(digest);
 
         let _lock = self.inner.chunk_store.mutex().lock().unwrap();
+        let _chunk_guard = self.lock_chunk_for_backend(digest)?;
 
         let mut counter = 0;
         let mut new_path = path.clone();
-- 
2.47.3



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

[pbs-devel] [PATCH proxmox-backup 13/17] fix #6961: datastore: verify: evict corrupt chunks from in-memory LRU cache

[Christian Ebner]

lock cache store on rename of fs backed datastores and additional
per-chunk file lock on s3 backend.

Chunks detected as corrupt have been renamed on both, the S3 backend
and the local datastore cache, but not evicted from the in-memory
cache containing the LRU chunk digests. This can lead to the chunks
being considered as already present if their digest is still cached,
and therefore not being re-inserted in the local store cache and S3
backend on backup upload.

Fix this by not only renaming the local datastore's chunk marker
file, but also removing it from the in-memory cache while holding the
chunk store mutex lock to exclude interference from concurrent chunk
inserts.

Reported-by: Fabian Grünbichler <f.gruenbichler@proxmox.com>
Fixes: https://bugzilla.proxmox.com/show_bug.cgi?id=6961
Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
---
 pbs-datastore/src/datastore.rs | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/pbs-datastore/src/datastore.rs b/pbs-datastore/src/datastore.rs
index e65f3a60c..1b3f3496c 100644
--- a/pbs-datastore/src/datastore.rs
+++ b/pbs-datastore/src/datastore.rs
@@ -2599,6 +2599,11 @@ impl DataStore {
             }
         }
 
+        if let Some(cache) = self.cache() {
+            // Locks are being held, so it is safe to call the method
+            let _ = unsafe { cache.remove(digest) };
+        }
+
         let result = match std::fs::rename(&path, &new_path) {
             Ok(_) => Ok(Some(new_path)),
             Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(None),
-- 
2.47.3



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

[pbs-devel] [PATCH proxmox-backup 14/17] datastore: add locking to protect against races on chunk insert for s3

[Christian Ebner]

Acquire the per-chunk file lock to get exclusive access to the chunk
on insert and make it possible to guarantee that s3 backend
operations and the local caches reflect the consistent state without
other operations such as verify chunk renaming or garbage collection
interfering.

Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
---
 pbs-datastore/src/datastore.rs | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/pbs-datastore/src/datastore.rs b/pbs-datastore/src/datastore.rs
index 1b3f3496c..b36424131 100644
--- a/pbs-datastore/src/datastore.rs
+++ b/pbs-datastore/src/datastore.rs
@@ -1867,6 +1867,7 @@ impl DataStore {
         digest: &[u8; 32],
         backend: &DatastoreBackend,
     ) -> Result<(bool, u64), Error> {
+        let _chunk_lock_guard = self.lock_chunk_for_backend(digest)?;
         match backend {
             DatastoreBackend::Filesystem => self.inner.chunk_store.insert_chunk(chunk, digest),
             DatastoreBackend::S3(s3_client) => self.insert_chunk_cached(chunk, digest, &s3_client),
@@ -1888,6 +1889,7 @@ impl DataStore {
         digest: &[u8; 32],
         backend: &DatastoreBackend,
     ) -> Result<(bool, u64), Error> {
+        let _chunk_lock_guard = self.lock_chunk_for_backend(digest)?;
         match backend {
             DatastoreBackend::Filesystem => self.inner.chunk_store.insert_chunk(chunk, digest),
             DatastoreBackend::S3(s3_client) => {
-- 
2.47.3



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

[pbs-devel] [PATCH proxmox-backup 15/17] GC: fix race with chunk upload/insert on s3 backends

[Christian Ebner]

The previous approach relying soly on local marker files was flawed
as it could not protect against all the possible races between chunk
upload to the s3 object store, insertion in the local datastore cache
and in-memory LRU cache.

Since these operations are now protected by getting a per-chunk file
lock, use that to check if it is safe to remove the chunk and do so
in a consistent manner by holding the chunk lock guard until it is
actually removed from the s3 backend and the caches.

Since an error when removing the chunk from cache could lead to
inconsistencies, GC must now fail in that case.

The chunk store lock is now only required on cache remove.

Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
---
 pbs-datastore/src/datastore.rs | 27 ++++++++++++++++++---------
 1 file changed, 18 insertions(+), 9 deletions(-)

diff --git a/pbs-datastore/src/datastore.rs b/pbs-datastore/src/datastore.rs
index b36424131..beca72e9a 100644
--- a/pbs-datastore/src/datastore.rs
+++ b/pbs-datastore/src/datastore.rs
@@ -1639,14 +1639,18 @@ impl DataStore {
 
             let mut delete_list = Vec::with_capacity(1000);
             loop {
-                let lock = self.inner.chunk_store.mutex().lock().unwrap();
-
                 for content in list_bucket_result.contents {
                     let (chunk_path, digest) = match self.chunk_path_from_object_key(&content.key) {
                         Some(path) => path,
                         None => continue,
                     };
 
+                    let timeout = std::time::Duration::from_secs(0);
+                    let _chunk_guard = match self.inner.chunk_store.lock_chunk(&digest, timeout) {
+                        Ok(guard) => guard,
+                        Err(_) => continue,
+                    };
+
                     // Check local markers (created or atime updated during phase1) and
                     // keep or delete chunk based on that.
                     let atime = match std::fs::metadata(&chunk_path) {
@@ -1675,10 +1679,10 @@ impl DataStore {
                             &mut gc_status,
                             || {
                                 if let Some(cache) = self.cache() {
-                                    // ignore errors, phase 3 will retry cleanup anyways
-                                    let _ = cache.remove(&digest);
+                                    let _guard = self.inner.chunk_store.mutex().lock().unwrap();
+                                    cache.remove(&digest)?;
                                 }
-                                delete_list.push(content.key);
+                                delete_list.push((content.key, _chunk_guard));
                                 Ok(())
                             },
                         )?;
@@ -1687,14 +1691,19 @@ impl DataStore {
                     chunk_count += 1;
                 }
 
-                drop(lock);
-
                 if !delete_list.is_empty() {
-                    let delete_objects_result =
-                        proxmox_async::runtime::block_on(s3_client.delete_objects(&delete_list))?;
+                    let delete_objects_result = proxmox_async::runtime::block_on(
+                        s3_client.delete_objects(
+                            &delete_list
+                                .iter()
+                                .map(|(key, _)| key.clone())
+                                .collect::<Vec<S3ObjectKey>>(),
+                        ),
+                    )?;
                     if let Some(_err) = delete_objects_result.error {
                         bail!("failed to delete some objects");
                     }
+                    // release all chunk guards
                     delete_list.clear();
                 }
 
-- 
2.47.3



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

[pbs-devel] [PATCH proxmox-backup 16/17] GC: lock chunk marker before cleanup in phase 3 on s3 backends

[Christian Ebner]

To make sure there is no race between atime check and deletion with
possible re-insertion.

By only acquiring the file lock if the chunk marker would be removed
and double stating, the file locking penalty is avoided for the other
cases.

Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
---
 pbs-datastore/src/chunk_store.rs | 28 +++++++++++++++++++++++++++-
 pbs-datastore/src/datastore.rs   |  2 ++
 2 files changed, 29 insertions(+), 1 deletion(-)

diff --git a/pbs-datastore/src/chunk_store.rs b/pbs-datastore/src/chunk_store.rs
index 49687b2fa..08519fe2b 100644
--- a/pbs-datastore/src/chunk_store.rs
+++ b/pbs-datastore/src/chunk_store.rs
@@ -5,6 +5,7 @@ use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
 use anyhow::{bail, format_err, Context, Error};
+use hex::FromHex;
 use tracing::{info, warn};
 
 use pbs_api_types::{DatastoreFSyncLevel, GarbageCollectionStatus};
@@ -22,7 +23,7 @@ use crate::data_blob::DataChunkBuilder;
 use crate::file_formats::{
     COMPRESSED_BLOB_MAGIC_1_0, ENCRYPTED_BLOB_MAGIC_1_0, UNCOMPRESSED_BLOB_MAGIC_1_0,
 };
-use crate::DataBlob;
+use crate::{DataBlob, LocalDatastoreLruCache};
 
 /// File system based chunk store
 pub struct ChunkStore {
@@ -366,6 +367,7 @@ impl ChunkStore {
         min_atime: i64,
         status: &mut GarbageCollectionStatus,
         worker: &dyn WorkerTaskContext,
+        cache: Option<&LocalDatastoreLruCache>,
     ) -> Result<(), Error> {
         // unwrap: only `None` in unit tests
         assert!(self.locker.is_some());
@@ -419,6 +421,30 @@ impl ChunkStore {
                         bad,
                         status,
                         || {
+                            if let Some(cache) = cache {
+                                // never lock bad chunks
+                                if filename.to_bytes().len() == 64 {
+                                    let digest = <[u8; 32]>::from_hex(filename.to_bytes())?;
+                                    match self.lock_chunk(&digest, Duration::from_secs(0)) {
+                                        Ok(_guard) => {
+                                            // don't remove if changed since locking
+                                            match fstatat(
+                                                Some(dirfd),
+                                                filename,
+                                                nix::fcntl::AtFlags::AT_SYMLINK_NOFOLLOW,
+                                            ) {
+                                                Ok(stat) if stat.st_atime < min_atime => {
+                                                    let _ = cache.remove(&digest);
+                                                    return Ok(());
+                                                }
+                                                _ => return Ok(()),
+                                            }
+                                        }
+                                        Err(_) => return Ok(()),
+                                    }
+                                }
+                            }
+
                             unlinkat(Some(dirfd), filename, UnlinkatFlags::NoRemoveDir).map_err(
                                 |err| {
                                     format_err!(
diff --git a/pbs-datastore/src/datastore.rs b/pbs-datastore/src/datastore.rs
index beca72e9a..33d589b67 100644
--- a/pbs-datastore/src/datastore.rs
+++ b/pbs-datastore/src/datastore.rs
@@ -1733,6 +1733,7 @@ impl DataStore {
                 min_atime,
                 &mut tmp_gc_status,
                 worker,
+                self.cache(),
             )?;
         } else {
             self.inner.chunk_store.sweep_unused_chunks(
@@ -1740,6 +1741,7 @@ impl DataStore {
                 min_atime,
                 &mut gc_status,
                 worker,
+                None,
             )?;
         }
 
-- 
2.47.3



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

[pbs-devel] [PATCH proxmox-backup 17/17] datastore: GC: drop overly verbose info message during s3 chunk sweep

[Christian Ebner]

This message does not provide much additional information and is
shown on correct operation anyways, leaving it however open for
misinterpretation of being an error/warning. Drop it in favor of
being less verbose and not potentially spam the task log in case of
many chunks being removed, still being visible in the final garbage
collection stats output.

Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
---
 pbs-datastore/src/datastore.rs | 1 -
 1 file changed, 1 deletion(-)

diff --git a/pbs-datastore/src/datastore.rs b/pbs-datastore/src/datastore.rs
index 33d589b67..b1aa85a72 100644
--- a/pbs-datastore/src/datastore.rs
+++ b/pbs-datastore/src/datastore.rs
@@ -1657,7 +1657,6 @@ impl DataStore {
                         Ok(stat) => stat.accessed()?,
                         Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
                             // File not found, delete by setting atime to unix epoch
-                            info!("Not found, mark for deletion: {}", content.key);
                             SystemTime::UNIX_EPOCH
                         }
                         Err(err) => return Err(err.into()),
-- 
2.47.3



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

Re: [pbs-devel] [PATCH proxmox-backup 10/17] datastore: implement per-chunk file locking helper for s3 backend

[Fabian Grünbichler]

On November 3, 2025 12:31 pm, Christian Ebner wrote:
> Adds a datastore helper method to create per-chunk file locks. These
> will be used to guard chunk operations on s3 backends to guarantee
> exclusive access when performing cache and backend operations.
> 
> Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
> ---
>  pbs-datastore/src/backup_info.rs |  2 +-
>  pbs-datastore/src/chunk_store.rs | 26 ++++++++++++++++++++++++++
>  pbs-datastore/src/datastore.rs   | 12 ++++++++++++
>  3 files changed, 39 insertions(+), 1 deletion(-)
> 
> diff --git a/pbs-datastore/src/backup_info.rs b/pbs-datastore/src/backup_info.rs
> index 4b10b6435..70c0fbe8a 100644
> --- a/pbs-datastore/src/backup_info.rs
> +++ b/pbs-datastore/src/backup_info.rs
> @@ -936,7 +936,7 @@ fn lock_file_path_helper(ns: &BackupNamespace, path: PathBuf) -> PathBuf {
>  /// deletion.
>  ///
>  /// It also creates the base directory for lock files.
> -fn lock_helper<F>(
> +pub(crate) fn lock_helper<F>(
>      store_name: &str,
>      path: &std::path::Path,
>      lock_fn: F,
> diff --git a/pbs-datastore/src/chunk_store.rs b/pbs-datastore/src/chunk_store.rs
> index ba7618e40..49687b2fa 100644
> --- a/pbs-datastore/src/chunk_store.rs
> +++ b/pbs-datastore/src/chunk_store.rs
> @@ -8,6 +8,7 @@ use anyhow::{bail, format_err, Context, Error};
>  use tracing::{info, warn};
>  
>  use pbs_api_types::{DatastoreFSyncLevel, GarbageCollectionStatus};
> +use pbs_config::BackupLockGuard;
>  use proxmox_io::ReadExt;
>  use proxmox_s3_client::S3Client;
>  use proxmox_sys::fs::{create_dir, create_path, file_type_from_file_stat, CreateOptions};
> @@ -16,6 +17,7 @@ use proxmox_sys::process_locker::{
>  };
>  use proxmox_worker_task::WorkerTaskContext;
>  
> +use crate::backup_info::DATASTORE_LOCKS_DIR;
>  use crate::data_blob::DataChunkBuilder;
>  use crate::file_formats::{
>      COMPRESSED_BLOB_MAGIC_1_0, ENCRYPTED_BLOB_MAGIC_1_0, UNCOMPRESSED_BLOB_MAGIC_1_0,
> @@ -759,6 +761,30 @@ impl ChunkStore {
>          ChunkStore::check_permissions(lockfile_path, 0o644)?;
>          Ok(())
>      }
> +
> +    /// Generates the path to the chunks lock file
> +    pub(crate) fn chunk_lock_path(&self, digest: &[u8]) -> PathBuf {
> +        let mut lock_path = Path::new(DATASTORE_LOCKS_DIR).join(self.name.clone());
> +        let digest_str = hex::encode(digest);
> +        lock_path.push(".chunks");
> +        let prefix = digest_to_prefix(digest);
> +        lock_path.push(&prefix);
> +        lock_path.push(&digest_str);
> +        lock_path

should we add "s3" or some suffix here, so that if we add another
backend in the future we already have specific paths?

> +    }
> +
> +    /// Get an exclusive lock on the chunks lock file
> +    pub(crate) fn lock_chunk(
> +        &self,
> +        digest: &[u8],
> +        timeout: Duration,
> +    ) -> Result<BackupLockGuard, Error> {
> +        let lock_path = self.chunk_lock_path(digest);
> +        let guard = crate::backup_info::lock_helper(self.name(), &lock_path, |path| {
> +            pbs_config::open_backup_lockfile(path, Some(timeout), true)
> +        })?;
> +        Ok(guard)
> +    }
>  }
>  
>  #[test]
> diff --git a/pbs-datastore/src/datastore.rs b/pbs-datastore/src/datastore.rs
> index 397c37e56..32f3562b3 100644
> --- a/pbs-datastore/src/datastore.rs
> +++ b/pbs-datastore/src/datastore.rs
> @@ -2568,6 +2568,18 @@ impl DataStore {
>          Ok(())
>      }
>  
> +    /// Locks the per chunk lock file if the backend requires it
> +    fn lock_chunk_for_backend(&self, digest: &[u8; 32]) -> Result<Option<BackupLockGuard>, Error> {
> +        // s3 put request times out after upload_size / 1 Kib/s, so about 2.3 hours for 8 MiB
> +        let timeout = Duration::from_secs(3 * 60 * 60);

could move into the S3 branch below.. or be made S3 specific in the
first place, since it is only called/effective there? the renaming
helper needs some rework then I guess..

but I am not sure if this logic here is really sound (any individual
caller waiting for longer than a single uploads max timeout might be
valid, since the locking is not fair and multiple locking attempts might
have queued up), I guess the instances where we end up taking this lock
are few enough that no progress over such a long time makes any progress
within reasonable time unlikely..

we currently take this lock for the duration of a chunk
upload/insertion, for the duration of a chunk rename after corruption
has been detected, and for a batch of GC chunk removal.

> +        match self.inner.backend_config.ty.unwrap_or_default() {
> +            DatastoreBackendType::Filesystem => Ok(None),
> +            DatastoreBackendType::S3 => {
> +                self.inner.chunk_store.lock_chunk(digest, timeout).map(Some)
> +            }
> +        }
> +    }
> +
>      /// Renames a corrupt chunk, returning the new path if the chunk was renamed successfully.
>      /// Returns with `Ok(None)` if the chunk source was not found.
>      pub fn rename_corrupt_chunk(&self, digest: &[u8; 32]) -> Result<Option<PathBuf>, Error> {
> -- 
> 2.47.3
> 
> 
> 
> _______________________________________________
> pbs-devel mailing list
> pbs-devel@lists.proxmox.com
> https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel
> 
> 
> 


_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

Re: [pbs-devel] [PATCH proxmox-backup 12/17] datastore: get per-chunk file lock for chunk rename on s3 backend

[Fabian Grünbichler]

On November 3, 2025 12:31 pm, Christian Ebner wrote:
> To guarantee exclusive access during s3 object store and cache
> operations, acquire the per-chunk file before renaming a chunk when
> the datastore is backed by s3.
> 
> This does not yet cover locking for the GC and chunk insert, part
> of subsequent changes.
> 
> Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
> ---
>  pbs-datastore/src/datastore.rs | 1 +
>  1 file changed, 1 insertion(+)
> 
> diff --git a/pbs-datastore/src/datastore.rs b/pbs-datastore/src/datastore.rs
> index e7ec87a7f..e65f3a60c 100644
> --- a/pbs-datastore/src/datastore.rs
> +++ b/pbs-datastore/src/datastore.rs
> @@ -2586,6 +2586,7 @@ impl DataStore {
>          let (path, digest_str) = self.chunk_path(digest);
>  
>          let _lock = self.inner.chunk_store.mutex().lock().unwrap();
> +        let _chunk_guard = self.lock_chunk_for_backend(digest)?;

lock inversion?

when inserting into S3, the chunk lock is obtained first, and then the
mutex is obtained for all insertions in chunk_store.insert_chunk while
the chunk lock is held.. while we do have a long timeout here, that
would still be quite bad..

>  
>          let mut counter = 0;
>          let mut new_path = path.clone();
> -- 
> 2.47.3
> 
> 
> 
> _______________________________________________
> pbs-devel mailing list
> pbs-devel@lists.proxmox.com
> https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel
> 
> 
> 


_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

Re: [pbs-devel] [PATCH proxmox-backup 04/17] api/datastore: move backup log upload by implementing datastore helper

[Fabian Grünbichler]

On November 3, 2025 12:31 pm, Christian Ebner wrote:
> In an effort to decouple the api from the datastore backend, move the
> backup task log upload to use a new add blob helper method of the
> datastore.
> 
> The new helper is fully sync and called on a blocking task, thereby
> now also solving the previously incorrectly blocking rename_file() in
> async context.
> 
> Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
> ---
>  pbs-datastore/src/datastore.rs | 22 ++++++++++++++++++++++
>  src/api2/admin/datastore.rs    | 25 +++++++------------------
>  2 files changed, 29 insertions(+), 18 deletions(-)
> 
> diff --git a/pbs-datastore/src/datastore.rs b/pbs-datastore/src/datastore.rs
> index 46600a88c..cc1267d78 100644
> --- a/pbs-datastore/src/datastore.rs
> +++ b/pbs-datastore/src/datastore.rs
> @@ -2453,4 +2453,26 @@ impl DataStore {
>          snapshot.destroy(false, &backend)?;
>          Ok(())
>      }
> +
> +    /// Adds the blob to the given snapshot.
> +    /// Requires the caller to hold the exclusive lock.
> +    pub fn add_blob(
> +        self: &Arc<Self>,
> +        filename: &str,
> +        snapshot: BackupDir,
> +        blob: DataBlob,

should this get a backend parameter to not require instantiating a new
one in contexts where lots of blobs might be added (backup env, sync
job)?

> +    ) -> Result<(), Error> {
> +        if let DatastoreBackend::S3(s3_client) = self.backend()? {
> +            let object_key = crate::s3::object_key_from_path(&snapshot.relative_path(), filename)
> +                .context("invalid client log object key")?;
> +            let data = hyper::body::Bytes::copy_from_slice(blob.raw_data());
> +            proxmox_async::runtime::block_on(s3_client.upload_replace_with_retry(object_key, data))
> +                .context("failed to upload client log to s3 backend")?;
> +        };
> +
> +        let mut path = snapshot.full_path();
> +        path.push(filename);
> +        replace_file(&path, blob.raw_data(), CreateOptions::new(), false)?;
> +        Ok(())
> +    }

the backup env also has this, and should switch to this new helper:

        // always verify blob/CRC at server side
        let blob = DataBlob::load_from_reader(&mut &data[..])?;

        let raw_data = blob.raw_data();
        if let DatastoreBackend::S3(s3_client) = &self.backend {
            let object_key = pbs_datastore::s3::object_key_from_path(
                &self.backup_dir.relative_path(),
                file_name,
            )
            .context("invalid blob object key")?;
            let data = hyper::body::Bytes::copy_from_slice(raw_data);
            proxmox_async::runtime::block_on(
                s3_client.upload_replace_with_retry(object_key.clone(), data),
            )
            .context("failed to upload blob to s3 backend")?;
            self.log(format!("Uploaded blob to object store: {object_key}"))
        }

        replace_file(&path, raw_data, CreateOptions::new(), false)?;


>  }
> diff --git a/src/api2/admin/datastore.rs b/src/api2/admin/datastore.rs
> index 763440df9..6881b4093 100644
> --- a/src/api2/admin/datastore.rs
> +++ b/src/api2/admin/datastore.rs
> @@ -28,9 +28,7 @@ use proxmox_router::{
>  use proxmox_rrd_api_types::{RrdMode, RrdTimeframe};
>  use proxmox_schema::*;
>  use proxmox_sortable_macro::sortable;
> -use proxmox_sys::fs::{
> -    file_read_firstline, file_read_optional_string, replace_file, CreateOptions,
> -};
> +use proxmox_sys::fs::{file_read_firstline, file_read_optional_string, CreateOptions};
>  use proxmox_time::CalendarEvent;
>  use proxmox_worker_task::WorkerTaskContext;
>  
> @@ -63,7 +61,7 @@ use pbs_datastore::manifest::BackupManifest;
>  use pbs_datastore::prune::compute_prune_info;
>  use pbs_datastore::{
>      check_backup_owner, ensure_datastore_is_mounted, task_tracking, BackupDir, DataStore,
> -    DatastoreBackend, LocalChunkReader, StoreProgress,
> +    LocalChunkReader, StoreProgress,
>  };
>  use pbs_tools::json::required_string_param;
>  use proxmox_rest_server::{formatter, worker_is_active, WorkerTask};
> @@ -1536,20 +1534,11 @@ pub fn upload_backup_log(
>          // always verify blob/CRC at server side
>          let blob = DataBlob::load_from_reader(&mut &data[..])?;
>  
> -        if let DatastoreBackend::S3(s3_client) = datastore.backend()? {
> -            let object_key = pbs_datastore::s3::object_key_from_path(
> -                &backup_dir.relative_path(),
> -                file_name.as_ref(),
> -            )
> -            .context("invalid client log object key")?;
> -            let data = hyper::body::Bytes::copy_from_slice(blob.raw_data());
> -            s3_client
> -                .upload_replace_with_retry(object_key, data)
> -                .await
> -                .context("failed to upload client log to s3 backend")?;
> -        };
> -
> -        replace_file(&path, blob.raw_data(), CreateOptions::new(), false)?;
> +        tokio::task::spawn_blocking(move || {
> +            datastore.add_blob(file_name.as_ref(), backup_dir, blob)
> +        })
> +        .await
> +        .map_err(|err| format_err!("{err:#?}"))??;
>  
>          // fixme: use correct formatter
>          Ok(formatter::JSON_FORMATTER.format_data(Value::Null, &*rpcenv))
> -- 
> 2.47.3
> 
> 
> 
> _______________________________________________
> pbs-devel mailing list
> pbs-devel@lists.proxmox.com
> https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel
> 
> 
> 


_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

Re: [pbs-devel] [PATCH proxmox-backup 02/17] api/datastore: move group notes setting to the datastore

[Fabian Grünbichler]

On November 3, 2025 12:31 pm, Christian Ebner wrote:
> In an effort to abstract away the datastore backend related logic
> from the api, move the set_group_notes to a helper method on the
> datastore.
> 
> The new helper method now also assures that the exclusive lock on
> the backup group is acquired before updating the notes, in order to
> avoid possible race conditions, e.g. with backup group destruction.
> 
> Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
> ---
>  pbs-datastore/src/datastore.rs | 26 ++++++++++++++++++++++++++
>  src/api2/admin/datastore.rs    | 17 ++++-------------
>  2 files changed, 30 insertions(+), 13 deletions(-)
> 
> diff --git a/pbs-datastore/src/datastore.rs b/pbs-datastore/src/datastore.rs
> index 127ba1c81..45f315aeb 100644
> --- a/pbs-datastore/src/datastore.rs
> +++ b/pbs-datastore/src/datastore.rs
> @@ -2418,4 +2418,30 @@ impl DataStore {
>              .map_err(|err| format_err!("{err:#}"))?;
>          Ok((backend_type, Some(s3_client)))
>      }
> +
> +    /// Creates or updates the notes associated with a backup group.
> +    /// Acquires exclusive lock on the backup group.
> +    pub fn set_group_notes(
> +        self: &Arc<Self>,
> +        notes: String,
> +        backup_group: BackupGroup,
> +    ) -> Result<(), Error> {
> +        let _lock = backup_group.lock().context("failed to lock backup group")?;

this takes an exclusive lock on group, which means all sorts of other
operations (including creating new snapshots?) are blocked while it is
held

> +
> +        if let DatastoreBackend::S3(s3_client) = self.backend()? {
> +            let mut path = backup_group.backup_ns().path();
> +            path.push(backup_group.group().to_string());
> +            let object_key = crate::s3::object_key_from_path(&path, "notes")
> +                .context("invalid owner file object key")?;
> +            let data = hyper::body::Bytes::copy_from_slice(notes.as_bytes());
> +            let _is_duplicate = proxmox_async::runtime::block_on(
> +                s3_client.upload_replace_with_retry(object_key, data),

but this can take a while, right? FWIW, the same is true of setting the
backup owner..

> +            )
> +            .context("failed to set notes on s3 backend")?;
> +        }
> +        let notes_path = self.group_notes_path(backup_group.backup_ns(), backup_group.group());
> +        replace_file(notes_path, notes.as_bytes(), CreateOptions::new(), false)
> +            .context("failed to replace group notes file")?;
> +        Ok(())
> +    }
>  }
> diff --git a/src/api2/admin/datastore.rs b/src/api2/admin/datastore.rs
> index d192ee390..131cdae51 100644
> --- a/src/api2/admin/datastore.rs
> +++ b/src/api2/admin/datastore.rs
> @@ -2009,19 +2009,10 @@ pub fn set_group_notes(
>          &backup_group,
>      )?;
>  
> -    if let DatastoreBackend::S3(s3_client) = datastore.backend()? {
> -        let mut path = ns.path();
> -        path.push(backup_group.to_string());
> -        let object_key = pbs_datastore::s3::object_key_from_path(&path, "notes")
> -            .context("invalid owner file object key")?;
> -        let data = hyper::body::Bytes::copy_from_slice(notes.as_bytes());
> -        let _is_duplicate =
> -            proxmox_async::runtime::block_on(s3_client.upload_replace_with_retry(object_key, data))
> -                .context("failed to set notes on s3 backend")?;
> -    }
> -    let notes_path = datastore.group_notes_path(&ns, &backup_group);
> -    replace_file(notes_path, notes.as_bytes(), CreateOptions::new(), false)?;
> -
> +    let backup_group = datastore.backup_group(ns, backup_group);
> +    datastore
> +        .set_group_notes(notes, backup_group)
> +        .map_err(|err| format_err!("failed to set group notes - {err:#?}"))?;
>      Ok(())
>  }
>  
> -- 
> 2.47.3
> 
> 
> 
> _______________________________________________
> pbs-devel mailing list
> pbs-devel@lists.proxmox.com
> https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel
> 
> 
> 


_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

Re: [pbs-devel] [PATCH proxmox-backup 01/17] sync: pull: instantiate backend only once per sync job

[Fabian Grünbichler]

Reviewed-by: Fabian Grünbichler <f.gruenbichler@proxmox.com>

On November 3, 2025 12:31 pm, Christian Ebner wrote:
> Currently the target datastores' backend is instatziated for each
> chunk to be inserted, which on s3 backed datastores leads to the
> s3-client being re-instantiated and a new connection being
> established.
> 
> Optimize this by only creating the backend once and sharing it for
> all the chunk inserts to be performed.
> 
> Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
> ---
>  src/server/pull.rs | 30 +++++++++++++++++++++---------
>  1 file changed, 21 insertions(+), 9 deletions(-)
> 
> diff --git a/src/server/pull.rs b/src/server/pull.rs
> index 817b57ac5..de8b140bc 100644
> --- a/src/server/pull.rs
> +++ b/src/server/pull.rs
> @@ -38,6 +38,8 @@ use crate::tools::parallel_handler::ParallelHandler;
>  pub(crate) struct PullTarget {
>      store: Arc<DataStore>,
>      ns: BackupNamespace,
> +    // Contains the active S3Client in case of S3 backend
> +    backend: DatastoreBackend,
>  }
>  
>  /// Parameters for a pull operation.
> @@ -114,10 +116,9 @@ impl PullParameters {
>                  ns: remote_ns,
>              })
>          };
> -        let target = PullTarget {
> -            store: DataStore::lookup_datastore(store, Some(Operation::Write))?,
> -            ns,
> -        };
> +        let store = DataStore::lookup_datastore(store, Some(Operation::Write))?;
> +        let backend = store.backend()?;
> +        let target = PullTarget { store, ns, backend };
>  
>          let group_filter = group_filter.unwrap_or_default();
>  
> @@ -141,6 +142,7 @@ async fn pull_index_chunks<I: IndexFile>(
>      target: Arc<DataStore>,
>      index: I,
>      downloaded_chunks: Arc<Mutex<HashSet<[u8; 32]>>>,
> +    backend: &DatastoreBackend,
>  ) -> Result<SyncStats, Error> {
>      use futures::stream::{self, StreamExt, TryStreamExt};
>  
> @@ -162,13 +164,14 @@ async fn pull_index_chunks<I: IndexFile>(
>      );
>  
>      let target2 = target.clone();
> +    let backend = backend.clone();
>      let verify_pool = ParallelHandler::new(
>          "sync chunk writer",
>          4,
>          move |(chunk, digest, size): (DataBlob, [u8; 32], u64)| {
>              // println!("verify and write {}", hex::encode(&digest));
>              chunk.verify_unencrypted(size as usize, &digest)?;
> -            match target2.backend()? {
> +            match &backend {
>                  DatastoreBackend::Filesystem => {
>                      target2.insert_chunk(&chunk, &digest)?;
>                  }
> @@ -283,6 +286,7 @@ async fn pull_single_archive<'a>(
>      snapshot: &'a pbs_datastore::BackupDir,
>      archive_info: &'a FileInfo,
>      downloaded_chunks: Arc<Mutex<HashSet<[u8; 32]>>>,
> +    backend: &DatastoreBackend,
>  ) -> Result<SyncStats, Error> {
>      let archive_name = &archive_info.filename;
>      let mut path = snapshot.full_path();
> @@ -317,6 +321,7 @@ async fn pull_single_archive<'a>(
>                      snapshot.datastore().clone(),
>                      index,
>                      downloaded_chunks,
> +                    backend,
>                  )
>                  .await?;
>                  sync_stats.add(stats);
> @@ -339,6 +344,7 @@ async fn pull_single_archive<'a>(
>                      snapshot.datastore().clone(),
>                      index,
>                      downloaded_chunks,
> +                    backend,
>                  )
>                  .await?;
>                  sync_stats.add(stats);
> @@ -495,15 +501,21 @@ async fn pull_snapshot<'a>(
>              }
>          }
>  
> -        let stats =
> -            pull_single_archive(reader.clone(), snapshot, item, downloaded_chunks.clone()).await?;
> +        let stats = pull_single_archive(
> +            reader.clone(),
> +            snapshot,
> +            item,
> +            downloaded_chunks.clone(),
> +            &params.target.backend,

nit: this is used 3 times here, and could be pulled into a binding as
well..

> +        )
> +        .await?;
>          sync_stats.add(stats);
>      }
>  
>      if let Err(err) = std::fs::rename(&tmp_manifest_name, &manifest_name) {
>          bail!("Atomic rename file {:?} failed - {}", manifest_name, err);
>      }
> -    if let DatastoreBackend::S3(s3_client) = snapshot.datastore().backend()? {
> +    if let DatastoreBackend::S3(s3_client) = &params.target.backend {
>          let object_key = pbs_datastore::s3::object_key_from_path(
>              &snapshot.relative_path(),
>              MANIFEST_BLOB_NAME.as_ref(),
> @@ -520,7 +532,7 @@ async fn pull_snapshot<'a>(
>      if !client_log_name.exists() {
>          reader.try_download_client_log(&client_log_name).await?;
>          if client_log_name.exists() {
> -            if let DatastoreBackend::S3(s3_client) = snapshot.datastore().backend()? {
> +            if let DatastoreBackend::S3(s3_client) = &params.target.backend {
>                  let object_key = pbs_datastore::s3::object_key_from_path(
>                      &snapshot.relative_path(),
>                      CLIENT_LOG_BLOB_NAME.as_ref(),
> -- 
> 2.47.3
> 
> 
> 
> _______________________________________________
> pbs-devel mailing list
> pbs-devel@lists.proxmox.com
> https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel
> 
> 
> 


_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel