diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index b4e349b39..5565a3f51 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -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) -> Result>> { 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 { + 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) } diff --git a/crates/scanner/src/lib.rs b/crates/scanner/src/lib.rs index 9091bd6d9..3a8cb8d6f 100644 --- a/crates/scanner/src/lib.rs +++ b/crates/scanner/src/lib.rs @@ -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( + api: Arc, + bucket: &str, + object: &str, + mut opts: ScannerObjectOptions, + expected_epoch: u64, + scanner_publication_commit_scope: Option, +) -> EcstoreResult +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( api: Arc, bucket: &str, @@ -763,10 +789,7 @@ pub(crate) async fn delete_config_with_publication_admission_for_epoch( 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 diff --git a/crates/scanner/src/scanner/usage_store.rs b/crates/scanner/src/scanner/usage_store.rs index 674f43b3f..9d181d52a 100644 --- a/crates/scanner/src/scanner/usage_store.rs +++ b/crates/scanner/src/scanner/usage_store.rs @@ -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, remote_lease_deadline: Option, scanner_publication_lease_fence: Option<&str>, + remote_lease_tokens: &[Uuid], + lease_release_safe: Arc, ) -> 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(