[PATCH proxmox-backup 0/2] fix #7606: verify: s3: handle transient fetch errors and add retry

[PATCH proxmox-backup 0/2] fix #7606: verify: s3: handle transient fetch errors and add retry

[Christian Ebner]

Transient errors during fetching of object from s3 backend currently
lead to snapshots being marked as verify failed. Subsequent runs of
verification on the same snapshot might succeed without issues, since
chunks are only flagged as bad if the chunk blob object could not be
decoded or content verification failed.

To fix this, discriminate between transient errors and permanent
errors and account for these separately. For snapshots encountering
only transient errors, but no permanent errors during verification,
do not set or strip the verify okay state, but always keep a
pre-existing verification failed state. This is to never allow reuse
of a snapshot as reference for backup/sync jobs, when the state
was bad before and could not be verified again. For the other case
there is not much to loose, if it was okay or not verified before
falling back to not verified reflects the current snapshot state.

Link to the bugtracker issue:
https://bugzilla.proxmox.com/show_bug.cgi?id=7606


proxmox-backup:

Christian Ebner (2):
  fix #7606: verify: s3: handle transient chunk load errors
  verify: s3: retry on transient response body collection errors

 src/backup/verify.rs | 235 ++++++++++++++++++++++++++++++-------------
 1 file changed, 165 insertions(+), 70 deletions(-)


Summary over all repositories:
  1 files changed, 165 insertions(+), 70 deletions(-)

-- 
Generated by murpp 0.11.0

[PATCH proxmox-backup 1/2] fix #7606: verify: s3: handle transient chunk load errors

[Christian Ebner]

Fetching an object for verification might fail when sending the
request to the S3 API or while receiving the response. Both are
currently handled as hard errors, leading to the manifest state
being set to verify failed.
This however does not reflect the actual state, as a subsequent
verification can work just fine. Therefore, distinguish these error
kinds by enum variants, the manifest only marked as failed
verification if the error is considered permanent.
If only transient errors but no permanent errors occur, strip any
verify okay state, since it is not possible to determine the state it
is in, but always keep a verify error state in order to avoid
allowing backups to reuse this snapshots as reference.

Fixes: https://bugzilla.proxmox.com/show_bug.cgi?id=7606
Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
---
 src/backup/verify.rs | 145 +++++++++++++++++++++++++++++++++----------
 1 file changed, 112 insertions(+), 33 deletions(-)

diff --git a/src/backup/verify.rs b/src/backup/verify.rs
index aeb78c032..e85e120e5 100644
--- a/src/backup/verify.rs
+++ b/src/backup/verify.rs
@@ -4,7 +4,7 @@ use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
 use std::sync::{Arc, Mutex};
 use std::time::Instant;
 
-use anyhow::{Error, bail};
+use anyhow::{Error, format_err};
 use http_body_util::BodyExt;
 use tracing::{error, info, warn};
 
@@ -39,6 +39,7 @@ struct IndexVerifyState {
     read_bytes: AtomicU64,
     decoded_bytes: AtomicU64,
     errors: AtomicUsize,
+    fetch_errors: AtomicUsize,
     datastore: Arc<DataStore>,
     corrupt_chunks: Arc<Mutex<HashSet<[u8; 32]>>>,
     verified_chunks: Arc<Mutex<HashSet<[u8; 32]>>>,
@@ -55,6 +56,7 @@ impl IndexVerifyState {
             read_bytes: AtomicU64::new(0),
             decoded_bytes: AtomicU64::new(0),
             errors: AtomicUsize::new(0),
+            fetch_errors: AtomicUsize::new(0),
             datastore: Arc::clone(datastore),
             corrupt_chunks: Arc::clone(corrupt_chunks),
             verified_chunks: Arc::clone(verified_chunks),
@@ -63,6 +65,11 @@ impl IndexVerifyState {
     }
 }
 
+enum VerificationError {
+    Transient(Error),
+    Permanent(Error),
+}
+
 impl VerifyWorker {
     /// Creates a new VerifyWorker for a given task worker and datastore.
     pub fn new(
@@ -95,27 +102,38 @@ impl VerifyWorker {
         })
     }
 
-    fn verify_blob(backup_dir: &BackupDir, info: &FileInfo) -> Result<(), Error> {
-        let blob = backup_dir.load_blob(info.filename.as_ref())?;
+    fn verify_blob(backup_dir: &BackupDir, info: &FileInfo) -> Result<(), VerificationError> {
+        let blob = backup_dir
+            .load_blob(info.filename.as_ref())
+            .map_err(VerificationError::Permanent)?;
 
         let raw_size = blob.raw_size();
         if raw_size != info.size {
-            bail!("wrong size ({} != {})", info.size, raw_size);
+            return Err(VerificationError::Permanent(format_err!(
+                "wrong size ({} != {})",
+                info.size,
+                raw_size
+            )));
         }
 
         let csum = openssl::sha::sha256(blob.raw_data());
         if csum != info.csum {
-            bail!("wrong index checksum");
+            return Err(VerificationError::Permanent(format_err!(
+                "wrong index checksum"
+            )));
         }
 
-        match blob.crypt_mode()? {
+        match blob.crypt_mode().map_err(VerificationError::Permanent)? {
             CryptMode::Encrypt => Ok(()),
             CryptMode::None => {
                 // digest already verified above
-                blob.decode(None, None)?;
+                blob.decode(None, None)
+                    .map_err(VerificationError::Permanent)?;
                 Ok(())
             }
-            CryptMode::SignOnly => bail!("Invalid CryptMode for blob"),
+            CryptMode::SignOnly => Err(VerificationError::Permanent(format_err!(
+                "Invalid CryptMode for blob"
+            ))),
         }
     }
 
@@ -123,7 +141,7 @@ impl VerifyWorker {
         &self,
         index: Box<dyn IndexFile + Send>,
         crypt_mode: CryptMode,
-    ) -> Result<(), Error> {
+    ) -> Result<(), VerificationError> {
         let verify_state = Arc::new(IndexVerifyState::new(
             &self.datastore,
             &self.corrupt_chunks,
@@ -190,7 +208,8 @@ impl VerifyWorker {
 
         let chunk_list = self
             .datastore
-            .get_chunks_in_order(&*index, skip_chunk, check_abort)?;
+            .get_chunks_in_order(&*index, skip_chunk, check_abort)
+            .map_err(VerificationError::Transient)?;
 
         let reader_pool = ParallelHandler::new("read chunks", self.read_threads, {
             let decoder_pool = decoder_pool.channel();
@@ -207,8 +226,12 @@ impl VerifyWorker {
             }
         });
         for (pos, _) in chunk_list {
-            self.worker.check_abort()?;
-            self.worker.fail_on_shutdown()?;
+            self.worker
+                .check_abort()
+                .map_err(VerificationError::Transient)?;
+            self.worker
+                .fail_on_shutdown()
+                .map_err(VerificationError::Transient)?;
 
             let info = index.chunk_info(pos).unwrap();
 
@@ -217,10 +240,14 @@ impl VerifyWorker {
                 continue; // already verified or marked corrupt
             }
 
-            reader_pool.send(info)?;
+            reader_pool
+                .send(info)
+                .map_err(|err| VerificationError::Transient(err.into()))?;
         }
 
-        reader_pool.complete()?;
+        reader_pool
+            .complete()
+            .map_err(|err| VerificationError::Transient(err.into()))?;
 
         let elapsed = verify_state.start_time.elapsed().as_secs_f64();
 
@@ -240,7 +267,15 @@ impl VerifyWorker {
         );
 
         if verify_state.errors.load(Ordering::SeqCst) > 0 {
-            bail!("chunks could not be verified");
+            return Err(VerificationError::Permanent(format_err!(
+                "chunks could not be verified"
+            )));
+        }
+
+        if verify_state.fetch_errors.load(Ordering::SeqCst) > 0 {
+            return Err(VerificationError::Transient(format_err!(
+                "chunks could not be fetched"
+            )));
         }
 
         Ok(())
@@ -296,7 +331,7 @@ impl VerifyWorker {
                                 }
                             }
                             Err(err) => {
-                                verify_state.errors.fetch_add(1, Ordering::SeqCst);
+                                verify_state.fetch_errors.fetch_add(1, Ordering::SeqCst);
                                 error!("can't verify chunk, load failed - {err}");
                             }
                         }
@@ -310,7 +345,7 @@ impl VerifyWorker {
                         ),
                     ),
                     Err(err) => {
-                        verify_state.errors.fetch_add(1, Ordering::SeqCst);
+                        verify_state.fetch_errors.fetch_add(1, Ordering::SeqCst);
                         error!("can't verify chunk, load failed - {err}");
                     }
                 }
@@ -332,37 +367,63 @@ impl VerifyWorker {
         }
     }
 
-    fn verify_fixed_index(&self, backup_dir: &BackupDir, info: &FileInfo) -> Result<(), Error> {
+    fn verify_fixed_index(
+        &self,
+        backup_dir: &BackupDir,
+        info: &FileInfo,
+    ) -> Result<(), VerificationError> {
         let mut path = backup_dir.relative_path();
         path.push(info.filename.as_ref());
 
-        let index = self.datastore.open_fixed_reader(&path)?;
+        let index = self
+            .datastore
+            .open_fixed_reader(&path)
+            .map_err(VerificationError::Permanent)?;
 
         let (csum, size) = index.compute_csum();
         if size != info.size {
-            bail!("wrong size ({} != {})", info.size, size);
+            return Err(VerificationError::Permanent(format_err!(
+                "wrong size ({} != {})",
+                info.size,
+                size
+            )));
         }
 
         if csum != info.csum {
-            bail!("wrong index checksum");
+            return Err(VerificationError::Permanent(format_err!(
+                "wrong index checksum"
+            )));
         }
 
         self.verify_index_chunks(Box::new(index), info.chunk_crypt_mode())
     }
 
-    fn verify_dynamic_index(&self, backup_dir: &BackupDir, info: &FileInfo) -> Result<(), Error> {
+    fn verify_dynamic_index(
+        &self,
+        backup_dir: &BackupDir,
+        info: &FileInfo,
+    ) -> Result<(), VerificationError> {
         let mut path = backup_dir.relative_path();
         path.push(info.filename.as_ref());
 
-        let index = self.datastore.open_dynamic_reader(&path)?;
+        let index = self
+            .datastore
+            .open_dynamic_reader(&path)
+            .map_err(VerificationError::Permanent)?;
 
         let (csum, size) = index.compute_csum();
         if size != info.size {
-            bail!("wrong size ({} != {})", info.size, size);
+            return Err(VerificationError::Permanent(format_err!(
+                "wrong size ({} != {})",
+                info.size,
+                size
+            )));
         }
 
         if csum != info.csum {
-            bail!("wrong index checksum");
+            return Err(VerificationError::Permanent(format_err!(
+                "wrong index checksum"
+            )));
         }
 
         self.verify_index_chunks(Box::new(index), info.chunk_crypt_mode())
@@ -438,7 +499,7 @@ impl VerifyWorker {
 
         let mut error_count = 0;
 
-        let mut verify_result = VerifyState::Ok;
+        let mut verify_result = Some(VerifyState::Ok);
         for info in manifest.files() {
             let result = proxmox_lang::try_block!({
                 info!("  check {}", info.filename);
@@ -453,22 +514,40 @@ impl VerifyWorker {
             self.worker.fail_on_shutdown()?;
 
             if let Err(err) = result {
+                let err = match err {
+                    VerificationError::Permanent(err) => {
+                        verify_result = Some(VerifyState::Failed);
+                        err
+                    }
+                    VerificationError::Transient(err) => {
+                        if let Some(SnapshotVerifyState {
+                            state: VerifyState::Failed,
+                            ..
+                        }) = manifest.verify_state()?
+                        {
+                            verify_result = Some(VerifyState::Failed);
+                        } else if let Some(VerifyState::Ok) = verify_result {
+                            verify_result = None;
+                        }
+                        err
+                    }
+                };
                 info!(
                     "verify {datastore_name}:{backup_dir_name}/{file_name} failed: {err}",
                     file_name = info.filename,
                 );
                 error_count += 1;
-                verify_result = VerifyState::Failed;
             }
         }
 
-        let verify_state = SnapshotVerifyState {
-            state: verify_result,
-            upid,
-        };
-
         if let Err(err) = {
-            let verify_state = serde_json::to_value(verify_state)?;
+            let verify_state = match verify_result {
+                Some(state) => {
+                    let state = SnapshotVerifyState { state, upid };
+                    serde_json::to_value(state)?
+                }
+                None => serde_json::Value::Null,
+            };
             backup_dir.update_manifest(&self.datastore.backend()?, |manifest| {
                 manifest.unprotected["verify_state"] = verify_state;
             })
-- 
2.47.3

[PATCH proxmox-backup 2/2] verify: s3: retry on transient response body collection errors

[Christian Ebner]

If collecting the response body for the S3 object fetching requests
fail, retry with exponential backoff time and only consider this
as transient fetch error on retry exhaustion. Sending the requests
is already retried by the s3-client, so not to be repeated.

This avoids that an otherwise long running verification which already
downloaded a lot of chunks needs to be redone due to transient
connection issues.

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

diff --git a/src/backup/verify.rs b/src/backup/verify.rs
index e85e120e5..b558bcaf1 100644
--- a/src/backup/verify.rs
+++ b/src/backup/verify.rs
@@ -2,7 +2,7 @@ use pbs_config::BackupLockGuard;
 use std::collections::HashSet;
 use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
 use std::sync::{Arc, Mutex};
-use std::time::Instant;
+use std::time::{Duration, Instant};
 
 use anyhow::{Error, format_err};
 use http_body_util::BodyExt;
@@ -305,48 +305,64 @@ impl VerifyWorker {
             },
             DatastoreBackend::S3(s3_client) => {
                 let object_key = pbs_datastore::s3::object_key_from_digest(&info.digest)?;
-                match proxmox_async::runtime::block_on(s3_client.get_object(object_key)) {
-                    Ok(Some(response)) => {
-                        match proxmox_async::runtime::block_on(response.content.collect()) {
-                            Ok(raw_chunk) => {
-                                match DataBlob::from_raw(raw_chunk.to_bytes().to_vec()) {
-                                    Ok(chunk) => {
-                                        let size = info.size();
-                                        verify_state
-                                            .read_bytes
-                                            .fetch_add(chunk.raw_size(), Ordering::SeqCst);
-                                        decoder_pool.send((chunk, info.digest, size))?;
-                                        verify_state
-                                            .decoded_bytes
-                                            .fetch_add(size, Ordering::SeqCst);
-                                    }
-                                    Err(err) => Self::add_corrupt_chunk(
-                                        verify_state,
-                                        info.digest,
-                                        &format!(
-                                            "can't verify chunk with digest {} - {err}",
-                                            hex::encode(info.digest)
+
+                for retry in 0..3 {
+                    let verify_state = Arc::clone(&verify_state);
+                    match proxmox_async::runtime::block_on(s3_client.get_object(object_key.clone()))
+                    {
+                        Ok(Some(response)) => {
+                            match proxmox_async::runtime::block_on(response.content.collect()) {
+                                Ok(raw_chunk) => {
+                                    match DataBlob::from_raw(raw_chunk.to_bytes().to_vec()) {
+                                        Ok(chunk) => {
+                                            let size = info.size();
+                                            verify_state
+                                                .read_bytes
+                                                .fetch_add(chunk.raw_size(), Ordering::SeqCst);
+                                            decoder_pool.send((chunk, info.digest, size))?;
+                                            verify_state
+                                                .decoded_bytes
+                                                .fetch_add(size, Ordering::SeqCst);
+                                        }
+                                        Err(err) => Self::add_corrupt_chunk(
+                                            verify_state,
+                                            info.digest,
+                                            &format!(
+                                                "can't verify chunk with digest {} - {err}",
+                                                hex::encode(info.digest)
+                                            ),
                                         ),
-                                    ),
+                                    }
+                                    break;
+                                }
+                                Err(err) => {
+                                    if retry < 3 {
+                                        warn!("retry reading chunk, load failed - {err}");
+                                        let backoff = Duration::from_secs(3_u64.pow(retry));
+                                        std::thread::sleep(backoff);
+                                    } else {
+                                        verify_state.fetch_errors.fetch_add(1, Ordering::SeqCst);
+                                        error!("can't verify chunk, load failed - {err}");
+                                    }
                                 }
-                            }
-                            Err(err) => {
-                                verify_state.fetch_errors.fetch_add(1, Ordering::SeqCst);
-                                error!("can't verify chunk, load failed - {err}");
                             }
                         }
-                    }
-                    Ok(None) => Self::add_corrupt_chunk(
-                        verify_state,
-                        info.digest,
-                        &format!(
-                            "can't verify missing chunk with digest {}",
-                            hex::encode(info.digest)
-                        ),
-                    ),
-                    Err(err) => {
-                        verify_state.fetch_errors.fetch_add(1, Ordering::SeqCst);
-                        error!("can't verify chunk, load failed - {err}");
+                        Ok(None) => {
+                            Self::add_corrupt_chunk(
+                                verify_state,
+                                info.digest,
+                                &format!(
+                                    "can't verify missing chunk with digest {}",
+                                    hex::encode(info.digest)
+                                ),
+                            );
+                            break;
+                        }
+                        Err(err) => {
+                            verify_state.fetch_errors.fetch_add(1, Ordering::SeqCst);
+                            error!("can't verify chunk, load failed - {err}");
+                            break;
+                        }
                     }
                 }
             }
-- 
2.47.3