diff --git a/crates/ecstore/src/object_api/types.rs b/crates/ecstore/src/object_api/types.rs index 5f5691796..b55dcdaad 100644 --- a/crates/ecstore/src/object_api/types.rs +++ b/crates/ecstore/src/object_api/types.rs @@ -420,6 +420,7 @@ struct ScannerPublicationCommitScopeInner { /// future. A detached mutation task keeps the scope alive and therefore /// keeps this guard alive until it reports a terminal state. movement_permit: Mutex>>, + lease_release_safe: Arc, } /// Storage-owned ownership scope for one fenced scanner metadata mutation. @@ -453,6 +454,25 @@ impl ScannerPublicationCommitScope { remote_lease_tokens: Vec, movement_permit: OwnedRwLockReadGuard<()>, ) -> Self { + Self::new_storage_owned_with_release_flag( + expected_movement_epoch, + safe_deadline, + remote_lease_tokens, + movement_permit, + Arc::new(std::sync::atomic::AtomicBool::new(true)), + ) + } + + pub(crate) fn new_storage_owned_with_release_flag( + expected_movement_epoch: u64, + safe_deadline: tokio::time::Instant, + remote_lease_tokens: Vec, + movement_permit: OwnedRwLockReadGuard<()>, + lease_release_safe: Arc, + ) -> Self { + // Admission itself is not a safe release point. The flag becomes true + // only after the storage mutation owner reports a terminal state. + lease_release_safe.store(false, Ordering::Release); Self { inner: Arc::new(ScannerPublicationCommitScopeInner { expected_movement_epoch, @@ -462,6 +482,7 @@ impl ScannerPublicationCommitScope { state: AtomicU8::new(SCANNER_PUBLICATION_SCOPE_ADMITTED), completed: Notify::new(), movement_permit: Mutex::new(Some(movement_permit)), + lease_release_safe, }), } } @@ -537,21 +558,20 @@ impl ScannerPublicationCommitScope { } pub fn mark_aborted_before_commit(&self) -> bool { - for expected in [SCANNER_PUBLICATION_SCOPE_ADMITTED, SCANNER_PUBLICATION_SCOPE_IN_FLIGHT] { - if self - .inner - .state - .compare_exchange( - expected, - SCANNER_PUBLICATION_SCOPE_ABORTED_BEFORE_COMMIT, - Ordering::AcqRel, - Ordering::Acquire, - ) - .is_ok() - { - self.inner.completed.notify_waiters(); - return true; - } + if self + .inner + .state + .compare_exchange( + SCANNER_PUBLICATION_SCOPE_ADMITTED, + SCANNER_PUBLICATION_SCOPE_ABORTED_BEFORE_COMMIT, + Ordering::AcqRel, + Ordering::Acquire, + ) + .is_ok() + { + self.inner.lease_release_safe.store(true, Ordering::Release); + self.inner.completed.notify_waiters(); + return true; } false } @@ -565,7 +585,12 @@ impl ScannerPublicationCommitScope { .state .compare_exchange(SCANNER_PUBLICATION_SCOPE_IN_FLIGHT, terminal.as_u8(), Ordering::AcqRel, Ordering::Acquire) .is_ok() - .then(|| self.inner.completed.notify_waiters()) + .then(|| { + if terminal.permits_lease_release() { + self.inner.lease_release_safe.store(true, Ordering::Release); + } + self.inner.completed.notify_waiters() + }) .is_some() } @@ -595,6 +620,14 @@ impl ScannerPublicationCommitScope { } } +impl Drop for ScannerPublicationCommitScopeInner { + fn drop(&mut self) { + if !ScannerPublicationCommitState::from_u8(self.state.load(Ordering::Acquire)).permits_lease_release() { + self.lease_release_safe.store(false, Ordering::Release); + } + } +} + #[derive(Default, Clone)] pub struct ObjectOptions { // Use the maximum parity (N/2), used when saving server configuration files diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 7fd95925a..48da4cbab 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -105,7 +105,9 @@ use crate::{ SnapshotLeaseToken, UpdateMetadataOpts, endpoint::Endpoint, error::DiskError, format::FormatV3, }, error::{StorageError, to_object_err}, - object_api::{GetObjectReader, NamespaceLockFence, ObjectInfo, ObjectLockConfigSnapshot, PutObjReader}, + object_api::{ + GetObjectReader, NamespaceLockFence, ObjectInfo, ObjectLockConfigSnapshot, PutObjReader, ScannerPublicationCommitScope, + }, // event::name::EventName, services::event_notification::{EventArgs, send_event}, store::init_format::{ @@ -3937,6 +3939,47 @@ impl SetDisks { owner.scanner_data_usage_publication_admission_guard().await } + /// Acquire a storage-owned scanner publication scope for this set's + /// instance movement fence. The scope keeps the read permit alive across + /// scanner future cancellation until the mutation owner drains. + pub async fn scanner_data_usage_publication_commit_scope( + &self, + expected_movement_epoch: u64, + safe_deadline: tokio::time::Instant, + remote_lease_tokens: Vec, + ) -> Option { + let (movement_permit, epoch) = self.scanner_data_usage_publication_admission_guard().await?; + if epoch != expected_movement_epoch { + return None; + } + Some(ScannerPublicationCommitScope::new_storage_owned( + epoch, + safe_deadline, + remote_lease_tokens, + movement_permit, + )) + } + + pub async fn scanner_data_usage_publication_commit_scope_with_release_flag( + &self, + expected_movement_epoch: u64, + safe_deadline: tokio::time::Instant, + remote_lease_tokens: Vec, + lease_release_safe: Arc, + ) -> Option { + let (movement_permit, epoch) = self.scanner_data_usage_publication_admission_guard().await?; + if epoch != expected_movement_epoch { + return None; + } + Some(ScannerPublicationCommitScope::new_storage_owned_with_release_flag( + epoch, + safe_deadline, + remote_lease_tokens, + movement_permit, + lease_release_safe, + )) + } + /// Whether both sets' namespace-lock implementations cover the same object key. pub(crate) async fn shares_namespace_lock_domain(&self, other: &Self) -> bool { match (self.ctx.is_dist_erasure().await, other.ctx.is_dist_erasure().await) { diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 7e0d86b19..dae4a6dad 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -3394,7 +3394,12 @@ impl SetDisks { let commit_tmp_dir = tmp_dir.clone(); let commit_object_lock_guard = object_lock_guard.take(); let commit_bucket_lifecycle_guard = bucket_lifecycle_guard.take(); - let commit_allows_early_ack = commit_object_lock_guard.is_some(); + let commit_scanner_publication_scope = opts.scanner_publication_commit_scope.clone(); + // A scanner publication scope owns the movement permit until the + // complete rename fan-out drains. Keep this path synchronous so + // its terminal state is known before the coordinator releases + // remote leases. + let commit_allows_early_ack = commit_object_lock_guard.is_some() && commit_scanner_publication_scope.is_none(); let detach_commit_owner = commit_allows_early_ack || commit_bucket_lifecycle_guard.is_some() || quota_mutation_fence; let commit_write_path_label = write_path.metric_label(); let commit_is_versioned = opts.versioned || opts.version_suspended; @@ -3491,7 +3496,7 @@ impl SetDisks { } Ok(()) }; - let pre_rename_result = if cancellation.is_some() || request_cancellation.is_some() { + let mut pre_rename_result = if cancellation.is_some() || request_cancellation.is_some() { tokio::select! { biased; _ = wait_for_put_object_commit_cancellation(cancellation.as_ref(), request_cancellation.as_ref()) => { @@ -3502,7 +3507,21 @@ impl SetDisks { } else { pre_rename.await }; + if pre_rename_result.is_ok() + && let Some(scope) = commit_scanner_publication_scope.as_ref() + && let Err(err) = scope.try_begin() + { + let _ = scope.mark_aborted_before_commit(); + pre_rename_result = Err(Error::other(format!("scanner publication commit scope cannot start: {err:?}"))); + } if let Err(err) = pre_rename_result { + if let Some(scope) = commit_scanner_publication_scope.as_ref() { + if scope.state() == crate::object_api::ScannerPublicationCommitState::Admitted { + let _ = scope.mark_aborted_before_commit(); + } else { + let _ = scope.mark_indeterminate(); + } + } SetDisks::abort_quota_reservation_after_fence( quota_reservation, &commit_disks, @@ -3540,6 +3559,13 @@ impl SetDisks { ), ) .await; + if let Some(scope) = commit_scanner_publication_scope.as_ref() { + if rename_result.is_ok() { + let _ = scope.mark_committed(); + } else { + let _ = scope.mark_indeterminate(); + } + } #[cfg(any(test, feature = "test-util"))] if rename_result.is_ok() { pause_put_object_commit(&commit_bucket, &commit_object, PutObjectCommitPause::AfterRenameQuorum).await; diff --git a/crates/ecstore/src/store/mod.rs b/crates/ecstore/src/store/mod.rs index fc24ec023..06d827d4e 100644 --- a/crates/ecstore/src/store/mod.rs +++ b/crates/ecstore/src/store/mod.rs @@ -544,6 +544,30 @@ impl ECStore { )) } + /// Variant used by the scanner supervisor to observe whether a scope was + /// dropped before reaching a safe terminal state. The flag is in-memory + /// only and lets the supervisor avoid releasing remote leases on an + /// indeterminate cancellation path. + pub async fn scanner_data_usage_publication_commit_scope_with_release_flag( + &self, + expected_movement_epoch: u64, + safe_deadline: tokio::time::Instant, + remote_lease_tokens: Vec, + lease_release_safe: Arc, + ) -> Option { + let (movement_permit, epoch) = self.scanner_data_usage_publication_admission_guard().await?; + if epoch != expected_movement_epoch { + return None; + } + Some(ScannerPublicationCommitScope::new_storage_owned_with_release_flag( + epoch, + safe_deadline, + remote_lease_tokens, + movement_permit, + lease_release_safe, + )) + } + /// Capture the current publication epoch without holding the movement /// gate across backend I/O. Callers must re-admit the same epoch before a /// mutation commits. @@ -1521,6 +1545,10 @@ mod tests { .expect("a second idle publication scope should be granted"); scope.try_begin().expect("scope should enter the mutation state"); scope.cancel(); + assert!( + !scope.mark_aborted_before_commit(), + "an in-flight mutation cannot claim pre-commit abort without storage proof" + ); assert!(scope.mark_indeterminate()); assert_eq!( scope.wait_for_completion().await, diff --git a/crates/scanner/src/lib.rs b/crates/scanner/src/lib.rs index c84588b3d..1d8c901a4 100644 --- a/crates/scanner/src/lib.rs +++ b/crates/scanner/src/lib.rs @@ -39,11 +39,11 @@ use storage_api::owner::{ EcstoreListPathRawOptions, EcstoreNsScannerOpenRequest, EcstoreObjectOpts, EcstoreReplicationConfigurationExt, EcstoreReplicationScannerBridge, EcstoreResultType, EcstoreScanGuard, EcstoreSetDisks, EcstoreStorageError, EcstoreStore, EcstoreVersioningApi, HTTPPreconditions, HTTPRangeSpec, ObjectIO, ObjectOperations, ObjectToDelete, - ScannerReplicationHealObject, ScannerReplicationHealResult, ScannerReplicationQueueAdmission, ecstore_apply_expiry_rule, - ecstore_apply_transition_rule, ecstore_expiry_state_handle, ecstore_get_global_tier_config_mgr, ecstore_get_lifecycle_config, - ecstore_get_object_lock_config, ecstore_get_replication_config, ecstore_invalidate_admin_data_usage_snapshot_cache, - ecstore_invalidate_data_usage_snapshot_cache, ecstore_is_erasure, ecstore_is_erasure_sd, - ecstore_is_reserved_or_invalid_bucket, ecstore_list_path_raw, ecstore_object_opts_from_object_info, + ScannerPublicationCommitScope, ScannerReplicationHealObject, ScannerReplicationHealResult, ScannerReplicationQueueAdmission, + ecstore_apply_expiry_rule, ecstore_apply_transition_rule, ecstore_expiry_state_handle, ecstore_get_global_tier_config_mgr, + ecstore_get_lifecycle_config, ecstore_get_object_lock_config, ecstore_get_replication_config, + ecstore_invalidate_admin_data_usage_snapshot_cache, ecstore_invalidate_data_usage_snapshot_cache, ecstore_is_erasure, + ecstore_is_erasure_sd, ecstore_is_reserved_or_invalid_bucket, ecstore_list_path_raw, ecstore_object_opts_from_object_info, ecstore_path2_bucket_object, ecstore_path2_bucket_object_with_base_path, ecstore_read_config, ecstore_replace_bucket_usage_memory_from_info, ecstore_resolve_object_store_handle, ecstore_save_config, scanner_replication_config_for_lifecycle_eval, @@ -55,6 +55,7 @@ use storage_api::owner::{ ecstore_new_disk, }; use tokio_util::sync::CancellationToken; +use uuid::Uuid; pub mod data_usage_define; pub mod error; @@ -804,6 +805,30 @@ pub(crate) async fn save_config_shared_with_preconditions_and_lease_fence( 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, + data: Bytes, + sha256hex: Option, + preconditions: HTTPPreconditions, + scanner_publication_lease_fence: Option<&str>, + scanner_publication_commit_scope: Option, +) -> EcstoreResult where S: ScannerObjectIO, { @@ -822,6 +847,7 @@ where &ScannerObjectOptions { max_parity: true, http_preconditions: Some(preconditions), + scanner_publication_commit_scope, user_defined, ..Default::default() }, @@ -886,6 +912,27 @@ pub trait ScannerConfigObjectDelete: Send + Sync + std::fmt::Debug + 'static { async fn scanner_data_usage_publication_admission(&self) -> Option { None } + + /// Acquire a storage-owned scope for a fenced scanner metadata mutation. + /// Implementations without a storage movement owner fail closed. + async fn scanner_data_usage_publication_commit_scope( + &self, + _expected_movement_epoch: u64, + _safe_deadline: tokio::time::Instant, + _remote_lease_tokens: Vec, + ) -> Option { + None + } + + async fn scanner_data_usage_publication_commit_scope_with_release_flag( + &self, + _expected_movement_epoch: u64, + _safe_deadline: tokio::time::Instant, + _remote_lease_tokens: Vec, + _lease_release_safe: Arc, + ) -> Option { + None + } } pub struct ScannerDataUsagePublicationAdmission { @@ -929,6 +976,32 @@ impl ScannerConfigObjectDelete for ECStore { let (read_guard, epoch) = self.scanner_data_usage_publication_admission_guard().await?; Some(ScannerDataUsagePublicationAdmission::fenced(read_guard, epoch)) } + + async fn scanner_data_usage_publication_commit_scope( + &self, + expected_movement_epoch: u64, + safe_deadline: tokio::time::Instant, + remote_lease_tokens: Vec, + ) -> Option { + self.scanner_data_usage_publication_commit_scope(expected_movement_epoch, safe_deadline, remote_lease_tokens) + .await + } + + async fn scanner_data_usage_publication_commit_scope_with_release_flag( + &self, + expected_movement_epoch: u64, + safe_deadline: tokio::time::Instant, + remote_lease_tokens: Vec, + lease_release_safe: Arc, + ) -> Option { + self.scanner_data_usage_publication_commit_scope_with_release_flag( + expected_movement_epoch, + safe_deadline, + remote_lease_tokens, + lease_release_safe, + ) + .await + } } #[async_trait::async_trait] @@ -946,6 +1019,32 @@ impl ScannerConfigObjectDelete for SetDisks { let (read_guard, epoch) = self.scanner_data_usage_publication_admission_guard().await?; Some(ScannerDataUsagePublicationAdmission::fenced(read_guard, epoch)) } + + async fn scanner_data_usage_publication_commit_scope( + &self, + expected_movement_epoch: u64, + safe_deadline: tokio::time::Instant, + remote_lease_tokens: Vec, + ) -> Option { + self.scanner_data_usage_publication_commit_scope(expected_movement_epoch, safe_deadline, remote_lease_tokens) + .await + } + + async fn scanner_data_usage_publication_commit_scope_with_release_flag( + &self, + expected_movement_epoch: u64, + safe_deadline: tokio::time::Instant, + remote_lease_tokens: Vec, + lease_release_safe: Arc, + ) -> Option { + self.scanner_data_usage_publication_commit_scope_with_release_flag( + expected_movement_epoch, + safe_deadline, + remote_lease_tokens, + lease_release_safe, + ) + .await + } } #[cfg(test)] diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 788426b64..ca744b53f 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -16,6 +16,7 @@ use std::collections::BTreeMap; use std::future::Future; #[cfg(test)] use std::sync::Mutex as StdMutex; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, LazyLock, RwLock}; use self::heal_info::{BackgroundHealInfoReadStatus, read_background_heal_info_with_epoch, save_background_heal_info_for_epoch}; @@ -63,7 +64,6 @@ use tokio_util::sync::CancellationToken; use tokio_util::task::AbortOnDropHandle; use tracing::{debug, error, info, instrument, warn}; -use crate::storage_api::owner::SCANNER_PUBLICATION_LEASE_TTL_MS; use crate::storage_api::scan::{ BucketOperations, BucketOptions, NamespaceLocking as _, SCANNER_ACTIVITY_LEGACY_PROTOCOL_VERSION, SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION, SCANNER_ACTIVITY_PROTOCOL_VERSION, @@ -1205,10 +1205,6 @@ fn data_usage_persist_timeout() -> Duration { DataUsageCache::persistence_timeout() } -fn scanner_publication_lease_budget_allows_persistence(timeout: Duration) -> bool { - timeout < Duration::from_millis(SCANNER_PUBLICATION_LEASE_TTL_MS) -} - #[cfg(not(test))] const SCANNER_CYCLE_EPOCH_FENCE_TIMEOUT: Duration = Duration::from_secs(30); #[cfg(test)] @@ -1493,11 +1489,6 @@ async fn run_data_scanner_cycle_with_budget( let mut remote_publication_leases = None; let remote_lease_defer_reason = if remote_publication_lease_targets.is_empty() { None - } else if !scanner_publication_lease_budget_allows_persistence(usage_persist_timeout) { - // The lease is intentionally fixed-duration and has no renewal path. - // Refuse a persistence budget that could outlive it instead of - // allowing the peer to admit movement while a local PUT is in flight. - Some(ScannerCycleDeferReason::PublicationLeaseBudgetExceeded) } else if let Some(notification_system) = storeapi.notification_system() { match notification_system .acquire_scanner_publication_leases(remote_publication_lease_targets.clone()) @@ -1548,26 +1539,16 @@ async fn run_data_scanner_cycle_with_budget( remote_lease_fence.is_some(), )) .then_some(ScannerCycleDeferReason::ActivityBaselineUnavailable); - let remote_lease_covers_persistence = remote_lease_deadline.is_none_or(|deadline| { - std::time::Instant::now() - .checked_add(usage_persist_timeout) - .is_some_and(|latest_finish| latest_finish < deadline) - }); let publication_defer_reason = publication_defer_reason .or(remote_lease_defer_reason) .or(remote_lease_fence_defer_reason); - let publication_defer_reason = (!remote_lease_covers_persistence) - .then_some(ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded) - .or(publication_defer_reason); // Include reasons discovered while acquiring or validating remote leases. - // In particular, the static budget gate above is reached after the scan - // result is classified, so computing this flag earlier would suppress its - // deferred metric. let publication_deferred = publication_defer_reason.is_some(); let budget_elapsed = cycle_budget.budget_elapsed() && !ctx.is_cancelled(); let remote_lease_probe = remote_publication_leases .as_ref() .map(|(notification_system, grants)| (Arc::clone(notification_system), grants.clone())); + let remote_lease_release_safe = Arc::new(AtomicBool::new(true)); let mut usage_persist_outcome = match publication_defer_reason { Some(reason) => { drop(receiver); @@ -1581,6 +1562,11 @@ async fn run_data_scanner_cycle_with_budget( let ctx_clone = ctx.clone(); let route_probe_store = storeapi.clone(); let remote_lease_fence = remote_lease_fence.clone(); + let remote_lease_release_safe_for_task = Arc::clone(&remote_lease_release_safe); + let remote_lease_tokens = remote_publication_leases + .as_ref() + .map(|(_, grants)| grants.iter().map(|grant| grant.lease.token).collect()) + .unwrap_or_default(); let mut usage_persist_task = AbortOnDropHandle::new(tokio::spawn(async move { store_data_usage_in_backend_with_outcome_for_epoch_and_baseline_and_route_probe_for_publication_epoch_and_lease_fence( ctx_clone, @@ -1592,7 +1578,9 @@ async fn run_data_scanner_cycle_with_budget( publication_epoch, remote_lease_deadline, remote_lease_fence, - ), + ) + .with_remote_lease_tokens(remote_lease_tokens) + .with_lease_release_flag(remote_lease_release_safe_for_task), move || { let storeapi = route_probe_store.clone(); let remote_lease_probe = remote_lease_probe.clone(); @@ -1657,7 +1645,16 @@ async fn run_data_scanner_cycle_with_budget( let lease_expired = remote_publication_leases .as_ref() .is_some_and(|(_, grants)| grants.iter().any(|grant| !grant.lease.is_valid())); - if let Some((notification_system, grants)) = remote_publication_leases.take() { + if !remote_lease_release_safe.load(Ordering::Acquire) { + // A cancelled or detached storage mutation did not report a safe + // terminal state. Keep remote grants until their own expiry rather + // than releasing movement admission while a commit may be unknown. + usage_persist_outcome = if usage_persist_outcome == DataUsagePersistOutcome::Failed { + DataUsagePersistOutcome::Failed + } else { + DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded) + }; + } else if let Some((notification_system, grants)) = remote_publication_leases.take() { let release_result = notification_system.release_scanner_publication_leases(grants).await; let lease_release_failed = release_result.is_err(); if lease_expired || lease_release_failed { diff --git a/crates/scanner/src/scanner/tests.rs b/crates/scanner/src/scanner/tests.rs index 49af15397..a41fc9692 100644 --- a/crates/scanner/src/scanner/tests.rs +++ b/crates/scanner/src/scanner/tests.rs @@ -4628,7 +4628,6 @@ fn scanner_cycle_cache_floor_stays_pending_during_deferred_usage_publication() { for reason in [ ScannerCycleDeferReason::DataMovement, ScannerCycleDeferReason::ActivityBaselineUnavailable, - ScannerCycleDeferReason::PublicationLeaseBudgetExceeded, ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded, ScannerCycleDeferReason::PublicationLeaseReleaseFailed, ] { @@ -4801,18 +4800,7 @@ fn data_usage_persist_wait_covers_cache_retries_and_backup() { } #[test] -fn scanner_publication_lease_budget_has_a_strict_ttl_boundary() { - let ttl = Duration::from_millis(SCANNER_PUBLICATION_LEASE_TTL_MS); - - assert!(scanner_publication_lease_budget_allows_persistence( - ttl.saturating_sub(Duration::from_millis(1)) - )); - assert!(!scanner_publication_lease_budget_allows_persistence(ttl)); - assert!(!scanner_publication_lease_budget_allows_persistence(ttl + Duration::from_millis(1))); - assert_eq!( - ScannerCycleDeferReason::PublicationLeaseBudgetExceeded.as_str(), - "publication_lease_budget_exceeded" - ); +fn scanner_publication_lease_deadline_reason_remains_distinct() { assert_eq!( ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded.as_str(), "publication_lease_deadline_exceeded" diff --git a/crates/scanner/src/scanner/usage_store.rs b/crates/scanner/src/scanner/usage_store.rs index 23bfe60d7..f7dc7cacd 100644 --- a/crates/scanner/src/scanner/usage_store.rs +++ b/crates/scanner/src/scanner/usage_store.rs @@ -13,7 +13,10 @@ // limitations under the License. /// Data-usage snapshot persistence: CAS store pipeline, epoch baselines, and observed-snapshot cleanup. use super::*; +use crate::storage_api::owner::ScannerPublicationCommitState; use std::collections::HashMap; +use std::sync::atomic::AtomicBool; +use uuid::Uuid; #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] pub(super) enum DataUsagePersistOutcome { @@ -34,6 +37,16 @@ fn remote_lease_expired(deadline: Option) -> bool { deadline.is_some_and(|deadline| std::time::Instant::now() >= deadline) } +fn scanner_publication_scope_deadline( + persist_timeout: Duration, + remote_lease_deadline: Option, +) -> tokio::time::Instant { + let configured_deadline = tokio::time::Instant::now() + persist_timeout; + remote_lease_deadline + .map(tokio::time::Instant::from_std) + .map_or(configured_deadline, |lease_deadline| configured_deadline.min(lease_deadline)) +} + #[derive(Clone, Debug)] pub(super) struct DataUsagePersistBaseline { pub(super) data: Option, @@ -126,6 +139,8 @@ pub(super) struct ScannerPublicationFence { pub(super) expected_publication_epoch: Option, pub(super) remote_lease_deadline: Option, pub(super) scanner_publication_lease_fence: Option, + pub(super) remote_lease_tokens: Vec, + pub(super) lease_release_safe: Arc, } impl ScannerPublicationFence { @@ -138,8 +153,20 @@ impl ScannerPublicationFence { expected_publication_epoch, remote_lease_deadline, scanner_publication_lease_fence, + remote_lease_tokens: Vec::new(), + lease_release_safe: Arc::new(AtomicBool::new(true)), } } + + pub(super) fn with_remote_lease_tokens(mut self, remote_lease_tokens: Vec) -> Self { + self.remote_lease_tokens = remote_lease_tokens; + self + } + + pub(super) fn with_lease_release_flag(mut self, lease_release_safe: Arc) -> Self { + self.lease_release_safe = lease_release_safe; + self + } } #[derive(Debug)] @@ -290,6 +317,8 @@ where expected_publication_epoch, remote_lease_deadline, scanner_publication_lease_fence, + remote_lease_tokens, + lease_release_safe, } = publication_fence; let mut outcome = DataUsagePersistOutcome::NoUpdate; let mut next_baseline = initial_baseline; @@ -580,25 +609,54 @@ where let done_save = Metrics::time(Metric::SaveUsage); let save_result = { - let Some(_publication_admission) = - scanner_publication_admission_for_epoch(storeapi.clone(), publication_epoch_for_save).await - else { - done_save(); - break DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement); + let publication_scope = storeapi + .scanner_data_usage_publication_commit_scope_with_release_flag( + publication_epoch_for_save, + scanner_publication_scope_deadline(data_usage_persist_timeout(), remote_lease_deadline), + remote_lease_tokens.clone(), + Arc::clone(&lease_release_safe), + ) + .await; + let legacy_publication_admission = if publication_scope.is_none() { + let Some(admission) = + scanner_publication_admission_for_epoch(storeapi.clone(), publication_epoch_for_save).await + else { + done_save(); + break DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement); + }; + Some(admission) + } else { + None }; if remote_lease_expired(remote_lease_deadline) { done_save(); break DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded); } - save_config_shared_with_preconditions_and_lease_fence( + let save_result = crate::save_config_shared_with_preconditions_and_lease_fence_and_scope( storeapi.clone(), target_path, data.clone(), sha256hex.clone(), revision.preconditions(), scanner_publication_lease_fence.as_deref(), + publication_scope.clone(), ) - .await + .await; + drop(legacy_publication_admission); + if let Some(scope) = publication_scope { + match scope.wait_for_completion().await { + ScannerPublicationCommitState::Committed | ScannerPublicationCommitState::AbortedBeforeCommit => { + save_result + } + ScannerPublicationCommitState::Indeterminate + | ScannerPublicationCommitState::Admitted + | ScannerPublicationCommitState::InFlight => Err(EcstoreError::other( + "scanner publication commit scope did not reach a safe terminal state", + )), + } + } else { + save_result + } }; done_save(); diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index e1abbf474..6b8eaeb6b 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -574,10 +574,6 @@ pub(crate) async fn scanner_set_disk_inventory(set: &SetDisks) -> Vec> pub(crate) enum ScannerCycleDeferReason { ActivityBaselineUnavailable, DataMovement, - /// The configured persistence budget cannot fit within the fixed remote - /// publication-lease TTL. This is a deterministic configuration/contract - /// mismatch, not evidence that a peer activity probe failed. - PublicationLeaseBudgetExceeded, /// A granted lease's absolute deadline cannot cover the persistence /// operation. This can occur even when the configured budget fits the /// nominal TTL because lease acquisition consumed part of the window. @@ -592,7 +588,6 @@ impl ScannerCycleDeferReason { match self { Self::ActivityBaselineUnavailable => "activity_baseline_unavailable", Self::DataMovement => "data_movement", - Self::PublicationLeaseBudgetExceeded => "publication_lease_budget_exceeded", Self::PublicationLeaseDeadlineExceeded => "publication_lease_deadline_exceeded", Self::PublicationLeaseReleaseFailed => "publication_lease_release_failed", } diff --git a/crates/scanner/src/storage_api.rs b/crates/scanner/src/storage_api.rs index 4278fdad8..670dbdb25 100644 --- a/crates/scanner/src/storage_api.rs +++ b/crates/scanner/src/storage_api.rs @@ -92,7 +92,9 @@ pub(crate) use rustfs_ecstore::api::event::{EventArgs as EcstoreEventArgs, send_ pub(crate) use rustfs_ecstore::api::layout::{ EndpointServerPools as EcstoreEndpointServerPools, Endpoints as EcstoreEndpoints, PoolEndpoints as EcstorePoolEndpoints, }; -pub(crate) use rustfs_ecstore::api::object::SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY; +pub(crate) use rustfs_ecstore::api::object::{ + SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY, ScannerPublicationCommitScope, ScannerPublicationCommitState, +}; #[cfg(test)] pub(crate) use rustfs_ecstore::api::rebalance::{ RebalStatus as EcstoreRebalStatus, RebalanceInfo as EcstoreRebalanceInfo, RebalanceMeta as EcstoreRebalanceMeta, @@ -106,9 +108,9 @@ pub(crate) use rustfs_ecstore::api::runtime::{ setup_is_erasure_sd as ecstore_is_erasure_sd, }; pub(crate) use rustfs_ecstore::api::set_disk::SetDisks as EcstoreSetDisks; +pub(crate) use rustfs_ecstore::api::storage::ECStore as EcstoreStore; #[cfg(test)] pub(crate) use rustfs_ecstore::api::storage::init_local_disks_with_instance_ctx as ecstore_init_local_disks_with_instance_ctx; -pub(crate) use rustfs_ecstore::api::storage::{ECStore as EcstoreStore, SCANNER_PUBLICATION_LEASE_TTL_MS}; use rustfs_storage_api as storage_contracts; pub(crate) mod owner { @@ -125,7 +127,7 @@ pub(crate) mod owner { EcstoreNsScannerOpenRequest, EcstoreObjectLockConfiguration, EcstoreObjectOpts, EcstoreReplicationConfigurationExt, EcstoreReplicationScannerBridge, EcstoreResultType, EcstoreScanGuard, EcstoreSetDisks, EcstoreStorageError, EcstoreStore, EcstoreVersioningApi, EcstoreVersioningConfiguration, SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY, - SCANNER_PUBLICATION_LEASE_TTL_MS, ScannerReplicationHealObject, ScannerReplicationHealResult, + ScannerPublicationCommitScope, ScannerPublicationCommitState, ScannerReplicationHealObject, ScannerReplicationHealResult, ScannerReplicationQueueAdmission, ecstore_apply_expiry_rule, ecstore_apply_transition_rule, ecstore_expiry_state_handle, ecstore_get_global_tier_config_mgr, ecstore_get_lifecycle_config, ecstore_get_object_lock_config, ecstore_get_replication_config, ecstore_invalidate_admin_data_usage_snapshot_cache,