From 727c241a34fd86d1737abc3a34c759ce377f5bc4 Mon Sep 17 00:00:00 2001 From: houseme Date: Sun, 30 Aug 2026 02:28:04 +0800 Subject: [PATCH] fix(storage): retain publication scope through fanout --- crates/ecstore/src/object_api/types.rs | 36 +++++++++++++++++++ .../src/set_disk/core/io_primitives.rs | 24 +++++++++++++ crates/ecstore/src/set_disk/ops/object.rs | 25 +++++++++++-- crates/ecstore/src/store/mod.rs | 31 ++++++++++++++++ 4 files changed, 114 insertions(+), 2 deletions(-) diff --git a/crates/ecstore/src/object_api/types.rs b/crates/ecstore/src/object_api/types.rs index b55dcdaad..0f4f787c3 100644 --- a/crates/ecstore/src/object_api/types.rs +++ b/crates/ecstore/src/object_api/types.rs @@ -433,6 +433,42 @@ 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; + }; + 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") diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index 03d29ff86..ea9c51a45 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)")] @@ -3995,6 +4005,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 +4363,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 +4395,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 +4417,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 +4450,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( diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index dae4a6dad..b4e349b39 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; @@ -2624,6 +2625,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; @@ -3400,7 +3405,10 @@ impl SetDisks { // its terminal state is known before the coordinator releases // remote leases. let commit_allows_early_ack = commit_object_lock_guard.is_some() && commit_scanner_publication_scope.is_none(); - let detach_commit_owner = commit_allows_early_ack || commit_bucket_lifecycle_guard.is_some() || quota_mutation_fence; + let 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; @@ -3514,6 +3522,13 @@ impl SetDisks { 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 { if let Some(scope) = commit_scanner_publication_scope.as_ref() { if scope.state() == crate::object_api::ScannerPublicationCommitState::Admitted { @@ -3556,7 +3571,8 @@ 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() { @@ -3880,6 +3896,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 }) diff --git a/crates/ecstore/src/store/mod.rs b/crates/ecstore/src/store/mod.rs index 06d827d4e..f6c089082 100644 --- a/crates/ecstore/src/store/mod.rs +++ b/crates/ecstore/src/store/mod.rs @@ -1557,6 +1557,37 @@ mod tests { 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()));