[PATCH proxmox-backup v2 3/5] fix #7606: verify: s3: handle transient chunk load errors

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

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 and propagate these error kinds by enum
variants, adapting return types and call sites accordingly.

Handle the variants respectively when setting/updating the verify
state for the manifest: if only transient errors, but no permanent
errors occur, strip any verify okay state since then it is not
possible to determine if the snapshot is still fine. That is best
represented by not having a verify state, as consequence sync jobs
with verified-only filtering will then exclude such snapshots as
well.

Independent of the error variant, always keep a previously set verify
error state in order to avoid allowing backups to reuse this snapshot
as reference. This covers both the new state as well as the manifest
state prior to verification.

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

diff --git a/src/backup/verify.rs b/src/backup/verify.rs
index 4d497c219..670a899ca 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,11 +240,17 @@ 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()?;
-        decoder_pool.complete()?;
+        reader_pool
+            .complete()
+            .map_err(|err| VerificationError::Transient(err.into()))?;
+        decoder_pool
+            .complete()
+            .map_err(|err| VerificationError::Transient(err.into()))?;
 
         let elapsed = verify_state.start_time.elapsed().as_secs_f64();
 
@@ -235,13 +264,22 @@ impl VerifyWorker {
         let decode_speed = decoded_bytes_mib / elapsed;
 
         let error_count = verify_state.errors.load(Ordering::SeqCst);
+        let fetch_error_count = verify_state.fetch_errors.load(Ordering::SeqCst);
 
         info!(
-            "  verified {read_bytes_mib:.2}/{decoded_bytes_mib:.2} MiB in {elapsed:.2} seconds, speed {read_speed:.2}/{decode_speed:.2} MiB/s ({error_count} errors)"
+            "  verified {read_bytes_mib:.2}/{decoded_bytes_mib:.2} MiB in {elapsed:.2} seconds, speed {read_speed:.2}/{decode_speed:.2} MiB/s ({error_count} errors, {fetch_error_count} fetch errors)"
         );
 
-        if verify_state.errors.load(Ordering::SeqCst) > 0 {
-            bail!("chunks could not be verified");
+        if error_count > 0 {
+            return Err(VerificationError::Permanent(format_err!(
+                "chunks could not be verified"
+            )));
+        }
+
+        if fetch_error_count > 0 {
+            return Err(VerificationError::Transient(format_err!(
+                "chunks could not be fetched"
+            )));
         }
 
         Ok(())
@@ -297,7 +335,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}");
                             }
                         }
@@ -311,7 +349,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}");
                     }
                 }
@@ -333,37 +371,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())
@@ -439,7 +503,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() {
             info!("  check {}", info.filename);
             let result = match info.filename.archive_type() {
@@ -451,28 +515,60 @@ impl VerifyWorker {
             self.worker.check_abort()?;
             self.worker.fail_on_shutdown()?;
 
-            if let Err(err) = result {
-                info!(
-                    "verify {datastore_name}:{backup_dir_name}/{file_name} failed: {err}",
-                    file_name = info.filename,
-                );
-                error_count += 1;
-                verify_result = VerifyState::Failed;
+            match result {
+                Ok(_) => (),
+                Err(VerificationError::Permanent(err)) => {
+                    error_count += 1;
+                    verify_result = Some(VerifyState::Failed);
+
+                    info!(
+                        "verify {datastore_name}:{backup_dir_name}/{file_name} failed with permanent error: {err}",
+                        file_name = info.filename,
+                    );
+                }
+                Err(VerificationError::Transient(err)) => {
+                    error_count += 1;
+                    match manifest.verify_state() {
+                        Ok(Some(SnapshotVerifyState {
+                            state: VerifyState::Failed,
+                            ..
+                        })) => {
+                            // keep failed state, to never allow chunk reuse for subsequent backup
+                            verify_result = Some(VerifyState::Failed);
+                        }
+                        Err(err) => {
+                            info!(
+                                "failed to parse verify state for {datastore_name}:{backup_dir_name}: {err}"
+                            );
+                            // set to failed since verify state parsing failed
+                            verify_result = Some(VerifyState::Failed);
+                        }
+                        _ => {
+                            // strip any VerifyState::Ok when updating the manifest
+                            if let Some(VerifyState::Ok) = verify_result {
+                                verify_result = None;
+                            }
+                        }
+                    };
+
+                    info!(
+                        "verify {datastore_name}:{backup_dir_name}/{file_name} failed with transient error: {err}",
+                        file_name = info.filename,
+                    );
+                }
             }
         }
 
-        let verify_state = SnapshotVerifyState {
-            state: verify_result,
-            upid,
-        };
-
-        if let Err(err) = {
-            let verify_state = serde_json::to_value(verify_state)?;
-            backup_dir.update_manifest(&self.datastore.backend()?, |manifest| {
+        let verify_state = verify_result.map(|state| SnapshotVerifyState { state, upid });
+        if let Err(err) = backup_dir.update_manifest(&self.datastore.backend()?, |manifest| {
+            if let Some(verify_state) = verify_state {
+                let verify_state = serde_json::to_value(verify_state)?;
                 manifest.unprotected["verify_state"] = verify_state;
-                Ok(())
-            })
-        } {
+            } else if let Some(unprotected) = manifest.unprotected.as_object_mut() {
+                unprotected.remove("verify_state");
+            }
+            Ok(())
+        }) {
             info!("verify {datastore_name}:{backup_dir_name} - manifest update error: {err}");
             return Ok(false);
         }
-- 
2.47.3