From ee39e4fccb19e9bfa735ef31cca8ac06ead8e936 Mon Sep 17 00:00:00 2001 From: houseme Date: Sun, 30 Aug 2026 10:39:07 +0800 Subject: [PATCH] fix(scanner): own publication mutations through storage drain (#6867) * fix(scanner): own publication mutations through storage drain Co-Authored-By: heihutu * fix(storage): remove unused rename data shim Co-Authored-By: heihutu --------- Co-authored-by: heihutu --- crates/ecstore/src/api/mod.rs | 7 +- crates/ecstore/src/disk/disk_store.rs | 146 ++++++++ crates/ecstore/src/disk/mod.rs | 34 +- crates/ecstore/src/object_api/types.rs | 340 ++++++++++++++++++ .../src/set_disk/core/io_primitives.rs | 134 ++++++- crates/ecstore/src/set_disk/mod.rs | 38 ++ crates/ecstore/src/set_disk/ops/object.rs | 102 +++++- crates/ecstore/src/store/mod.rs | 160 ++++++++- crates/scanner/src/lib.rs | 119 +++++- crates/scanner/src/scanner.rs | 96 ++++- crates/scanner/src/scanner/tests.rs | 68 ++++ crates/scanner/src/scanner/usage_store.rs | 108 +++++- crates/scanner/src/storage_api.rs | 21 +- rustfs/src/storage/rpc/node_service/disk.rs | 82 +++-- rustfs/src/storage/storage_api.rs | 19 - 15 files changed, 1360 insertions(+), 114 deletions(-) diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 24b587bd2..fec7e009e 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -415,9 +415,10 @@ pub mod object { GetObjectBodyCacheHookLookup, GetObjectBodySource, GetObjectReader, NamespaceLockFence, ObjectEncryptionResolver, ObjectInfo, ObjectLockConfigSnapshot, ObjectMutationHook, ObjectOptions, PutObjReader, QuotaAdmission, RangedDecompressReader, ReadEncryptionMaterial, ReadEncryptionMode, ReadEncryptionRequest, - SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY, StreamConsumer, get_object_body_cache_plaintext_len, - lookup_get_object_body_cache_hook, register_get_object_body_cache_hook, register_object_mutation_hook, - unregister_get_object_body_cache_hook, unregister_object_mutation_hook, + SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY, ScannerPublicationCommitScope, ScannerPublicationCommitStartError, + ScannerPublicationCommitState, StreamConsumer, get_object_body_cache_plaintext_len, lookup_get_object_body_cache_hook, + register_get_object_body_cache_hook, register_object_mutation_hook, unregister_get_object_body_cache_hook, + unregister_object_mutation_hook, }; pub use crate::store::{ PrepareSelectObjectSnapshotError, PreparedGetObjectReader, SelectObjectSnapshot, SelectObjectSnapshotReadError, diff --git a/crates/ecstore/src/disk/disk_store.rs b/crates/ecstore/src/disk/disk_store.rs index 7ce84ad06..415b530e4 100644 --- a/crates/ecstore/src/disk/disk_store.rs +++ b/crates/ecstore/src/disk/disk_store.rs @@ -250,6 +250,40 @@ pub(crate) trait DiskStoreRenameDataExt { dst_volume: &str, dst_path: &str, ) -> Result; + + async fn rename_data_borrowed_with_guard( + &self, + src_volume: &str, + src_path: &str, + fi: &FileInfo, + dst_volume: &str, + dst_path: &str, + external_guard: Option>, + ) -> Result { + let _ = external_guard; + self.rename_data_borrowed(src_volume, src_path, fi, dst_volume, dst_path) + .await + } +} + +/// Run a mutation in an owned task when a caller supplied publication guard. +/// RPC cancellation drops only the waiter; the mutation owner keeps the guard +/// until its operation has returned, including any detached blocking syscall. +async fn run_owned_mutation(external_guard: Option>, operation: F) -> Result +where + T: Send + 'static, + F: FnOnce() -> Fut + Send + 'static, + Fut: std::future::Future> + Send + 'static, +{ + if external_guard.is_none() { + return operation().await; + } + tokio::spawn(async move { + let _external_guard = external_guard; + operation().await + }) + .await + .map_err(|_| Error::other("owned mutation task failed"))? } impl DiskStoreRenameDataExt for LocalDiskWrapper { @@ -273,6 +307,49 @@ impl DiskStoreRenameDataExt for LocalDiskWrapper { ) .await } + + async fn rename_data_borrowed_with_guard( + &self, + src_volume: &str, + src_path: &str, + fi: &FileInfo, + dst_volume: &str, + dst_path: &str, + external_guard: Option>, + ) -> Result { + let operation = self.clone(); + let src_volume = src_volume.to_owned(); + let src_path = src_path.to_owned(); + let fi = fi.clone(); + let dst_volume = dst_volume.to_owned(); + let dst_path = dst_path.to_owned(); + let timeout_duration = if external_guard.is_some() { + // A fenced mutation owns the publication guard until the storage + // operation returns. Timing out this waiter would cancel the + // LocalDisk future while a spawn_blocking namespace syscall could + // still be committing, reopening the movement window. The caller + // may drop its waiter; the owned task drains the mutation. + Duration::ZERO + } else { + get_max_timeout_duration() + }; + run_owned_mutation(external_guard, move || async move { + operation + .track_disk_health_mutation( + "rename_data", + DiskMetricMutation::Write, + || async { + operation + .disk + .rename_data_borrowed(&src_volume, &src_path, &fi, &dst_volume, &dst_path) + .await + }, + timeout_duration, + ) + .await + }) + .await + } } pub fn get_drive_walkdir_timeout() -> Duration { @@ -1113,6 +1190,37 @@ impl LocalDiskWrapper { ) } + /// Run a delete under an owned coordinator task when a publication guard + /// is present. This keeps the guard alive if the RPC waiter is cancelled + /// while the local namespace mutation is still in progress. + pub(crate) async fn delete_with_publication_guard( + &self, + volume: &str, + path: &str, + options: DeleteOptions, + external_guard: Option>, + ) -> Result<()> { + let operation = self.clone(); + let volume = volume.to_owned(); + let path = path.to_owned(); + let timeout_duration = if external_guard.is_some() { + Duration::ZERO + } else { + get_max_timeout_duration() + }; + run_owned_mutation(external_guard, move || async move { + operation + .track_disk_health_mutation( + "delete", + DiskMetricMutation::Delete, + || async { operation.disk.delete(&volume, &path, options).await }, + timeout_duration, + ) + .await + }) + .await + } + pub(crate) fn new_with_reconnect_state( disk: Arc, health_check: bool, @@ -2263,6 +2371,44 @@ mod tests { }; use tokio::io::AsyncWrite; + struct DropProbe(Arc); + + impl Drop for DropProbe { + fn drop(&mut self) { + self.0.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + } + } + + #[tokio::test] + async fn owned_mutation_keeps_publication_guard_after_waiter_cancellation() { + let drops = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let guard: Arc = Arc::new(DropProbe(Arc::clone(&drops))); + let (started_tx, started_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = tokio::sync::oneshot::channel(); + let (finished_tx, finished_rx) = tokio::sync::oneshot::channel(); + + let waiter = tokio::spawn(run_owned_mutation(Some(guard), move || async move { + started_tx.send(()).expect("mutation should signal start"); + release_rx.await.expect("mutation should be released"); + finished_tx.send(()).expect("mutation should signal completion"); + Ok::<_, Error>(()) + })); + + started_rx.await.expect("mutation owner should start"); + waiter.abort(); + assert_eq!(drops.load(std::sync::atomic::Ordering::SeqCst), 0); + + release_tx.send(()).expect("mutation owner should still be alive"); + finished_rx.await.expect("mutation owner should finish"); + tokio::time::timeout(Duration::from_secs(1), async { + while drops.load(std::sync::atomic::Ordering::SeqCst) == 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("publication guard should be released after mutation completion"); + } + struct PendingWriter; #[test] diff --git a/crates/ecstore/src/disk/mod.rs b/crates/ecstore/src/disk/mod.rs index fb69085c4..7801274ef 100644 --- a/crates/ecstore/src/disk/mod.rs +++ b/crates/ecstore/src/disk/mod.rs @@ -677,15 +677,20 @@ impl DiskAPI for Disk { } impl Disk { - pub(crate) async fn delete_with_scanner_publication_lease( + pub async fn delete_with_scanner_publication_lease_and_guard( &self, volume: &str, path: &str, opts: DeleteOptions, scanner_publication_lease_token: Option, + external_guard: Option>, ) -> Result<()> { match self { - Disk::Local(local_disk) => local_disk.delete(volume, path, opts).await, + Disk::Local(local_disk) => { + local_disk + .delete_with_publication_guard(volume, path, opts, external_guard) + .await + } Disk::Remote(remote_disk) => { remote_disk .delete_with_scanner_publication_lease(volume, path, opts, scanner_publication_lease_token) @@ -714,11 +719,34 @@ impl Disk { dst_volume: &str, dst_path: &str, scanner_publication_lease_token: Option, + ) -> Result { + self.rename_data_borrowed_with_fence_and_guard( + src_volume, + src_path, + fi, + dst_volume, + dst_path, + scanner_publication_lease_token, + None, + ) + .await + } + + #[allow(clippy::too_many_arguments)] + pub async fn rename_data_borrowed_with_fence_and_guard( + &self, + src_volume: &str, + src_path: &str, + fi: &FileInfo, + dst_volume: &str, + dst_path: &str, + scanner_publication_lease_token: Option, + external_guard: Option>, ) -> Result { match self { Disk::Local(local_disk) => { local_disk - .rename_data_borrowed(src_volume, src_path, fi, dst_volume, dst_path) + .rename_data_borrowed_with_guard(src_volume, src_path, fi, dst_volume, dst_path, external_guard) .await } Disk::Remote(remote_disk) => { diff --git a/crates/ecstore/src/object_api/types.rs b/crates/ecstore/src/object_api/types.rs index be761a8c7..694fbf0c0 100644 --- a/crates/ecstore/src/object_api/types.rs +++ b/crates/ecstore/src/object_api/types.rs @@ -19,6 +19,9 @@ use crate::storage_api_contracts::{ HTTPPreconditions, ObjectLockRetentionOptions, ObjectPreconditionError, ObjectPreconditionPart, ObjectPreconditionState, }, }; +use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; +use tokio::sync::{Mutex, Notify, OwnedRwLockReadGuard}; +use tokio_util::sync::CancellationToken; #[derive(Clone)] pub struct NamespaceLockFence { @@ -347,6 +350,337 @@ impl QuotaAdmission { } } +const SCANNER_PUBLICATION_SCOPE_ADMITTED: u8 = 0; +const SCANNER_PUBLICATION_SCOPE_IN_FLIGHT: u8 = 1; +const SCANNER_PUBLICATION_SCOPE_COMMITTED: u8 = 2; +const SCANNER_PUBLICATION_SCOPE_ABORTED_BEFORE_COMMIT: u8 = 3; +const SCANNER_PUBLICATION_SCOPE_INDETERMINATE: u8 = 4; + +/// The terminal result of a storage-owned scanner publication mutation. +/// +/// This state is deliberately not serialized. It is the ownership hand-off +/// between the scanner coordinator and the storage mutation task, so a +/// detached rename/cleanup task can retain the movement permit until it has +/// reported a definitive result. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ScannerPublicationCommitState { + Admitted, + InFlight, + Committed, + AbortedBeforeCommit, + Indeterminate, +} + +impl ScannerPublicationCommitState { + fn as_u8(self) -> u8 { + match self { + Self::Admitted => SCANNER_PUBLICATION_SCOPE_ADMITTED, + Self::InFlight => SCANNER_PUBLICATION_SCOPE_IN_FLIGHT, + Self::Committed => SCANNER_PUBLICATION_SCOPE_COMMITTED, + Self::AbortedBeforeCommit => SCANNER_PUBLICATION_SCOPE_ABORTED_BEFORE_COMMIT, + Self::Indeterminate => SCANNER_PUBLICATION_SCOPE_INDETERMINATE, + } + } + + fn from_u8(value: u8) -> Self { + match value { + SCANNER_PUBLICATION_SCOPE_IN_FLIGHT => Self::InFlight, + SCANNER_PUBLICATION_SCOPE_COMMITTED => Self::Committed, + SCANNER_PUBLICATION_SCOPE_ABORTED_BEFORE_COMMIT => Self::AbortedBeforeCommit, + SCANNER_PUBLICATION_SCOPE_INDETERMINATE => Self::Indeterminate, + _ => Self::Admitted, + } + } + + /// A caller may release its remote lease only after one of these states. + /// `Indeterminate` is intentionally excluded: the mutation may have + /// committed after cancellation or a transport failure. + pub fn permits_lease_release(self) -> bool { + matches!(self, Self::Committed | Self::AbortedBeforeCommit) + } +} + +/// Why a storage-owned publication scope could not start its mutation. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ScannerPublicationCommitStartError { + Cancelled, + DeadlineExceeded, + AlreadyStarted, + Terminal, +} + +struct ScannerPublicationCommitScopeInner { + expected_movement_epoch: u64, + safe_deadline: tokio::time::Instant, + remote_lease_tokens: Arc<[Uuid]>, + cancellation: CancellationToken, + state: AtomicU8, + completed: Notify, + /// Set once a storage mutation task has taken ownership of the scope. + /// The caller-side RAII guard must not classify cancellation as + /// indeterminate while that owner can still report a definitive result. + owner_attached: AtomicBool, + /// The permit is storage-owned rather than borrowed from the scanner + /// 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. +/// +/// The scope is an in-memory capability. It is intentionally carried through +/// [`ObjectOptions`] as a hidden field and never participates in serde, object +/// metadata, RPC wire structures, or on-disk formats. +#[derive(Clone)] +pub struct ScannerPublicationCommitScope { + inner: Arc, +} + +/// RAII fallback for storage paths that return before their commit closure +/// takes ownership. An in-flight scope is never guessed to be aborted: it is +/// marked indeterminate so remote lease release remains blocked. +pub(crate) struct ScannerPublicationCommitScopeGuard { + scope: Option, +} + +impl ScannerPublicationCommitScopeGuard { + pub(crate) fn new(scope: ScannerPublicationCommitScope) -> Self { + Self { scope: Some(scope) } + } + + pub(crate) fn disarm(&mut self) { + self.scope = None; + } +} + +impl Drop for ScannerPublicationCommitScopeGuard { + fn drop(&mut self) { + let Some(scope) = self.scope.as_ref() else { + return; + }; + if scope.owner_attached() { + return; + } + match scope.state() { + ScannerPublicationCommitState::Admitted => { + let _ = scope.mark_aborted_before_commit(); + } + ScannerPublicationCommitState::InFlight => { + let _ = scope.mark_indeterminate(); + } + ScannerPublicationCommitState::Committed + | ScannerPublicationCommitState::AbortedBeforeCommit + | ScannerPublicationCommitState::Indeterminate => {} + } + } +} + +impl Debug for ScannerPublicationCommitScope { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ScannerPublicationCommitScope") + .field("expected_movement_epoch", &self.expected_movement_epoch()) + .field("safe_deadline", &self.safe_deadline()) + .field("remote_lease_token_count", &self.remote_lease_tokens().len()) + .field("state", &self.state()) + .finish() + } +} + +impl ScannerPublicationCommitScope { + /// Construct a scope after the storage layer has acquired its movement + /// read permit. Callers must keep the scope attached to the actual + /// mutation owner until [`Self::wait_for_completion`] has resolved. + pub(crate) fn new_storage_owned( + expected_movement_epoch: u64, + safe_deadline: tokio::time::Instant, + 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(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 { + lease_release_safe.store(false, Ordering::Release); + Self { + inner: Arc::new(ScannerPublicationCommitScopeInner { + expected_movement_epoch, + safe_deadline, + remote_lease_tokens: remote_lease_tokens.into(), + cancellation: CancellationToken::new(), + state: AtomicU8::new(SCANNER_PUBLICATION_SCOPE_ADMITTED), + completed: Notify::new(), + owner_attached: AtomicBool::new(false), + movement_permit: Mutex::new(Some(movement_permit)), + lease_release_safe, + }), + } + } + + pub fn expected_movement_epoch(&self) -> u64 { + self.inner.expected_movement_epoch + } + + pub fn safe_deadline(&self) -> tokio::time::Instant { + self.inner.safe_deadline + } + + pub fn is_expired(&self) -> bool { + tokio::time::Instant::now() >= self.safe_deadline() + } + + pub fn remote_lease_tokens(&self) -> &[Uuid] { + &self.inner.remote_lease_tokens + } + + pub fn cancellation_token(&self) -> CancellationToken { + self.inner.cancellation.clone() + } + + pub fn is_cancelled(&self) -> bool { + self.inner.cancellation.is_cancelled() + } + + /// Whether a mutation that has already begun may still enter its durable + /// commit boundary. The storage owner must check this immediately before + /// starting each irreversible fan-out/rename operation. + pub fn can_commit(&self) -> bool { + self.state() == ScannerPublicationCommitState::InFlight && !self.is_cancelled() && !self.is_expired() + } + + /// Transfer terminal-state responsibility from the caller to a detached + /// storage mutation owner. Once set, dropping a scanner waiter leaves the + /// scope in-flight until that owner reports committed or indeterminate. + pub fn attach_mutation_owner(&self) { + self.inner.owner_attached.store(true, Ordering::Release); + } + + fn owner_attached(&self) -> bool { + self.inner.owner_attached.load(Ordering::Acquire) + } + + pub fn state(&self) -> ScannerPublicationCommitState { + ScannerPublicationCommitState::from_u8(self.inner.state.load(Ordering::Acquire)) + } + + /// Request cancellation without claiming that a mutation has stopped. + /// The owner must still report `AbortedBeforeCommit` or `Indeterminate`. + pub fn cancel(&self) { + self.inner.cancellation.cancel(); + } + + pub fn try_begin(&self) -> std::result::Result<(), ScannerPublicationCommitStartError> { + if self.inner.cancellation.is_cancelled() { + return Err(ScannerPublicationCommitStartError::Cancelled); + } + if self.is_expired() { + return Err(ScannerPublicationCommitStartError::DeadlineExceeded); + } + self.inner + .state + .compare_exchange( + SCANNER_PUBLICATION_SCOPE_ADMITTED, + SCANNER_PUBLICATION_SCOPE_IN_FLIGHT, + Ordering::AcqRel, + Ordering::Acquire, + ) + .map(|_| ()) + .map_err(|state| { + if ScannerPublicationCommitState::from_u8(state).permits_lease_release() { + ScannerPublicationCommitStartError::Terminal + } else { + ScannerPublicationCommitStartError::AlreadyStarted + } + }) + } + + pub fn mark_committed(&self) -> bool { + self.mark_terminal(ScannerPublicationCommitState::Committed) + } + + pub fn mark_aborted_before_commit(&self) -> bool { + 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 + } + + pub fn mark_indeterminate(&self) -> bool { + self.mark_terminal(ScannerPublicationCommitState::Indeterminate) + } + + fn mark_terminal(&self, terminal: ScannerPublicationCommitState) -> bool { + self.inner + .state + .compare_exchange(SCANNER_PUBLICATION_SCOPE_IN_FLIGHT, terminal.as_u8(), Ordering::AcqRel, Ordering::Acquire) + .is_ok() + .then(|| { + if terminal.permits_lease_release() { + self.inner.lease_release_safe.store(true, Ordering::Release); + } + self.inner.completed.notify_waiters() + }) + .is_some() + } + + /// Wait until the mutation owner has reported a definitive terminal + /// state. The permit remains owned by this scope until all scope clones are + /// dropped or [`Self::release_movement_permit`] is called safely. + pub async fn wait_for_completion(&self) -> ScannerPublicationCommitState { + loop { + let notified = self.inner.completed.notified(); + tokio::pin!(notified); + notified.as_mut().enable(); + let state = self.state(); + if state != ScannerPublicationCommitState::Admitted && state != ScannerPublicationCommitState::InFlight { + return state; + } + notified.await; + } + } + + /// Release the storage-owned movement permit only after a known-safe + /// terminal result. Returns `false` for in-flight or indeterminate work. + pub async fn release_movement_permit(&self) -> bool { + if !self.state().permits_lease_release() { + return false; + } + self.inner.movement_permit.lock().await.take().is_some() + } +} + +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 @@ -384,6 +718,11 @@ pub struct ObjectOptions { #[doc(hidden)] pub put_object_cancellation: Option, + /// Storage-owned scanner publication capability. This field is an + /// in-memory hand-off only; it is never copied into object metadata. + #[doc(hidden)] + pub scanner_publication_commit_scope: Option, + pub data_movement: bool, pub raw_data_movement_read: bool, /// Materialize the data-movement per-part checksum sidecar for APIs that @@ -473,6 +812,7 @@ impl std::fmt::Debug for ObjectOptions { .field("skip_rebalancing", &self.skip_rebalancing) .field("skip_free_version", &self.skip_free_version) .field("put_object_cancellation", &self.put_object_cancellation.is_some()) + .field("scanner_publication_commit_scope", &self.scanner_publication_commit_scope) .field("data_movement", &self.data_movement) .field("raw_data_movement_read", &self.raw_data_movement_read) .field("include_part_checksums", &self.include_part_checksums) diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index 03d29ff86..37fba0dca 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -3657,6 +3657,7 @@ pub(in crate::set_disk) struct RenameTailOutcome { pub(in crate::set_disk) struct RenameDataFenceOptions<'a> { write_quorum: usize, scanner_publication_lease_tokens: Option<&'a HashMap>, + scanner_publication_commit_scope: Option, } impl<'a> RenameDataFenceOptions<'a> { @@ -3667,8 +3668,17 @@ impl<'a> RenameDataFenceOptions<'a> { Self { write_quorum, scanner_publication_lease_tokens, + scanner_publication_commit_scope: None, } } + + pub(in crate::set_disk) fn with_publication_scope( + mut self, + scanner_publication_commit_scope: Option, + ) -> Self { + self.scanner_publication_commit_scope = scanner_publication_commit_scope; + self + } } #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")] @@ -3778,6 +3788,37 @@ pub(in crate::set_disk) async fn finish_rename_tail_heal< } } +async fn run_scanner_publication_delete_owner( + scope: Option, + operation: F, +) -> disk::error::Result<()> +where + F: FnOnce() -> Fut + Send + 'static, + Fut: Future> + Send + 'static, +{ + if scope.is_none() { + return operation().await; + } + if let Some(scope) = scope.as_ref() { + scope.attach_mutation_owner(); + } + tokio::spawn(async move { + let result = operation().await; + if let Some(scope) = scope.as_ref() { + if result.is_ok() { + let _ = scope.mark_committed(); + } else { + // A failed quorum does not prove that no replica committed; + // keep the permit indeterminate for supervisor reconciliation. + let _ = scope.mark_indeterminate(); + } + } + result + }) + .await + .map_err(|_| DiskError::other("scanner publication delete owner failed"))? +} + impl SetDisks { pub(in crate::set_disk) fn default_read_quorum(&self) -> usize { self.set_drive_count - self.default_parity_count @@ -3995,6 +4036,7 @@ impl SetDisks { let RenameDataFenceOptions { write_quorum, scanner_publication_lease_tokens, + scanner_publication_commit_scope: _scanner_publication_commit_scope, } = fence_options; if let Some(file_info) = disks .iter() @@ -4352,6 +4394,7 @@ impl SetDisks { let RenameDataFenceOptions { write_quorum, scanner_publication_lease_tokens, + scanner_publication_commit_scope, } = fence_options; if let Some(file_info) = disks .iter() @@ -4383,11 +4426,15 @@ impl SetDisks { let fanout_src_object = src_object.clone(); let fanout_dst_bucket = dst_bucket.clone(); let fanout_dst_object = dst_object.clone(); + let fanout_publication_scope = scanner_publication_commit_scope.clone(); // Keep one coordinator task so a cancelled caller cannot drop partially // completed disk mutations. Per-disk futures stay ordered in `join_all`, // preserving slot-indexed quorum and convergence accounting without a // scheduler task for every disk. let fanout = tokio::spawn(async move { + // Keep the storage-owned movement permit attached to the actual + // fan-out owner, even if the caller future is cancelled. + let _fanout_publication_scope = fanout_publication_scope; let successful_rename_completion_rank = rustfs_io_metrics::put_stage_metrics_enabled().then(|| Arc::new(AtomicUsize::new(0))); let futures = fanout_disks @@ -4401,6 +4448,7 @@ impl SetDisks { let dst_object = fanout_dst_object.clone(); let dst_bucket = fanout_dst_bucket.clone(); let successful_rename_completion_rank = successful_rename_completion_rank.clone(); + let publication_scope = scanner_publication_commit_scope.clone(); std::panic::AssertUnwindSafe(async move { // Test-only introspection guard: counts this operation as @@ -4433,6 +4481,13 @@ impl SetDisks { return Err(err); } + if let Some(scope) = publication_scope.as_ref() + && !scope.can_commit() + { + let _ = scope.mark_indeterminate(); + return Err(DiskError::other("scanner publication commit scope deadline or cancellation reached")); + } + let disk_wait_started = rustfs_io_metrics::put_stage_timer(); let result = disk .rename_data_borrowed_with_fence( @@ -5841,7 +5896,8 @@ impl SetDisks { #[cfg(test)] pub(in crate::set_disk) async fn delete_prefix(&self, bucket: &str, prefix: &str) -> disk::error::Result<()> { - self.delete_prefix_with_scanner_publication_lease(bucket, prefix, None).await + self.delete_prefix_with_scanner_publication_lease(bucket, prefix, None, None) + .await } /// Delete a prefix with an optional per-remote-disk scanner publication @@ -5852,6 +5908,7 @@ impl SetDisks { bucket: &str, prefix: &str, scanner_publication_lease_tokens: Option<&HashMap>, + scanner_publication_commit_scope: Option, ) -> disk::error::Result<()> { let disks = self.get_disks_internal().await; let write_quorum = disks.len() / 2 + 1; @@ -5860,11 +5917,21 @@ impl SetDisks { let mut futures = Vec::with_capacity(disks.len()); for (disk_op, scanner_publication_lease_token) in disks.iter().zip(fanout_fence_tokens) { + let disk_op = disk_op.clone(); let bucket = bucket.to_string(); let prefix = prefix.to_string(); + let scanner_publication_commit_scope = scanner_publication_commit_scope.clone(); futures.push(async move { if let Some(disk) = disk_op { - disk.delete_with_scanner_publication_lease( + if let Some(scope) = scanner_publication_commit_scope.as_ref() + && !scope.can_commit() + { + return Err(DiskError::other("scanner publication delete scope cannot commit")); + } + let external_guard = scanner_publication_commit_scope + .as_ref() + .map(|scope| Arc::new(scope.clone()) as Arc); + disk.delete_with_scanner_publication_lease_and_guard( &bucket, &prefix, DeleteOptions { @@ -5873,6 +5940,7 @@ impl SetDisks { ..Default::default() }, scanner_publication_lease_token, + external_guard, ) .await } else { @@ -5881,7 +5949,10 @@ impl SetDisks { }); } - Self::reduce_delete_prefix_results(join_all(futures).await, write_quorum) + run_scanner_publication_delete_owner(scanner_publication_commit_scope, move || async move { + Self::reduce_delete_prefix_results(join_all(futures).await, write_quorum) + }) + .await } /// Scan a single disk's copy of `prefix` and decide whether it is an orphan @@ -6809,6 +6880,63 @@ mod tests { use tempfile::TempDir; use tokio::io::AsyncReadExt; + #[tokio::test] + async fn scanner_delete_owner_survives_waiter_cancellation() { + let movement_gate = Arc::new(tokio::sync::RwLock::new(())); + let movement_permit = movement_gate.clone().read_owned().await; + let scope = crate::object_api::ScannerPublicationCommitScope::new_storage_owned( + 7, + tokio::time::Instant::now() + std::time::Duration::from_secs(30), + Vec::new(), + movement_permit, + ); + scope.try_begin().expect("delete scope should enter flight"); + let scope_guard = crate::object_api::ScannerPublicationCommitScopeGuard::new(scope.clone()); + let (started_tx, started_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = tokio::sync::oneshot::channel(); + let (finished_tx, finished_rx) = tokio::sync::oneshot::channel(); + + let waiter = tokio::spawn(run_scanner_publication_delete_owner(Some(scope.clone()), move || async move { + started_tx.send(()).expect("delete owner should start"); + release_rx.await.expect("delete owner should be released"); + finished_tx.send(()).expect("delete owner should finish"); + Ok(()) + })); + started_rx.await.expect("delete owner should run"); + drop(scope_guard); + waiter.abort(); + assert_eq!( + scope.state(), + crate::object_api::ScannerPublicationCommitState::InFlight, + "caller cancellation must not classify an owned delete as indeterminate" + ); + + let mut movement_writer = Box::pin(movement_gate.write_owned()); + assert!( + tokio::time::timeout(std::time::Duration::from_millis(20), &mut movement_writer) + .await + .is_err(), + "movement transition must remain fenced while delete owner drains" + ); + release_tx.send(()).expect("delete owner should remain alive"); + finished_rx.await.expect("delete owner should drain"); + tokio::time::timeout(std::time::Duration::from_secs(1), async { + loop { + if scope.state() == crate::object_api::ScannerPublicationCommitState::Committed { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("delete owner should report a terminal result"); + assert!( + scope.release_movement_permit().await, + "terminal delete should release its movement permit" + ); + movement_writer.await; + } + #[test] fn write_precondition_lookup_errors_fail_closed_unless_absence_is_known() { let create_only = HTTPPreconditions { diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 4403d0db8..88ebc3ac9 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -3938,6 +3938,44 @@ impl SetDisks { owner.scanner_data_usage_publication_admission_guard().await } + 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(crate::object_api::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(crate::object_api::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 c92ad68de..4d08ceba7 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -66,6 +66,7 @@ use crate::bucket::lifecycle::bucket_lifecycle_ops::LifecycleOps; use crate::bucket::utils::is_meta_bucketname; use crate::bucket::versioning::VersioningApi; use crate::disk::DiskAPI; +use crate::object_api::ScannerPublicationCommitScopeGuard; use crate::set_disk::coding; use crate::set_disk::core::io_primitives::GetCodecStreamingReaderBuildOutcome; use crate::set_disk::mem; @@ -272,6 +273,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(|_| Error::other("scanner publication delete scope cannot start"))?; + } + 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); @@ -2627,6 +2644,10 @@ impl SetDisks { opts: &ObjectOptions, ) -> Result<(ObjectInfo, Option)> { crate::hp_guard!("SetDisks::put_object"); + let mut scope_outcome_guard = opts + .scanner_publication_commit_scope + .clone() + .map(ScannerPublicationCommitScopeGuard::new); let storage_class_config = self.storage_class_config_snapshot(); self.invalidate_get_object_metadata_cache(bucket, object).await; @@ -3397,8 +3418,16 @@ 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 detach_commit_owner = commit_allows_early_ack || commit_bucket_lifecycle_guard.is_some() || quota_mutation_fence; + 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_scanner_publication_scope.is_some() + || 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; let commit_versioned = opts.versioned; @@ -3494,7 +3523,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()) => { @@ -3505,6 +3534,20 @@ 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 pre_rename_result.is_ok() + && let Some(scope) = commit_scanner_publication_scope.as_ref() + && !scope.can_commit() + { + let _ = scope.mark_indeterminate(); + pre_rename_result = Err(StorageError::OperationCanceled); + } if let Err(err) = pre_rename_result { SetDisks::abort_quota_reservation_after_fence( quota_reservation, @@ -3540,9 +3583,17 @@ impl SetDisks { crate::set_disk::core::io_primitives::RenameDataFenceOptions::new( write_quorum, commit_scanner_publication_lease_tokens.as_ref(), - ), + ) + .with_publication_scope(commit_scanner_publication_scope.clone()), ) .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; @@ -3857,6 +3908,11 @@ impl SetDisks { let _ = handoff.send(()); } if detach_commit_owner { + if let Some(scope_outcome_guard) = scope_outcome_guard.as_mut() { + // The spawned commit closure owns the scope clone and is + // now responsible for its terminal outcome. + scope_outcome_guard.disarm(); + } let mut cancellation = PutObjectCommitCancellation::new(); let child_token = cancellation.child_token(); let result = tokio::spawn(async move { Box::pin(commit(Some(child_token))).await }) @@ -7054,6 +7110,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. @@ -7148,6 +7209,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; @@ -7162,6 +7224,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 { @@ -7173,10 +7236,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()); } @@ -7184,10 +7251,19 @@ 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)?; - 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]))?; + 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(), + scanner_publication_commit_scope.clone(), + ) + .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()); } @@ -7260,10 +7336,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)); } @@ -7337,6 +7417,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]))?; @@ -7348,6 +7429,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); } @@ -7373,6 +7457,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]))?; @@ -7398,6 +7483,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/ecstore/src/store/mod.rs b/crates/ecstore/src/store/mod.rs index 0c874690f..c07a25fab 100644 --- a/crates/ecstore/src/store/mod.rs +++ b/crates/ecstore/src/store/mod.rs @@ -58,7 +58,7 @@ use crate::{ core::sets::Sets, disk::{BUCKET_META_PREFIX, DiskOption, DiskStore, RUSTFS_META_BUCKET}, layout::endpoints::EndpointServerPools, - object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader}, + object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader, ScannerPublicationCommitScope}, }; use futures::future::join_all; use http::HeaderMap; @@ -522,6 +522,48 @@ impl ECStore { Some((operation_guard, self.ctx.data_movement_operation_epoch())) } + /// Acquire a storage-owned scanner publication scope. Unlike the legacy + /// admission helper, the movement permit is owned by the returned scope + /// and therefore survives cancellation of the scanner coordinator while + /// the actual metadata mutation 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, + )) + } + /// 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. @@ -1409,9 +1451,29 @@ mod tests { .await .expect("movement writer should proceed after lease expiry") .expect("expiry writer task should not panic"); + assert!( + store.validate_scanner_publication_lease(expiring_token, 0).await.is_err(), + "an expired lease must not validate after its read guard is released" + ); assert!(!store.release_scanner_publication_lease(expiring_token).await); } + #[tokio::test] + async fn scanner_publication_lease_rejects_a_new_movement_generation() { + let store = build_store_with_ctx(Arc::new(InstanceContext::new())); + let (token, generation) = store + .acquire_scanner_publication_lease(0, crate::runtime::instance::SCANNER_PUBLICATION_LEASE_TTL) + .await + .expect("an idle store should grant a publication lease"); + + assert_eq!(store.ctx.advance_data_movement_generation(), Some(1)); + assert!( + store.validate_scanner_publication_lease(token, generation).await.is_err(), + "a lease from the prior movement generation must fail closed" + ); + assert!(store.release_scanner_publication_lease(token).await); + } + #[tokio::test] async fn scanner_publication_lease_rejects_stale_generation_before_install() { let store = build_store_with_ctx(Arc::new(InstanceContext::new())); @@ -1422,6 +1484,102 @@ mod tests { assert!(error.to_string().contains("generation is stale")); } + #[tokio::test(start_paused = true)] + async fn scanner_publication_commit_scope_owns_permit_until_terminal_drain() { + let store = build_store_with_ctx(Arc::new(InstanceContext::new())); + let scope = store + .scanner_data_usage_publication_commit_scope( + 0, + tokio::time::Instant::now() + Duration::from_secs(30), + vec![Uuid::new_v4()], + ) + .await + .expect("idle storage should grant a publication scope"); + assert_eq!(scope.state(), crate::object_api::ScannerPublicationCommitState::Admitted); + assert_eq!(scope.remote_lease_tokens().len(), 1); + + let gate = store.ctx.data_movement_operation_gate(); + let writer = tokio::spawn(async move { gate.write_owned().await }); + tokio::task::yield_now().await; + assert!(!writer.is_finished(), "the scope must own its movement permit after the caller returns"); + + scope.cancel(); + assert!(scope.mark_aborted_before_commit()); + assert_eq!( + scope.wait_for_completion().await, + crate::object_api::ScannerPublicationCommitState::AbortedBeforeCommit + ); + assert!(scope.release_movement_permit().await); + tokio::time::timeout(Duration::from_secs(1), writer) + .await + .expect("movement writer should proceed after the scope drains") + .expect("movement writer task should not panic"); + } + + #[tokio::test(start_paused = true)] + async fn scanner_publication_commit_scope_rejects_late_start_and_keeps_indeterminate_permit() { + let store = build_store_with_ctx(Arc::new(InstanceContext::new())); + let scope = store + .scanner_data_usage_publication_commit_scope(0, tokio::time::Instant::now() + Duration::from_secs(1), Vec::new()) + .await + .expect("idle storage should grant a publication scope"); + tokio::time::advance(Duration::from_secs(1)).await; + assert_eq!( + scope.try_begin(), + Err(crate::object_api::ScannerPublicationCommitStartError::DeadlineExceeded) + ); + assert!( + !scope.release_movement_permit().await, + "an admitted scope is not safe to release before owner resolution" + ); + assert!(scope.mark_aborted_before_commit()); + assert!(scope.release_movement_permit().await); + + let scope = store + .scanner_data_usage_publication_commit_scope(0, tokio::time::Instant::now() + Duration::from_secs(30), Vec::new()) + .await + .expect("a second idle publication scope should be granted"); + scope.try_begin().expect("scope should enter the mutation state"); + scope.cancel(); + assert!(scope.mark_indeterminate()); + assert_eq!( + scope.wait_for_completion().await, + crate::object_api::ScannerPublicationCommitState::Indeterminate + ); + assert!(!scope.release_movement_permit().await, "indeterminate mutation must retain the permit"); + } + + #[tokio::test] + async fn scanner_publication_scope_guard_classifies_early_returns_conservatively() { + let store = build_store_with_ctx(Arc::new(InstanceContext::new())); + let permit = store.ctx.data_movement_operation_gate().read_owned().await; + let scope = ScannerPublicationCommitScope::new_storage_owned( + 0, + tokio::time::Instant::now() + Duration::from_secs(30), + Vec::new(), + permit, + ); + { + let _guard = crate::object_api::ScannerPublicationCommitScopeGuard::new(scope.clone()); + } + assert_eq!(scope.state(), crate::object_api::ScannerPublicationCommitState::AbortedBeforeCommit); + assert!(scope.release_movement_permit().await); + + let permit = store.ctx.data_movement_operation_gate().read_owned().await; + let scope = ScannerPublicationCommitScope::new_storage_owned( + 0, + tokio::time::Instant::now() + Duration::from_secs(30), + Vec::new(), + permit, + ); + scope.try_begin().expect("scope should enter the mutation state"); + { + let _guard = crate::object_api::ScannerPublicationCommitScopeGuard::new(scope.clone()); + } + assert_eq!(scope.state(), crate::object_api::ScannerPublicationCommitState::Indeterminate); + assert!(!scope.release_movement_permit().await); + } + #[tokio::test] async fn scanner_target_guard_keeps_movement_writer_fenced_after_lease_release() { let store = build_store_with_ctx(Arc::new(InstanceContext::new())); diff --git a/crates/scanner/src/lib.rs b/crates/scanner/src/lib.rs index c84588b3d..3a8cb8d6f 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; @@ -752,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, @@ -762,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 @@ -796,13 +820,14 @@ where Some(admission) } -pub(crate) async fn save_config_shared_with_preconditions_and_lease_fence( +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 a1a832d9a..a83b8a737 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}; @@ -62,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, @@ -71,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, }; @@ -453,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, @@ -460,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 { @@ -530,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 { @@ -1546,26 +1604,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); @@ -1579,6 +1627,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, @@ -1590,7 +1643,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(); @@ -1655,7 +1710,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 a5877802c..5cc87c063 100644 --- a/crates/scanner/src/scanner/tests.rs +++ b/crates/scanner/src/scanner/tests.rs @@ -352,6 +352,7 @@ struct MemoryConfigStore { objects: Mutex>>, revisions: Mutex>, insert_after_gets: Mutex>>, + delayed_gets: Mutex>, non_regular_objects: Mutex>, fail_put_number: Mutex>, object_not_found_put_number: Mutex>, @@ -399,6 +400,9 @@ impl crate::storage_api::scanner_io::ObjectIO for MemoryConfigStore { _opts: &ObjectOptions, ) -> EcstoreResult { let key = memory_config_key(bucket, object); + if let Some(delay) = self.delayed_gets.lock().await.remove(&key) { + tokio::time::sleep(delay).await; + } let inserted_data = self.insert_after_gets.lock().await.remove(&key); let data = { let mut objects = self.objects.lock().await; @@ -3532,6 +3536,47 @@ async fn coordinator_classifies_an_expired_publication_lease() { assert!(store.put_counts.lock().await.is_empty(), "expired lease must prevent a PUT"); } +#[tokio::test] +async fn backup_sync_checks_the_lease_deadline_after_a_slow_backup_read() { + let store = Arc::new(MemoryConfigStore::default()); + let primary_path = DATA_USAGE_OBJ_NAME_PATH.as_str(); + let backup_path = format!("{primary_path}.bkp"); + let primary_key = memory_config_key(RUSTFS_META_BUCKET, primary_path); + let backup_key = memory_config_key(RUSTFS_META_BUCKET, &backup_path); + let primary = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0); + store + .objects + .lock() + .await + .insert(primary_key, serde_json::to_vec(&primary).expect("primary usage snapshot should encode")); + store + .delayed_gets + .lock() + .await + .insert(backup_key.clone(), Duration::from_millis(20)); + + // The primary read is allowed to start, but the backup read consumes the + // remaining lease window. The second deadline check must prevent a stale + // backup PUT after that window has elapsed. + let deadline = std::time::Instant::now() + .checked_add(std::time::Duration::from_millis(5)) + .expect("test deadline should support a five-millisecond window"); + let result = sync_data_usage_backup_from_primary_for_epoch_and_lease_and_fence( + &CancellationToken::new(), + store.clone(), + None, + Some(deadline), + None, + ) + .await; + + assert!(scanner_publication_epoch_changed( + &result.expect_err("an expired backup lease must defer publication") + )); + assert!(!store.objects.lock().await.contains_key(&backup_key)); + assert_eq!(store.put_counts.lock().await.get(&backup_key), None); +} + #[tokio::test] #[serial] async fn test_deferred_usage_save_keeps_last_real_save_metric() { @@ -4783,6 +4828,29 @@ async fn data_usage_persist_wait_aborts_after_timeout() { assert!(task.is_finished()); } +#[tokio::test(start_paused = true)] +async fn data_usage_persist_timeout_drops_owned_task_without_a_late_commit() { + let ctx = CancellationToken::new(); + let commit_started = Arc::new(AtomicBool::new(false)); + let commit_started_by_task = commit_started.clone(); + let task_ready = Arc::new(tokio::sync::Notify::new()); + let task_ready_by_task = task_ready.clone(); + let mut task = AbortOnDropHandle::new(tokio::spawn(async move { + task_ready_by_task.notify_one(); + std::future::pending::<()>().await; + commit_started_by_task.store(true, Ordering::Release); + DataUsagePersistOutcome::Saved + })); + task_ready.notified().await; + + let result = wait_for_data_usage_persist_task(&ctx, &mut task, Duration::from_secs(1)).await; + + assert!(matches!(result, DataUsagePersistTaskResult::TimedOut)); + assert!(task.is_finished(), "the timed-out persistence task must be drained before return"); + tokio::task::yield_now().await; + assert!(!commit_started.load(Ordering::Acquire), "an owned task must not commit after its timeout"); +} + #[tokio::test(start_paused = true)] async fn maintenance_feature_inspection_preserves_base_cycle_after_timeout() { let ctx = CancellationToken::new(); diff --git a/crates/scanner/src/scanner/usage_store.rs b/crates/scanner/src/scanner/usage_store.rs index 23bfe60d7..9d181d52a 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) } +pub(super) 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(); @@ -696,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 { @@ -719,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 { @@ -761,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 { @@ -778,12 +842,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(); @@ -817,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; @@ -885,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(), @@ -904,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( diff --git a/crates/scanner/src/storage_api.rs b/crates/scanner/src/storage_api.rs index 4c406185c..2ac236f96 100644 --- a/crates/scanner/src/storage_api.rs +++ b/crates/scanner/src/storage_api.rs @@ -93,7 +93,9 @@ pub(crate) use rustfs_ecstore::api::layout::{ EndpointServerPools as EcstoreEndpointServerPools, Endpoints as EcstoreEndpoints, PoolEndpoints as EcstorePoolEndpoints, }; pub(crate) use rustfs_ecstore::api::notification::scanner_peer_transport_error_message_is_retryable; -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, @@ -126,14 +128,15 @@ pub(crate) mod owner { EcstoreNsScannerOpenRequest, EcstoreObjectLockConfiguration, EcstoreObjectOpts, EcstoreReplicationConfigurationExt, EcstoreReplicationScannerBridge, EcstoreResultType, EcstoreScanGuard, EcstoreSetDisks, EcstoreStorageError, EcstoreStore, EcstoreVersioningApi, EcstoreVersioningConfiguration, SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY, - 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, ecstore_send_event, scanner_replication_config_for_lifecycle_eval, + 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, + 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, + ecstore_send_event, scanner_replication_config_for_lifecycle_eval, }; #[cfg(test)] diff --git a/rustfs/src/storage/rpc/node_service/disk.rs b/rustfs/src/storage/rpc/node_service/disk.rs index e2457c996..19fb86c80 100644 --- a/rustfs/src/storage/rpc/node_service/disk.rs +++ b/rustfs/src/storage/rpc/node_service/disk.rs @@ -33,6 +33,7 @@ use rustfs_io_metrics::internode_metrics::{ use rustfs_protos::proto_gen::node_service::*; use serde::de::DeserializeOwned; use std::io::Cursor; +use std::sync::Arc; use std::time::Instant; use tonic::{Request, Response, Status}; use tracing::debug; @@ -1230,37 +1231,40 @@ impl NodeService { // The target owns this read guard. It must span the complete // disk rename, not merely the preflight, so a movement transition // cannot restart after validation and before rename linearization. - let _scanner_publication_lease_guard = if let Some(token) = scanner_publication_lease_token { - let Some(store) = self.resolve_object_store() else { - return Ok(Response::new(RenameDataResponse { - success: false, - rename_data_resp: String::new(), - rename_data_resp_bin: Vec::new().into(), - error: Some(DiskError::other("scanner publication lease owner is unavailable").into()), - })); - }; - match store.acquire_scanner_publication_lease_guard(token).await { - Ok(guard) => Some(guard), - Err(err) => { + let scanner_publication_lease_guard: Option> = + if let Some(token) = scanner_publication_lease_token { + let Some(store) = self.resolve_object_store() else { return Ok(Response::new(RenameDataResponse { success: false, rename_data_resp: String::new(), rename_data_resp_bin: Vec::new().into(), - error: Some(DiskError::other(err.to_string()).into()), + error: Some(DiskError::other("scanner publication lease owner is unavailable").into()), })); + }; + match store.acquire_scanner_publication_lease_guard(token).await { + Ok(guard) => Some(Arc::new(guard)), + Err(err) => { + return Ok(Response::new(RenameDataResponse { + success: false, + rename_data_resp: String::new(), + rename_data_resp_bin: Vec::new().into(), + error: Some(DiskError::other(err.to_string()).into()), + })); + } } - } - } else { - None - }; + } else { + None + }; let request_decoded_from_msgpack = decoded_file_info.from_msgpack; match disk - .rename_data( + .rename_data_borrowed_with_fence_and_guard( &request.src_volume, &request.src_path, &decoded_file_info.value, &request.dst_volume, &request.dst_path, + scanner_publication_lease_token, + scanner_publication_lease_guard, ) .await { @@ -1641,26 +1645,36 @@ impl NodeService { // The target-side guard spans the complete delete operation. A // lease expiry or movement transition cannot occur between this // validation and the disk delete linearization point. - let _scanner_publication_lease_guard = if let Some(token) = scanner_publication_lease_token { - let Some(store) = self.resolve_object_store() else { - return Ok(Response::new(DeleteResponse { - success: false, - error: Some(DiskError::other("scanner publication lease owner is unavailable").into()), - })); - }; - match store.acquire_scanner_publication_lease_guard(token).await { - Ok(guard) => Some(guard), - Err(err) => { + let scanner_publication_lease_guard: Option> = + if let Some(token) = scanner_publication_lease_token { + let Some(store) = self.resolve_object_store() else { return Ok(Response::new(DeleteResponse { success: false, - error: Some(DiskError::other(err.to_string()).into()), + error: Some(DiskError::other("scanner publication lease owner is unavailable").into()), })); + }; + match store.acquire_scanner_publication_lease_guard(token).await { + Ok(guard) => Some(Arc::new(guard)), + Err(err) => { + return Ok(Response::new(DeleteResponse { + success: false, + error: Some(DiskError::other(err.to_string()).into()), + })); + } } - } - } else { - None - }; - match disk.delete(&request.volume, &request.path, options).await { + } else { + None + }; + match disk + .delete_with_scanner_publication_lease_and_guard( + &request.volume, + &request.path, + options, + scanner_publication_lease_token, + scanner_publication_lease_guard, + ) + .await + { Ok(_) => Ok(Response::new(DeleteResponse { success: true, error: None, diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 2e778c0c6..81c0040ca 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -1301,14 +1301,6 @@ pub(crate) trait StorageDiskRpcExt { async fn list_volumes(&self) -> DiskResult>; async fn make_volume(&self, volume: &str) -> DiskResult<()>; async fn make_volumes(&self, volume: Vec<&str>) -> DiskResult<()>; - async fn rename_data( - &self, - src_volume: &str, - src_path: &str, - file_info: &rustfs_filemeta::FileInfo, - dst_volume: &str, - dst_path: &str, - ) -> DiskResult; async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: i32) -> DiskResult>; async fn read_file(&self, volume: &str, path: &str) -> DiskResult; async fn read_file_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> DiskResult; @@ -1452,17 +1444,6 @@ where ecstore_disk::DiskAPI::make_volumes(self, volume).await } - async fn rename_data( - &self, - src_volume: &str, - src_path: &str, - file_info: &rustfs_filemeta::FileInfo, - dst_volume: &str, - dst_path: &str, - ) -> DiskResult { - ecstore_disk::DiskAPI::rename_data(self, src_volume, src_path, file_info.clone(), dst_volume, dst_path).await - } - async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: i32) -> DiskResult> { ecstore_disk::DiskAPI::list_dir(self, origvolume, volume, dir_path, count).await }