gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-05 19:55:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(scanner): fence observed cleanup mutations

Browse Source 
Co-Authored-By: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
2026-08-30 03:04:42 +08:00
parent 290deb8665
commit 52961d2726
3 changed files with 101 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/set_disk/ops/object.rs
+43
View File
@@ -270,6 +270,22 @@ const OLD_DATA_CLEANUP_RECEIPT_FILE: &str = ".rustfs-old-data-cleanup-receipt.js
const SCANNER_PUBLICATION_LEASE_FENCE_MAX_BYTES: usize = 64 * 1024;
const SCANNER_PUBLICATION_LEASE_FENCE_MAX_ENTRIES: usize = 256;
fn begin_scanner_publication_delete_mutation(scope: Option<&crate::object_api::ScannerPublicationCommitScope>) -> Result<()> {
let Some(scope) = scope else {
return Ok(());
};
if scope.state() == crate::object_api::ScannerPublicationCommitState::Admitted {
scope
.try_begin()
.map_err(|err| Error::other(format!("scanner publication delete scope cannot start: {err:?}")))?;
}
if !scope.can_commit() {
let _ = scope.mark_indeterminate();
return Err(StorageError::OperationCanceled);
}
Ok(())
}
fn take_scanner_publication_lease_tokens(user_defined: &mut HashMap<String, String>) -> Result<Option<HashMap<String, Uuid>>> {
let Some(encoded) = user_defined.remove(SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY) else {
return Ok(None);
@@ -7098,6 +7114,11 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
#[tracing::instrument(skip(self, opts))]
async fn delete_object(&self, bucket: &str, object: &str, mut opts: ObjectOptions) -> Result<ObjectInfo> {
let _scope_outcome_guard = opts
.scanner_publication_commit_scope
.clone()
.map(ScannerPublicationCommitScopeGuard::new);
let scanner_publication_commit_scope = opts.scanner_publication_commit_scope.clone();
// Scanner cleanup carries the per-peer lease fence as transient
// request metadata. Consume it before any delete-prefix fanout so it
// cannot be persisted or treated as user metadata.
@@ -7192,6 +7213,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
}
delete_request.set_skip_tier_free_version();
}
begin_scanner_publication_delete_mutation(scanner_publication_commit_scope.as_ref())?;
self.delete_object_version(bucket, object, &delete_request, false).await?;
if let Some((_, deleted_object)) = replication_delete {
ReplicationLifecycleBridge::schedule_delete(bucket.to_string(), deleted_object).await;
@@ -7206,6 +7228,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
..Default::default()
};
delete_request.set_tier_free_version_id(&Uuid::new_v4().to_string());
begin_scanner_publication_delete_mutation(scanner_publication_commit_scope.as_ref())?;
self.delete_object_version(bucket, object, &delete_request, false).await?;
}
for version in &versions.free_versions {
@@ -7217,10 +7240,14 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
..Default::default()
};
delete_request.set_tier_free_version();
begin_scanner_publication_delete_mutation(scanner_publication_commit_scope.as_ref())?;
self.delete_object_version(bucket, object, &delete_request, false).await?;
}
}
}
if let Some(scope) = scanner_publication_commit_scope.as_ref() {
let _ = scope.mark_committed();
}
self.invalidate_get_object_metadata_cache(bucket, object).await;
return Ok(ObjectInfo::default());
}
@@ -7228,10 +7255,14 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
self.validate_bucket_incarnation(bucket, expected_incarnation_id).await?;
}
ensure_delete_commit_locks_held(_lock_guard.as_ref(), bucket, object, &opts)?;
begin_scanner_publication_delete_mutation(scanner_publication_commit_scope.as_ref())?;
self.delete_prefix_with_scanner_publication_lease(bucket, object, scanner_publication_lease_tokens.as_ref())
.await
.map_err(|e| to_object_err(e.into(), vec![bucket, object]))?;
if let Some(scope) = scanner_publication_commit_scope.as_ref() {
let _ = scope.mark_committed();
}
self.invalidate_all_get_object_metadata_cache();
return Ok(ObjectInfo::default());
}
@@ -7304,10 +7335,14 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
..Default::default()
};
ensure_delete_commit_locks_held(_lock_guard.as_ref(), bucket, object, &opts)?;
begin_scanner_publication_delete_mutation(scanner_publication_commit_scope.as_ref())?;
self.delete_object_version(bucket, object, &dfi, false)
.await
.map_err(|e| to_object_err(e, vec![bucket, object]))?;
self.invalidate_get_object_metadata_cache(bucket, object).await;
if let Some(scope) = scanner_publication_commit_scope.as_ref() {
let _ = scope.mark_committed();
}
return Ok(ObjectInfo::from_file_info(&dfi, bucket, object, opts.versioned || opts.version_suspended));
}
@@ -7381,6 +7416,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
};
ensure_delete_commit_locks_held(_lock_guard.as_ref(), bucket, object, &opts)?;
begin_scanner_publication_delete_mutation(scanner_publication_commit_scope.as_ref())?;
self.delete_object_version(bucket, object, &fi, should_force_delete_marker_for_missing_version(&opts))
.await
.map_err(|e| to_object_err(e, vec![bucket, object]))?;
@@ -7392,6 +7428,9 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
oi.user_tags = Arc::clone(&goi.user_tags);
oi.replication_decision = goi.replication_decision;
self.invalidate_get_object_metadata_cache(bucket, object).await;
if let Some(scope) = scanner_publication_commit_scope.as_ref() {
let _ = scope.mark_committed();
}
return Ok(oi);
}
@@ -7417,6 +7456,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
}
ensure_delete_commit_locks_held(_lock_guard.as_ref(), bucket, object, &opts)?;
begin_scanner_publication_delete_mutation(scanner_publication_commit_scope.as_ref())?;
self.delete_object_version(bucket, object, &dfi, opts.delete_marker)
.await
.map_err(|e| to_object_err(e, vec![bucket, object]))?;
@@ -7442,6 +7482,9 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
obj_info.delete_marker = true;
}
self.invalidate_get_object_metadata_cache(bucket, object).await;
if let Some(scope) = scanner_publication_commit_scope.as_ref() {
let _ = scope.mark_committed();
}
Ok(obj_info)
}
crates/scanner/src/lib.rs
+27 -4
View File
@@ -753,6 +753,32 @@ pub(crate) fn scanner_publication_epoch_changed(error: &EcstoreError) -> bool {
)
}
pub(crate) async fn delete_config_with_publication_scope_for_epoch<S>(
api: Arc<S>,
bucket: &str,
object: &str,
mut opts: ScannerObjectOptions,
expected_epoch: u64,
scanner_publication_commit_scope: Option<ScannerPublicationCommitScope>,
) -> EcstoreResult<ScannerObjectInfo>
where
S: ScannerObjectIO + ScannerConfigObjectDelete,
{
let legacy_admission = if scanner_publication_commit_scope.is_none() {
Some(
scanner_publication_admission_for_epoch(api.clone(), expected_epoch)
.await
.ok_or_else(|| EcstoreError::other(SCANNER_PUBLICATION_EPOCH_CHANGED))?,
)
} else {
None
};
opts.scanner_publication_commit_scope = scanner_publication_commit_scope;
let result = api.delete_config_object(bucket, object, opts).await;
drop(legacy_admission);
result
}
pub(crate) async fn delete_config_with_publication_admission_for_epoch<S>(
api: Arc<S>,
bucket: &str,
@@ -763,10 +789,7 @@ pub(crate) async fn delete_config_with_publication_admission_for_epoch<S>(
where
S: ScannerObjectIO + ScannerConfigObjectDelete,
{
let Some(_admission) = scanner_publication_admission_for_epoch(api.clone(), expected_epoch).await else {
return Err(EcstoreError::other(SCANNER_PUBLICATION_EPOCH_CHANGED));
};
api.delete_config_object(bucket, object, opts).await
delete_config_with_publication_scope_for_epoch(api, bucket, object, opts, expected_epoch, None).await
}
/// Capture the storage-owned publication epoch without retaining the read
crates/scanner/src/scanner/usage_store.rs
+31 -1
View File
@@ -754,6 +754,8 @@ where
expected_publication_epoch,
remote_lease_deadline,
scanner_publication_lease_fence.as_deref(),
&remote_lease_tokens,
Arc::clone(&lease_release_safe),
)
.await;
if expected_publication_epoch.is_some() && !cleanup_ok {
@@ -777,6 +779,8 @@ where
expected_publication_epoch,
remote_lease_deadline,
scanner_publication_lease_fence.as_deref(),
&remote_lease_tokens,
Arc::clone(&lease_release_safe),
)
.await;
if expected_publication_epoch.is_some() && !cleanup_ok {
@@ -819,6 +823,8 @@ where
expected_publication_epoch,
remote_lease_deadline,
scanner_publication_lease_fence.as_deref(),
&remote_lease_tokens,
Arc::clone(&lease_release_safe),
)
.await;
if expected_publication_epoch.is_some() && !cleanup_ok {
@@ -877,6 +883,8 @@ async fn cleanup_observed_data_usage_snapshot_for_epoch_and_lease(
expected_publication_epoch: Option<u64>,
remote_lease_deadline: Option<std::time::Instant>,
scanner_publication_lease_fence: Option<&str>,
remote_lease_tokens: &[Uuid],
lease_release_safe: Arc<AtomicBool>,
) -> bool {
if remote_lease_expired(remote_lease_deadline) {
return false;
@@ -945,7 +953,15 @@ async fn cleanup_observed_data_usage_snapshot_for_epoch_and_lease(
return false;
}
let result = delete_config_with_publication_admission_for_epoch(
let publication_scope = storeapi
.scanner_data_usage_publication_commit_scope_with_release_flag(
read_epoch,
scanner_publication_scope_deadline(data_usage_persist_timeout(), remote_lease_deadline),
remote_lease_tokens.to_vec(),
Arc::clone(&lease_release_safe),
)
.await;
let result = crate::delete_config_with_publication_scope_for_epoch(
storeapi,
RUSTFS_META_BUCKET,
DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str(),
@@ -964,9 +980,23 @@ async fn cleanup_observed_data_usage_snapshot_for_epoch_and_lease(
..Default::default()
},
read_epoch,
publication_scope.clone(),
)
.await;
let result = if let Some(scope) = publication_scope {
match scope.wait_for_completion().await {
ScannerPublicationCommitState::Committed | ScannerPublicationCommitState::AbortedBeforeCommit => result,
ScannerPublicationCommitState::Indeterminate
| ScannerPublicationCommitState::Admitted
| ScannerPublicationCommitState::InFlight => Err(EcstoreError::other(
"scanner publication cleanup scope did not reach a safe terminal state",
)),
}
} else {
result
};
match result {
Ok(_)
| Err(