From ee39e4fccb19e9bfa735ef31cca8ac06ead8e936 Mon Sep 17 00:00:00 2001 From: houseme Date: Sun, 30 Aug 2026 10:39:07 +0800 Subject: [PATCH 1/4] 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 } From 0c1801244239a2060fd7ae9d15c56c44cbb8c099 Mon Sep 17 00:00:00 2001 From: cxymds Date: Sun, 30 Aug 2026 10:42:10 +0800 Subject: [PATCH 2/4] fix(admin): version remote target credential capabilities (#6876) --- crates/ecstore/src/api/mod.rs | 7 ++--- crates/ecstore/src/bucket/replication/mod.rs | 16 +++++------ .../replication_config_boundary.rs | 16 +++++------ crates/replication/src/config.rs | 12 +++++++-- crates/replication/src/lib.rs | 17 ++++++------ rustfs/src/admin/handlers/replication.rs | 27 ++++++++++++++----- rustfs/src/admin/handlers/system.rs | 20 ++++++++++---- rustfs/src/admin/storage_api.rs | 8 +++--- 8 files changed, 79 insertions(+), 44 deletions(-) diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index fec7e009e..791c0c56d 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -194,9 +194,10 @@ pub mod bucket { BucketReplicationResyncStatus, BucketReplicationStat, BucketReplicationStats, BucketStats, DeleteReplicationConfigSnapshot, DeletedObjectReplicationInfo, DurableMrfBacklog, DynReplicationPool, InQueueMetric, MrfOpKind, MrfReplicateEntry, MustReplicateOptions, ObjectOpts, OperatorRuleContract, - REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, - REPLICATE_INCOMING_DELETE, REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, - REPLICATION_WRITABLE_FIELDS, ReplicateDecision, ReplicateObjectInfo, ReplicationBatchAdmission, ReplicationConfig, + REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS, + REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, REPLICATE_INCOMING_DELETE, + REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, + ReplicateDecision, ReplicateObjectInfo, ReplicationBatchAdmission, ReplicationConfig, ReplicationConfigStructureError, ReplicationConfigurationExt, ReplicationDeleteScheduleInput, ReplicationDeleteStateSource, ReplicationHealQueueResult, ReplicationObjectBridge, ReplicationObjectIO, ReplicationOperation, ReplicationPoolTrait, ReplicationPriority, ReplicationQueueAdmission, ReplicationScannerBridge, diff --git a/crates/ecstore/src/bucket/replication/mod.rs b/crates/ecstore/src/bucket/replication/mod.rs index 091b3a964..a60dd018d 100644 --- a/crates/ecstore/src/bucket/replication/mod.rs +++ b/crates/ecstore/src/bucket/replication/mod.rs @@ -44,14 +44,14 @@ mod replication_versioning_boundary; mod runtime_boundary; pub use replication_config_boundary::{ - ObjectOpts, OperatorRuleContract, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, - REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, - REPLICATION_WRITABLE_FIELDS, ReplicationConfigStructureError, ReplicationConfigurationExt, ReplicationTargetValidationError, - assign_site_replication_rule_priorities, invalid_replication_config_status_field, is_site_replication_role, - is_site_replication_rule, merge_incoming_replication_config, merge_user_replication_config, - replication_target_arn_deployment_id, replication_target_arns, should_remove_replication_target, - site_replication_rule_deployment_id, unsupported_replication_config_field, validate_replication_config_structure, - validate_replication_config_target_arns, + ObjectOpts, OperatorRuleContract, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS, + REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION, + REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, ReplicationConfigStructureError, + ReplicationConfigurationExt, ReplicationTargetValidationError, assign_site_replication_rule_priorities, + invalid_replication_config_status_field, is_site_replication_role, is_site_replication_rule, + merge_incoming_replication_config, merge_user_replication_config, replication_target_arn_deployment_id, + replication_target_arns, should_remove_replication_target, site_replication_rule_deployment_id, + unsupported_replication_config_field, validate_replication_config_structure, validate_replication_config_target_arns, }; pub(crate) use replication_filemeta_boundary::version_purge_statuses_map; pub use replication_filemeta_boundary::{ diff --git a/crates/ecstore/src/bucket/replication/replication_config_boundary.rs b/crates/ecstore/src/bucket/replication/replication_config_boundary.rs index fbd9e88b1..1eb8c3ad3 100644 --- a/crates/ecstore/src/bucket/replication/replication_config_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_config_boundary.rs @@ -13,12 +13,12 @@ // limitations under the License. pub use rustfs_replication::{ - ObjectOpts, OperatorRuleContract, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, - REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, - REPLICATION_WRITABLE_FIELDS, ReplicationConfigStructureError, ReplicationConfigurationExt, ReplicationRuleExt, - ReplicationTargetValidationError, assign_site_replication_rule_priorities, invalid_replication_config_status_field, - is_site_replication_role, is_site_replication_rule, merge_incoming_replication_config, merge_user_replication_config, - replication_target_arn_deployment_id, replication_target_arns, should_remove_replication_target, - site_replication_rule_deployment_id, unsupported_replication_config_field, validate_replication_config_structure, - validate_replication_config_target_arns, + ObjectOpts, OperatorRuleContract, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS, + REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION, + REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, ReplicationConfigStructureError, + ReplicationConfigurationExt, ReplicationRuleExt, ReplicationTargetValidationError, assign_site_replication_rule_priorities, + invalid_replication_config_status_field, is_site_replication_role, is_site_replication_rule, + merge_incoming_replication_config, merge_user_replication_config, replication_target_arn_deployment_id, + replication_target_arns, should_remove_replication_target, site_replication_rule_deployment_id, + unsupported_replication_config_field, validate_replication_config_structure, validate_replication_config_target_arns, }; diff --git a/crates/replication/src/config.rs b/crates/replication/src/config.rs index c85a92d44..abdb6cab4 100644 --- a/crates/replication/src/config.rs +++ b/crates/replication/src/config.rs @@ -60,8 +60,9 @@ pub const REPLICATION_READ_ONLY_HISTORICAL_FIELDS: &[&str] = &[ "Destination.ReplicationTime", ]; -// v3: remote targets accept temporary credential session tokens and expiry. -pub const REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION: u32 = 3; +// v4: temporary-credential fields moved from read-only historical metadata to +// writable fields because remote targets now use them for request signing. +pub const REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION: u32 = 4; pub const REMOTE_TARGET_WRITABLE_FIELDS: &[&str] = &[ "sourcebucket", @@ -91,6 +92,13 @@ pub const REMOTE_TARGET_WRITABLE_FIELDS: &[&str] = &[ "disableProxy", ]; +/// Remote target fields that are readable for persisted-data compatibility but +/// cannot be written through the admin API. +/// +/// The empty slice remains public for source compatibility with consumers of +/// the v3 capability API. +pub const REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS: &[&str] = &[]; + pub const REMOTE_TARGET_UNSUPPORTED_FIELDS: &[&str] = &["edge", "edgeSyncBeforeExpiry"]; #[derive(Debug, Clone, Serialize, Deserialize, Default)] diff --git a/crates/replication/src/lib.rs b/crates/replication/src/lib.rs index 4e32b8a3e..e9ab2801d 100644 --- a/crates/replication/src/lib.rs +++ b/crates/replication/src/lib.rs @@ -29,14 +29,15 @@ mod storage_api; pub mod tagging; pub use config::{ - ObjectOpts, OperatorRuleContract, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, - REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, - REPLICATION_WRITABLE_FIELDS, ReplicationConfigStructureError, ReplicationConfigurationExt, ReplicationTargetValidationError, - active_replication_rule_destination_arns, assign_site_replication_rule_priorities, invalid_replication_config_status_field, - is_reconciler_owned_site_replication_rule, is_site_replication_role, is_site_replication_rule, - merge_incoming_replication_config, merge_user_replication_config, replication_target_arn_deployment_id, - replication_target_arns, should_remove_replication_target, site_replication_rule_deployment_id, - unsupported_replication_config_field, validate_replication_config_structure, validate_replication_config_target_arns, + ObjectOpts, OperatorRuleContract, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS, + REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION, + REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, ReplicationConfigStructureError, + ReplicationConfigurationExt, ReplicationTargetValidationError, active_replication_rule_destination_arns, + assign_site_replication_rule_priorities, invalid_replication_config_status_field, is_reconciler_owned_site_replication_rule, + is_site_replication_role, is_site_replication_rule, merge_incoming_replication_config, merge_user_replication_config, + replication_target_arn_deployment_id, replication_target_arns, should_remove_replication_target, + site_replication_rule_deployment_id, unsupported_replication_config_field, validate_replication_config_structure, + validate_replication_config_target_arns, }; pub use delete::{ DeletedObjectReplicationInfo, delete_marker_purge_mrf_entry, delete_marker_purge_version_id, diff --git a/rustfs/src/admin/handlers/replication.rs b/rustfs/src/admin/handlers/replication.rs index 957a92816..b2b2b80dc 100644 --- a/rustfs/src/admin/handlers/replication.rs +++ b/rustfs/src/admin/handlers/replication.rs @@ -23,9 +23,9 @@ use crate::admin::storage_api::bucket::metadata::BUCKET_TARGETS_FILE; use crate::admin::storage_api::bucket::metadata_sys; use crate::admin::storage_api::bucket::metadata_sys::get_replication_config; use crate::admin::storage_api::bucket::replication::REMOTE_TARGET_UNSUPPORTED_FIELDS; -#[cfg(test)] -use crate::admin::storage_api::bucket::replication::REMOTE_TARGET_WRITABLE_FIELDS; use crate::admin::storage_api::bucket::replication::{BucketStats, ReplicationStatusType}; +#[cfg(test)] +use crate::admin::storage_api::bucket::replication::{REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS}; use crate::admin::storage_api::bucket::target::{ BucketTarget, BucketTargetType, Credentials as TargetCredentials, LatencyStat, duration_from_secs_or_nanos, }; @@ -1480,10 +1480,10 @@ impl Operation for ReplicationMrfHandler { #[cfg(test)] mod tests { use super::{ - REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, RemoteTargetCredentialsRequest, RemoteTargetRequest, - ReplicationDiffEntry, SUPPORTED_REMOTE_TARGET_API, TargetUpdateOp, build_mrf_response, extract_query_params, - parse_remote_target_update_ops, render_mrf_backlog, render_replication_diff, unique_replication_peers, - validate_remote_target_tls_settings, + REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS, REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, + RemoteTargetCredentialsRequest, RemoteTargetRequest, ReplicationDiffEntry, SUPPORTED_REMOTE_TARGET_API, TargetUpdateOp, + build_mrf_response, extract_query_params, parse_remote_target_update_ops, render_mrf_backlog, render_replication_diff, + unique_replication_peers, validate_remote_target_tls_settings, }; use crate::admin::storage_api::bucket::target::{BucketTarget, Credentials as TargetCredentials, LatencyStat}; use crate::admin::storage_api::replication::{BucketStats, DurableMrfBacklog, MrfOpKind, MrfReplicateEntry}; @@ -2557,6 +2557,21 @@ mod tests { #[test] fn remote_target_capability_fields_do_not_overlap() { + assert!( + REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS.is_empty(), + "v4 must not retain writable temporary-credential fields as historical-only" + ); + for field in REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS { + assert!( + !REMOTE_TARGET_WRITABLE_FIELDS.contains(field), + "remote target field {field} cannot be both historical-only and writable" + ); + assert!( + !REMOTE_TARGET_UNSUPPORTED_FIELDS.contains(field), + "remote target field {field} cannot be both historical-only and unsupported" + ); + } + for field in REMOTE_TARGET_UNSUPPORTED_FIELDS { assert!( !REMOTE_TARGET_WRITABLE_FIELDS.contains(field), diff --git a/rustfs/src/admin/handlers/system.rs b/rustfs/src/admin/handlers/system.rs index 82df34f05..062159287 100644 --- a/rustfs/src/admin/handlers/system.rs +++ b/rustfs/src/admin/handlers/system.rs @@ -24,8 +24,9 @@ use crate::admin::runtime_sources::{ DefaultAdminUsecase, QueryServerInfoRequest, current_endpoints_handle, default_admin_usecase, object_store_from_req, }; use crate::admin::storage_api::bucket::replication::{ - REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, - REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, + REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS, REMOTE_TARGET_UNSUPPORTED_FIELDS, + REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, + REPLICATION_WRITABLE_FIELDS, }; use crate::admin::storage_api::cluster::{ CapabilityState, CapabilityStatus, ObservabilitySnapshotProvider, TopologySnapshot, TopologySnapshotProvider, @@ -729,6 +730,15 @@ impl ReplicationCapabilities { name, state: ReplicationFieldState::Supported, }) + .chain( + REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS + .iter() + .copied() + .map(|name| ReplicationFieldCapability { + name, + state: ReplicationFieldState::ReadOnlyHistorical, + }), + ) .chain( REMOTE_TARGET_UNSUPPORTED_FIELDS .iter() @@ -1296,8 +1306,8 @@ mod tests { assert_eq!(response.summary.manual_transition_jobs.state, CapabilityState::Supported); assert_eq!(response.replication.contract_version, 1); assert_eq!(response.replication.bucket_replication.contract_version, 1); - // v3: temporary-credential fields are writable and used for signing. - assert_eq!(response.replication.remote_targets.contract_version, 3); + // v4: temporary-credential fields moved from historical-only to writable. + assert_eq!(response.replication.remote_targets.contract_version, 4); assert_eq!(response.replication.bucket_replication.status.state, CapabilityState::Supported); assert_eq!(response.replication.remote_targets.status.state, CapabilityState::Supported); assert_eq!( @@ -1418,7 +1428,7 @@ mod tests { assert_eq!(value["summary"]["manual_transition_jobs"]["state"], "supported"); assert_eq!(value["replication"]["contract_version"], 1); assert_eq!(value["replication"]["bucket_replication"]["contract_version"], 1); - assert_eq!(value["replication"]["remote_targets"]["contract_version"], 3); + assert_eq!(value["replication"]["remote_targets"]["contract_version"], 4); assert_eq!(value["replication"]["bucket_replication"]["status"]["state"], "supported"); assert_eq!(value["replication"]["remote_targets"]["status"]["state"], "supported"); assert_eq!( diff --git a/rustfs/src/admin/storage_api.rs b/rustfs/src/admin/storage_api.rs index 1c8a14c26..5ea5b3f74 100644 --- a/rustfs/src/admin/storage_api.rs +++ b/rustfs/src/admin/storage_api.rs @@ -443,10 +443,10 @@ pub(crate) mod quota { pub(crate) mod replication { pub(crate) use super::ecstore_bucket::replication::{ - OperatorRuleContract, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, - REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, - REPLICATION_WRITABLE_FIELDS, assign_site_replication_rule_priorities, merge_incoming_replication_config, - replication_target_arn_deployment_id, + OperatorRuleContract, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS, + REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION, + REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, assign_site_replication_rule_priorities, + merge_incoming_replication_config, replication_target_arn_deployment_id, }; pub(crate) type BucketReplicationResyncStatus = super::ecstore_bucket::replication::BucketReplicationResyncStatus; pub(crate) type BucketStats = super::ecstore_bucket::replication::BucketStats; From b2a2e637a5bf3d71eda83736e096a856f8f571bb Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Sun, 30 Aug 2026 10:42:14 +0800 Subject: [PATCH 3/4] fix(ci): refresh Linux full E2E selection (#6875) --- .config/e2e-full-selection.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.config/e2e-full-selection.txt b/.config/e2e-full-selection.txt index 51af8c732..e7b2b9be0 100644 --- a/.config/e2e-full-selection.txt +++ b/.config/e2e-full-selection.txt @@ -1,2 +1,2 @@ sha256-darwin=d6aa36cfaae2c4d8590482c7e47138c5965b335b34a75f50d11ffc3366e9021e -sha256-linux=96db8060fce98addda4f69092d297ca236bec4892d820617a26a261eedac61b0 +sha256-linux=e3eb4ab7fc72224abf58c546ac0706d6605d3bd26bac7d8ce338829fd3daecc2 From cf362282f019ac70272bf88a58b861c0b0e7f9c4 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Sun, 30 Aug 2026 10:42:23 +0800 Subject: [PATCH 4/4] fix(test): serialize transition matrix tests under nextest (#6874) The transition_matrix_tests use #[serial_test::serial] which has no effect under nextest (each test runs in a separate process). When running alongside thousands of other ecstore tests, the shared metadata cache generation counter can race, causing intermittent 'metadata read should publish the generation under test' panics. Add both tests to the ecstore-serial-flaky test group in both default and ci nextest profiles so they run single-threaded. --- .config/nextest.toml | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/.config/nextest.toml b/.config/nextest.toml index 9907f4f4f..044a3ff47 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -100,6 +100,16 @@ test-group = 'embedded-test-ports' filter = 'package(rustfs-ecstore) & test(manual_transition_page_checkpoint_persists_durable_job_progress)' test-group = 'ecstore-serial-flaky' +# Serialize the transition matrix tests. They build a 4-disk hermetic erasure +# set, populate the get_object_metadata_cache, and assert generation lifecycle +# semantics. serial_test's #[serial] has no effect across nextest's process +# boundary, so concurrent execution races the shared metadata-cache generation +# counter and causes spurious "metadata read should publish the generation" +# panics. Preventive serialization, no retries. +[[profile.default.overrides]] +filter = 'package(rustfs-ecstore) & test(set_disk::transition_matrix_tests::)' +test-group = 'ecstore-serial-flaky' + # The durable ILM decommission regressions build isolated multi-pool stores and # deliberately take source or target disks offline while checking fencing. [[profile.default.overrides]] @@ -232,6 +242,12 @@ test-group = 'embedded-test-ports' filter = 'package(rustfs-ecstore) & test(manual_transition_page_checkpoint_persists_durable_job_progress)' test-group = 'ecstore-serial-flaky' +# Serialize the transition matrix tests under the ci profile too (see the +# matching default-profile override near the top). No retries. +[[profile.ci.overrides]] +filter = 'package(rustfs-ecstore) & test(set_disk::transition_matrix_tests::)' +test-group = 'ecstore-serial-flaky' + [[profile.ci.overrides]] filter = 'package(rustfs-ecstore) & (test(decommission_migrates_and_verifies_registered_durable_ilm_records) | test(decommission_durable_ilm_target_read_error_is_not_masked_by_peer_success) | test(decommission_durable_ilm_terminal_receipt_recovers_failed_source_cleanup) | test(decommission_durable_ilm_receipt_pagination_fails_closed_on_second_page) | test(decommission_durable_ilm_recovery_keeps_multiple_active_sources))' test-group = 'ecstore-serial-flaky'