fix(scanner): own publication mutations through storage drain (#6867)

* fix(scanner): own publication mutations through storage drain

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(storage): remove unused rename data shim

Co-Authored-By: heihutu <heihutu@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>

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,

crates/ecstore/src/disk/disk_store.rs

@@ -250,6 +250,40 @@ pub(crate) trait DiskStoreRenameDataExt {
         dst_volume: &str,
         dst_path: &str,
     ) -> Result<RenameDataResp>;
+
+    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<Arc<dyn Send + Sync>>,
+    ) -> Result<RenameDataResp> {
+        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<T, F, Fut>(external_guard: Option<Arc<dyn Send + Sync>>, operation: F) -> Result<T>
+where
+    T: Send + 'static,
+    F: FnOnce() -> Fut + Send + 'static,
+    Fut: std::future::Future<Output = Result<T>> + 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<Arc<dyn Send + Sync>>,
+    ) -> Result<RenameDataResp> {
+        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<Arc<dyn Send + Sync>>,
+    ) -> 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<LocalDisk>,
         health_check: bool,
@@ -2263,6 +2371,44 @@ mod tests {
     };
     use tokio::io::AsyncWrite;
 
+    struct DropProbe(Arc<std::sync::atomic::AtomicUsize>);
+
+    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<dyn Send + Sync> = 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]

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<Uuid>,
+        external_guard: Option<Arc<dyn Send + Sync>>,
     ) -> 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<Uuid>,
+    ) -> Result<RenameDataResp> {
+        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<Uuid>,
+        external_guard: Option<Arc<dyn Send + Sync>>,
     ) -> Result<RenameDataResp> {
         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) => {

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<Option<OwnedRwLockReadGuard<()>>>,
+    lease_release_safe: Arc<AtomicBool>,
+}
+
+/// 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<ScannerPublicationCommitScopeInner>,
+}
+
+/// 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<ScannerPublicationCommitScope>,
+}
+
+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<Uuid>,
+        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<Uuid>,
+        movement_permit: OwnedRwLockReadGuard<()>,
+        lease_release_safe: Arc<AtomicBool>,
+    ) -> 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<tokio_util::sync::CancellationToken>,
 
+    /// 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<ScannerPublicationCommitScope>,
+
     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)

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<String, Uuid>>,
+    scanner_publication_commit_scope: Option<crate::object_api::ScannerPublicationCommitScope>,
 }
 
 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<crate::object_api::ScannerPublicationCommitScope>,
+    ) -> 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<F, Fut>(
+    scope: Option<crate::object_api::ScannerPublicationCommitScope>,
+    operation: F,
+) -> disk::error::Result<()>
+where
+    F: FnOnce() -> Fut + Send + 'static,
+    Fut: Future<Output = disk::error::Result<()>> + 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<String, Uuid>>,
+        scanner_publication_commit_scope: Option<crate::object_api::ScannerPublicationCommitScope>,
     ) -> 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<dyn Send + Sync>);
+                    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 {

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<Uuid>,
+    ) -> Option<crate::object_api::ScannerPublicationCommitScope> {
+        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<Uuid>,
+        lease_release_safe: Arc<std::sync::atomic::AtomicBool>,
+    ) -> Option<crate::object_api::ScannerPublicationCommitScope> {
+        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) {

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<String, String>) -> Result<Option<HashMap<String, Uuid>>> {
     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<OldCurrentSize>)> {
         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<ObjectInfo> {
+        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)
     }
 

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<Uuid>,
+    ) -> Option<ScannerPublicationCommitScope> {
+        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<Uuid>,
+        lease_release_safe: Arc<std::sync::atomic::AtomicBool>,
+    ) -> Option<ScannerPublicationCommitScope> {
+        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()));