From 6b9976e91cfcd744ed0d7539e03c1c3e72729e4c Mon Sep 17 00:00:00 2001 From: houseme Date: Sun, 30 Aug 2026 02:51:47 +0800 Subject: [PATCH] fix(scanner): fence backup publication scope Co-Authored-By: heihutu --- crates/scanner/src/lib.rs | 23 --------- crates/scanner/src/scanner.rs | 63 +++++++++++++++++++++-- crates/scanner/src/scanner/usage_store.rs | 6 ++- 3 files changed, 64 insertions(+), 28 deletions(-) diff --git a/crates/scanner/src/lib.rs b/crates/scanner/src/lib.rs index 1d8c901a4..9091bd6d9 100644 --- a/crates/scanner/src/lib.rs +++ b/crates/scanner/src/lib.rs @@ -797,29 +797,6 @@ where Some(admission) } -pub(crate) async fn save_config_shared_with_preconditions_and_lease_fence( - api: Arc, - file: &str, - data: Bytes, - sha256hex: Option, - preconditions: HTTPPreconditions, - scanner_publication_lease_fence: Option<&str>, -) -> EcstoreResult -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( api: Arc, file: &str, diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index ca744b53f..f213165e3 100644 --- a/crates/scanner/src/scanner.rs +++ b/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, @@ -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, expected_publication_epoch: Option, remote_lease_deadline: Option, 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, + expected_publication_epoch: Option, + remote_lease_deadline: Option, + scanner_publication_lease_fence: Option<&str>, + remote_lease_tokens: Vec, + lease_release_safe: Arc, ) -> 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 { diff --git a/crates/scanner/src/scanner/usage_store.rs b/crates/scanner/src/scanner/usage_store.rs index f7dc7cacd..674f43b3f 100644 --- a/crates/scanner/src/scanner/usage_store.rs +++ b/crates/scanner/src/scanner/usage_store.rs @@ -37,7 +37,7 @@ fn remote_lease_expired(deadline: Option) -> 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, ) -> 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();