fix(scanner): fence backup publication scope

Co-Authored-By: heihutu <heihutu@gmail.com>

crates/scanner/src/lib.rs

@@ -797,29 +797,6 @@ where
     Some(admission)
 }
 
-pub(crate) async fn save_config_shared_with_preconditions_and_lease_fence<S>(
-    api: Arc<S>,
-    file: &str,
-    data: Bytes,
-    sha256hex: Option<String>,
-    preconditions: HTTPPreconditions,
-    scanner_publication_lease_fence: Option<&str>,
-) -> EcstoreResult<ScannerObjectInfo>
-where
-    S: ScannerObjectIO,
-{
-    save_config_shared_with_preconditions_and_lease_fence_and_scope(
-        api,
-        file,
-        data,
-        sha256hex,
-        preconditions,
-        scanner_publication_lease_fence,
-        None,
-    )
-    .await
-}
-
 pub(crate) async fn save_config_shared_with_preconditions_and_lease_fence_and_scope<S>(
     api: Arc<S>,
     file: &str,

crates/scanner/src/scanner.rs

@@ -63,6 +63,7 @@ use tokio::time::{Duration, Instant};
 use tokio_util::sync::CancellationToken;
 use tokio_util::task::AbortOnDropHandle;
 use tracing::{debug, error, info, instrument, warn};
+use uuid::Uuid;
 
 use crate::storage_api::scan::{
     BucketOperations, BucketOptions, NamespaceLocking as _, SCANNER_ACTIVITY_LEGACY_PROTOCOL_VERSION,
@@ -72,7 +73,7 @@ use crate::{
     ECStore, EcstoreError, RUSTFS_META_BUCKET, SCANNER_PUBLICATION_EPOCH_CHANGED, ScannerLifecycleConfigExt as _,
     ScannerReplicationConfigExt as _, delete_config_with_publication_admission_for_epoch, get_lifecycle_config,
     get_replication_config, invalidate_admin_data_usage_snapshot_cache, invalidate_data_usage_snapshot_cache, read_config,
-    replace_bucket_usage_memory_from_info, save_config, save_config_shared_with_preconditions_and_lease_fence,
+    replace_bucket_usage_memory_from_info, save_config, save_config_shared_with_preconditions_and_lease_fence_and_scope,
     save_config_with_preconditions, save_config_with_publication_admission_for_epoch, scanner_is_erasure_sd,
     scanner_publication_admission_for_epoch, scanner_publication_epoch, scanner_publication_epoch_changed,
 };
@@ -454,6 +455,7 @@ fn data_usage_backup_due(data_usage_info: &DataUsageInfo) -> bool {
 }
 
 #[cfg(test)]
+#[allow(dead_code)]
 async fn sync_data_usage_backup_from_primary(
     ctx: &CancellationToken,
     storeapi: Arc<impl ScannerObjectIO + ScannerConfigObjectDelete>,
@@ -461,12 +463,34 @@ async fn sync_data_usage_backup_from_primary(
     sync_data_usage_backup_from_primary_for_epoch_and_lease_and_fence(ctx, storeapi, None, None, None).await
 }
 
+#[allow(dead_code)]
 async fn sync_data_usage_backup_from_primary_for_epoch_and_lease_and_fence(
     ctx: &CancellationToken,
     storeapi: Arc<impl ScannerObjectIO + ScannerConfigObjectDelete>,
     expected_publication_epoch: Option<u64>,
     remote_lease_deadline: Option<std::time::Instant>,
     scanner_publication_lease_fence: Option<&str>,
+) -> Result<(), EcstoreError> {
+    sync_data_usage_backup_from_primary_for_epoch_and_lease_and_fence_and_scope(
+        ctx,
+        storeapi,
+        expected_publication_epoch,
+        remote_lease_deadline,
+        scanner_publication_lease_fence,
+        Vec::new(),
+        Arc::new(AtomicBool::new(true)),
+    )
+    .await
+}
+
+async fn sync_data_usage_backup_from_primary_for_epoch_and_lease_and_fence_and_scope(
+    ctx: &CancellationToken,
+    storeapi: Arc<impl ScannerObjectIO + ScannerConfigObjectDelete>,
+    expected_publication_epoch: Option<u64>,
+    remote_lease_deadline: Option<std::time::Instant>,
+    scanner_publication_lease_fence: Option<&str>,
+    remote_lease_tokens: Vec<Uuid>,
+    lease_release_safe: Arc<AtomicBool>,
 ) -> Result<(), EcstoreError> {
     let backup_path = format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str());
     for retry in 0..=SCANNER_PERSIST_CAS_RETRIES {
@@ -531,15 +555,48 @@ async fn sync_data_usage_backup_from_primary_for_epoch_and_lease_and_fence(
                 }
                 return Err(EcstoreError::other(SCANNER_PUBLICATION_EPOCH_CHANGED));
             };
-            save_config_shared_with_preconditions_and_lease_fence(
+            let publication_scope = match expected_publication_epoch {
+                Some(expected_epoch) => {
+                    storeapi
+                        .scanner_data_usage_publication_commit_scope_with_release_flag(
+                            expected_epoch,
+                            usage_store::scanner_publication_scope_deadline(data_usage_persist_timeout(), remote_lease_deadline),
+                            remote_lease_tokens.clone(),
+                            Arc::clone(&lease_release_safe),
+                        )
+                        .await
+                }
+                None => None,
+            };
+            if expected_publication_epoch.is_some() && publication_scope.is_none() {
+                if retry < SCANNER_PERSIST_CAS_RETRIES {
+                    continue;
+                }
+                return Err(EcstoreError::other(SCANNER_PUBLICATION_EPOCH_CHANGED));
+            }
+            let save_result = save_config_shared_with_preconditions_and_lease_fence_and_scope(
                 storeapi.clone(),
                 &backup_path,
                 primary.clone(),
                 sha256hex,
                 revision.preconditions(),
                 scanner_publication_lease_fence,
+                publication_scope.clone(),
             )
-            .await
+            .await;
+            if let Some(scope) = publication_scope {
+                match scope.wait_for_completion().await {
+                    crate::storage_api::owner::ScannerPublicationCommitState::Committed
+                    | crate::storage_api::owner::ScannerPublicationCommitState::AbortedBeforeCommit => save_result,
+                    crate::storage_api::owner::ScannerPublicationCommitState::Indeterminate
+                    | crate::storage_api::owner::ScannerPublicationCommitState::Admitted
+                    | crate::storage_api::owner::ScannerPublicationCommitState::InFlight => Err(EcstoreError::other(
+                        "scanner backup publication commit scope did not reach a safe terminal state",
+                    )),
+                }
+            } else {
+                save_result
+            }
         };
 
         match save_result {

crates/scanner/src/scanner/usage_store.rs

@@ -37,7 +37,7 @@ fn remote_lease_expired(deadline: Option<std::time::Instant>) -> bool {
     deadline.is_some_and(|deadline| std::time::Instant::now() >= deadline)
 }
 
-fn scanner_publication_scope_deadline(
+pub(super) fn scanner_publication_scope_deadline(
     persist_timeout: Duration,
     remote_lease_deadline: Option<std::time::Instant>,
 ) -> tokio::time::Instant {
@@ -836,12 +836,14 @@ where
 
         if backup_due {
             let done_save = Metrics::time(Metric::SaveUsage);
-            let backup_result = sync_data_usage_backup_from_primary_for_epoch_and_lease_and_fence(
+            let backup_result = sync_data_usage_backup_from_primary_for_epoch_and_lease_and_fence_and_scope(
                 &ctx,
                 storeapi.clone(),
                 expected_publication_epoch,
                 remote_lease_deadline,
                 scanner_publication_lease_fence.as_deref(),
+                remote_lease_tokens.clone(),
+                Arc::clone(&lease_release_safe),
             )
             .await;
             done_save();