diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 1966538a0..ab709a7d7 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -416,9 +416,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/object_api/types.rs b/crates/ecstore/src/object_api/types.rs index be761a8c7..5f5691796 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::{AtomicU8, Ordering}; +use tokio::sync::{Mutex, Notify, OwnedRwLockReadGuard}; +use tokio_util::sync::CancellationToken; #[derive(Clone)] pub struct NamespaceLockFence { @@ -347,6 +350,251 @@ 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, + /// 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>>, +} + +/// 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, +} + +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 { + 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(), + movement_permit: Mutex::new(Some(movement_permit)), + }), + } + } + + 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() + } + + 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 { + for expected in [SCANNER_PUBLICATION_SCOPE_ADMITTED, SCANNER_PUBLICATION_SCOPE_IN_FLIGHT] { + if self + .inner + .state + .compare_exchange( + expected, + SCANNER_PUBLICATION_SCOPE_ABORTED_BEFORE_COMMIT, + Ordering::AcqRel, + Ordering::Acquire, + ) + .is_ok() + { + self.inner.completed.notify_waiters(); + return true; + } + } + 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(|| 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() + } +} + #[derive(Default, Clone)] pub struct ObjectOptions { // Use the maximum parity (N/2), used when saving server configuration files @@ -384,6 +632,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 +726,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/store/mod.rs b/crates/ecstore/src/store/mod.rs index 0c874690f..c401c8636 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,28 @@ 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, + )) + } + /// 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. @@ -1422,6 +1444,71 @@ 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_target_guard_keeps_movement_writer_fenced_after_lease_release() { let store = build_store_with_ctx(Arc::new(InstanceContext::new()));