fix(scanner): retain cleanup scope after cancellation

Co-Authored-By: heihutu <[email protected]>

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,

crates/ecstore/src/cluster/rpc/peer_rest_client.rs

@@ -725,7 +725,7 @@ impl PeerRestClient {
     /// never take it offline no matter what its message says. The substring
     /// fallback only covers failures that exist purely as text, such as the
     /// dial errors `get_client` wraps.
-    fn is_network_like_error(err: &Error) -> bool {
+    pub(crate) fn is_network_like_error(err: &Error) -> bool {
         if let Error::Io(io_err) = err
             && let Some(status) = embedded_tonic_status(io_err)
         {

crates/ecstore/src/disk/disk_store.rs

@@ -250,6 +250,37 @@ 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,
+{
+    tokio::spawn(async move {
+        let _external_guard = external_guard;
+        operation().await
+    })
+    .await
+    .map_err(|err| Error::other(format!("owned mutation task failed: {err}")))?
 }
 
 impl DiskStoreRenameDataExt for LocalDiskWrapper {
@@ -273,6 +304,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 {
@@ -1097,6 +1171,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,
@@ -2247,6 +2352,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

@@ -694,6 +694,28 @@ impl Disk {
         }
     }
 
+    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_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)
+                    .await
+            }
+        }
+    }
+
     pub(crate) async fn rename_data_borrowed(
         &self,
         src_volume: &str,
@@ -714,11 +736,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::{AtomicU8, Ordering};
+use tokio::sync::{Mutex, Notify, OwnedRwLockReadGuard};
+use tokio_util::sync::CancellationToken;
 
 #[derive(Clone)]
 pub struct NamespaceLockFence {
@@ -347,6 +350,320 @@ 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<Option<OwnedRwLockReadGuard<()>>>,
+    lease_release_safe: Arc<std::sync::atomic::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;
+        };
+        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(std::sync::atomic::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<std::sync::atomic::AtomicBool>,
+    ) -> Self {
+        // Admission itself is not a safe release point. The flag becomes true
+        // only after the storage mutation owner reports a terminal state.
+        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(),
+                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()
+    }
+
+    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 +701,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 +795,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/services/notification_sys.rs

@@ -1483,7 +1483,7 @@ impl NotificationSys {
             futures.push(async move {
                 let client = client.ok_or_else(|| Error::other(format!("scanner activity peer[{idx}] is unreachable")))?;
                 let host = client.grid_host.clone();
-                scanner_activity_with_timeout(SCANNER_ACTIVITY_PROBE_TIMEOUT, &host, client.scanner_activity())
+                scanner_activity_with_retry(&client, &host)
                     .await
                     .map(|activity| (host, activity))
             });
@@ -1962,6 +1962,44 @@ where
         .map_err(|_| Error::other(format!("scanner activity peer {host} timed out after {timeout_duration:?}")))?
 }
 
+fn scanner_activity_should_retry(first_error: Option<&Error>, timed_out: bool) -> bool {
+    timed_out || first_error.is_some_and(PeerRestClient::is_network_like_error)
+}
+
+/// Retry one activity probe after a bounded reconnect when the first attempt
+/// failed at the transport boundary.  A peer that answered with an invalid or
+/// incompatible activity response is not retried here: it must remain a hard
+/// fail-closed result for the all-peer publication proof.
+async fn scanner_activity_with_retry(client: &PeerRestClient, host: &str) -> Result<ScannerPeerActivity> {
+    let first = timeout(SCANNER_ACTIVITY_PROBE_TIMEOUT, client.scanner_activity()).await;
+    let should_retry = match &first {
+        Ok(Ok(_)) => false,
+        Ok(Err(err)) => scanner_activity_should_retry(Some(err), false),
+        Err(_) => scanner_activity_should_retry(None, true),
+    };
+
+    match first {
+        Ok(Ok(activity)) => return Ok(activity),
+        Ok(Err(err)) if !should_retry => return Err(err),
+        Ok(Err(err)) => {
+            debug!(peer = host, error = %err, "scanner activity probe failed on first transport attempt; reconnecting");
+            client.prepare_retry().await;
+        }
+        Err(_) => {
+            debug!(peer = host, timeout = ?SCANNER_ACTIVITY_PROBE_TIMEOUT, "scanner activity probe timed out on first attempt; reconnecting");
+            client.prepare_retry().await;
+        }
+    }
+
+    match timeout(SCANNER_ACTIVITY_PROBE_TIMEOUT, client.scanner_activity()).await {
+        Ok(result) => result,
+        Err(_) => {
+            client.evict_connection().await;
+            Err(Error::Timeout)
+        }
+    }
+}
+
 #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 async fn call_peer_with_timeout<F, Fut>(
     timeout_dur: Duration,
@@ -2882,6 +2920,20 @@ mod tests {
         assert!(err.to_string().contains("peer-1"));
     }
 
+    #[test]
+    fn scanner_activity_retry_only_reconnects_transport_failures() {
+        assert!(scanner_activity_should_retry(None, true));
+        assert!(scanner_activity_should_retry(Some(&Error::other("connection refused")), false));
+        assert!(!scanner_activity_should_retry(
+            Some(&Error::other("peer returned an invalid scanner activity response proof")),
+            false
+        ));
+        assert!(!scanner_activity_should_retry(
+            Some(&Error::from(tonic::Status::internal("peer rejected activity"))),
+            false
+        ));
+    }
+
     #[tokio::test]
     async fn scanner_dirty_usage_acknowledgement_rejects_missing_and_duplicate_targets() {
         let sys = NotificationSys {

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)")]
@@ -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(

crates/ecstore/src/set_disk/mod.rs

@@ -105,7 +105,9 @@ use crate::{
         SnapshotLeaseToken, UpdateMetadataOpts, endpoint::Endpoint, error::DiskError, format::FormatV3,
     },
     error::{StorageError, to_object_err},
-    object_api::{GetObjectReader, NamespaceLockFence, ObjectInfo, ObjectLockConfigSnapshot, PutObjReader},
+    object_api::{
+        GetObjectReader, NamespaceLockFence, ObjectInfo, ObjectLockConfigSnapshot, PutObjReader, ScannerPublicationCommitScope,
+    },
     // event::name::EventName,
     services::event_notification::{EventArgs, send_event},
     store::init_format::{
@@ -3937,6 +3939,47 @@ impl SetDisks {
         owner.scanner_data_usage_publication_admission_guard().await
     }
 
+    /// Acquire a storage-owned scanner publication scope for this set's
+    /// instance movement fence. The scope keeps the read permit alive across
+    /// scanner future cancellation until the mutation owner 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,
+        ))
+    }
+
     /// 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;
@@ -269,6 +270,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(|err| Error::other(format!("scanner publication delete scope cannot start: {err:?}")))?;
+    }
+    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);
@@ -2624,6 +2641,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;
 
@@ -3394,8 +3415,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;
@@ -3491,7 +3520,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()) => {
@@ -3502,7 +3531,28 @@ 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 {
+                    if let Some(scope) = commit_scanner_publication_scope.as_ref() {
+                        if scope.state() == crate::object_api::ScannerPublicationCommitState::Admitted {
+                            let _ = scope.mark_aborted_before_commit();
+                        } else {
+                            let _ = scope.mark_indeterminate();
+                        }
+                    }
                     SetDisks::abort_quota_reservation_after_fence(
                         quota_reservation,
                         &commit_disks,
@@ -3537,9 +3587,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;
@@ -3854,6 +3912,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 })
@@ -7051,6 +7114,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.
@@ -7145,6 +7213,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;
@@ -7159,6 +7228,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 {
@@ -7170,10 +7240,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());
             }
@@ -7181,10 +7255,14 @@ 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)?;
+            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())
                 .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());
         }
@@ -7257,10 +7335,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));
         }
 
@@ -7334,6 +7416,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]))?;
@@ -7345,6 +7428,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);
         }
 
@@ -7370,6 +7456,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]))?;
@@ -7395,6 +7482,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,52 @@ 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,
+        ))
+    }
+
+    /// Variant used by the scanner supervisor to observe whether a scope was
+    /// dropped before reaching a safe terminal state. The flag is in-memory
+    /// only and lets the supervisor avoid releasing remote leases on an
+    /// indeterminate cancellation path.
+    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 +1455,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 +1488,106 @@ 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_aborted_before_commit(),
+            "an in-flight mutation cannot claim pre-commit abort without storage proof"
+        );
+        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()));

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,48 @@ pub(crate) fn scanner_publication_epoch_changed(error: &EcstoreError) -> bool {
     )
 }
 
+pub(crate) async fn delete_config_with_publication_scope_for_epoch<S>(
+    api: Arc<S>,
+    bucket: &str,
+    object: &str,
+    mut opts: ScannerObjectOptions,
+    expected_epoch: u64,
+    scanner_publication_commit_scope: Option<ScannerPublicationCommitScope>,
+) -> EcstoreResult<ScannerObjectInfo>
+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
+    };
+    let result = if let Some(scope) = scanner_publication_commit_scope {
+        // Keep the storage mutation and its publication scope alive after the
+        // scanner persistence waiter is cancelled. The delete path can fan
+        // out to remote RPCs or blocking local namespace syscalls; dropping
+        // only the waiter must not release the movement permit early.
+        opts.scanner_publication_commit_scope = Some(scope.clone());
+        let bucket = bucket.to_owned();
+        let object = object.to_owned();
+        tokio::spawn(async move {
+            let _publication_scope_owner = scope;
+            api.delete_config_object(&bucket, &object, opts).await
+        })
+        .await
+        .map_err(|err| EcstoreError::other(format!("scanner publication delete owner failed: {err}")))?
+    } else {
+        opts.scanner_publication_commit_scope = None;
+        api.delete_config_object(bucket, object, opts).await
+    };
+    drop(legacy_admission);
+    result
+}
+
 pub(crate) async fn delete_config_with_publication_admission_for_epoch<S>(
     api: Arc<S>,
     bucket: &str,
@@ -762,10 +805,7 @@ pub(crate) async fn delete_config_with_publication_admission_for_epoch<S>(
 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 +836,14 @@ where
     Some(admission)
 }
 
-pub(crate) async fn save_config_shared_with_preconditions_and_lease_fence<S>(
+pub(crate) async fn save_config_shared_with_preconditions_and_lease_fence_and_scope<S>(
     api: Arc<S>,
     file: &str,
     data: Bytes,
     sha256hex: Option<String>,
     preconditions: HTTPPreconditions,
     scanner_publication_lease_fence: Option<&str>,
+    scanner_publication_commit_scope: Option<ScannerPublicationCommitScope>,
 ) -> EcstoreResult<ScannerObjectInfo>
 where
     S: ScannerObjectIO,
@@ -822,6 +863,7 @@ where
         &ScannerObjectOptions {
             max_parity: true,
             http_preconditions: Some(preconditions),
+            scanner_publication_commit_scope,
             user_defined,
             ..Default::default()
         },
@@ -886,6 +928,27 @@ pub trait ScannerConfigObjectDelete: Send + Sync + std::fmt::Debug + 'static {
     async fn scanner_data_usage_publication_admission(&self) -> Option<ScannerDataUsagePublicationAdmission> {
         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<Uuid>,
+    ) -> Option<ScannerPublicationCommitScope> {
+        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<Uuid>,
+        _lease_release_safe: Arc<std::sync::atomic::AtomicBool>,
+    ) -> Option<ScannerPublicationCommitScope> {
+        None
+    }
 }
 
 pub struct ScannerDataUsagePublicationAdmission {
@@ -929,6 +992,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<Uuid>,
+    ) -> Option<ScannerPublicationCommitScope> {
+        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<Uuid>,
+        lease_release_safe: Arc<std::sync::atomic::AtomicBool>,
+    ) -> Option<ScannerPublicationCommitScope> {
+        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 +1035,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<Uuid>,
+    ) -> Option<ScannerPublicationCommitScope> {
+        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<Uuid>,
+        lease_release_safe: Arc<std::sync::atomic::AtomicBool>,
+    ) -> Option<ScannerPublicationCommitScope> {
+        self.scanner_data_usage_publication_commit_scope_with_release_flag(
+            expected_movement_epoch,
+            safe_deadline,
+            remote_lease_tokens,
+            lease_release_safe,
+        )
+        .await
+    }
 }
 
 #[cfg(test)]

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,8 +63,8 @@ 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::owner::SCANNER_PUBLICATION_LEASE_TTL_MS;
 use crate::storage_api::scan::{
     BucketOperations, BucketOptions, NamespaceLocking as _, SCANNER_ACTIVITY_LEGACY_PROTOCOL_VERSION,
     SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION, SCANNER_ACTIVITY_PROTOCOL_VERSION,
@@ -72,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,
 };
@@ -454,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<impl ScannerObjectIO + ScannerConfigObjectDelete>,
@@ -461,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<impl ScannerObjectIO + ScannerConfigObjectDelete>,
     expected_publication_epoch: Option<u64>,
     remote_lease_deadline: Option<std::time::Instant>,
     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<impl ScannerObjectIO + ScannerConfigObjectDelete>,
+    expected_publication_epoch: Option<u64>,
+    remote_lease_deadline: Option<std::time::Instant>,
+    scanner_publication_lease_fence: Option<&str>,
+    remote_lease_tokens: Vec<Uuid>,
+    lease_release_safe: Arc<AtomicBool>,
 ) -> Result<(), EcstoreError> {
     let backup_path = format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str());
     for retry in 0..=SCANNER_PERSIST_CAS_RETRIES {
@@ -531,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 {
@@ -1416,13 +1473,10 @@ async fn run_data_scanner_cycle_with_budget(
         mark_scan_cycle_idle(cycle_info, &mut cycle_metrics_guard).await;
         return ScannerCycleOutcome::Deferred(ScannerCycleDeferReason::DataMovement);
     };
-    let usage_persist_baseline_result = read_config_with_revision(storeapi.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str()).await;
+    let usage_persist_baseline_result = read_data_usage_persist_baseline(storeapi.clone()).await;
     drop(baseline_publication_guard);
     let usage_persist_baseline = match usage_persist_baseline_result {
-        Ok((data, revision)) => DataUsagePersistBaseline {
-            data: data.map(Bytes::from),
-            revision,
-        },
+        Ok(baseline) => baseline,
         Err(err) => {
             error!(
                 target: "rustfs::scanner",
@@ -1462,10 +1516,23 @@ async fn run_data_scanner_cycle_with_budget(
         {
             Some(ScannerCycleDeferReason::DataMovement)
         }
+        // A complete walk can still be retained as an observational snapshot
+        // when only the final activity proof was unavailable.  It must not
+        // block the observation receiver: the authoritative publication
+        // fence remains enforced by the usage store and the cycle is advanced
+        // as partial without acknowledging dirty usage.
+        Ok(result)
+            if result.has_observational_snapshot()
+                && matches!(
+                    result.status,
+                    ScannerCycleStatus::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable)
+                ) =>
+        {
+            None
+        }
         Ok(result) => final_data_usage_publication_defer_reason(storeapi.as_ref(), result.status).await,
         Err(_) => Some(ScannerCycleDeferReason::ActivityBaselineUnavailable),
     };
-    let publication_deferred = publication_defer_reason.is_some();
     let publication_epoch = scan_result.as_ref().ok().and_then(ScannerCycleResult::publication_epoch);
     let remote_publication_lease_targets = if publication_defer_reason.is_none() {
         scan_result
@@ -1479,11 +1546,6 @@ async fn run_data_scanner_cycle_with_budget(
     let mut remote_publication_leases = None;
     let remote_lease_defer_reason = if remote_publication_lease_targets.is_empty() {
         None
-    } else if usage_persist_timeout >= Duration::from_millis(SCANNER_PUBLICATION_LEASE_TTL_MS) {
-        // The lease is intentionally fixed-duration and has no renewal path.
-        // Refuse a persistence budget that could outlive it instead of
-        // allowing the peer to admit movement while a local PUT is in flight.
-        Some(ScannerCycleDeferReason::ActivityBaselineUnavailable)
     } else if let Some(notification_system) = storeapi.notification_system() {
         match notification_system
             .acquire_scanner_publication_leases(remote_publication_lease_targets.clone())
@@ -1534,21 +1596,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::ActivityBaselineUnavailable)
-        .or(publication_defer_reason);
+    // Include reasons discovered while acquiring or validating remote leases.
+    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);
@@ -1562,6 +1619,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,
@@ -1573,7 +1635,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();
@@ -1638,16 +1702,28 @@ 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;
-        if lease_expired || release_result.is_err() {
+        let lease_release_failed = release_result.is_err();
+        if lease_expired || lease_release_failed {
             // A lease that expired or could not be released is never treated
             // as a successful authoritative publication. The peer may have
             // admitted movement immediately after the lease ended.
             usage_persist_outcome = if usage_persist_outcome == DataUsagePersistOutcome::Failed {
                 DataUsagePersistOutcome::Failed
+            } else if lease_release_failed {
+                DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::PublicationLeaseReleaseFailed)
             } else {
-                DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable)
+                DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded)
             };
         }
     }
@@ -2803,12 +2879,7 @@ fn finalize_scanner_cycle_result(
     scan_cycle_result: crate::scanner_io::ScannerCycleResult,
     usage_persist_outcome: DataUsagePersistOutcome,
 ) -> (ScannerCycleOutcome, bool, Vec<ScannerDirtyUsageAcknowledgement>) {
-    let completion_outcome = scanner_cycle_completion_outcome(
-        scan_cycle_result.status,
-        usage_persist_outcome,
-        scan_cycle_result.has_dirty_usage_to_acknowledge(),
-        scan_cycle_result.has_failed_dirty_usage(),
-    );
+    let completion_outcome = scanner_cycle_completion_outcome_for_result(&scan_cycle_result, usage_persist_outcome);
     let pending_maintenance_work = scan_cycle_result.has_pending_maintenance_work();
     let durable_complete_snapshot = scan_cycle_result.status == ScannerCycleStatus::Complete
         && matches!(
@@ -2823,6 +2894,34 @@ fn finalize_scanner_cycle_result(
     (completion_outcome, pending_maintenance_work, remote_dirty_usage_acknowledgements)
 }
 
+fn scanner_cycle_completion_outcome_for_result(
+    scan_cycle_result: &crate::scanner_io::ScannerCycleResult,
+    usage_persist_outcome: DataUsagePersistOutcome,
+) -> ScannerCycleOutcome {
+    let has_dirty_usage = scan_cycle_result.has_dirty_usage_to_acknowledge();
+    let has_failed_dirty_usage = scan_cycle_result.has_failed_dirty_usage();
+    if scan_cycle_result.has_observational_snapshot()
+        && matches!(
+            scan_cycle_result.status,
+            ScannerCycleStatus::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable)
+        )
+    {
+        return match usage_persist_outcome {
+            DataUsagePersistOutcome::Saved
+            | DataUsagePersistOutcome::AlreadyDurable
+            | DataUsagePersistOutcome::PriorCycleDurable
+            | DataUsagePersistOutcome::Current
+                if !has_failed_dirty_usage =>
+            {
+                ScannerCycleOutcome::Partial
+            }
+            DataUsagePersistOutcome::Deferred(reason) => ScannerCycleOutcome::Deferred(reason),
+            _ => ScannerCycleOutcome::Failed,
+        };
+    }
+    scanner_cycle_completion_outcome(scan_cycle_result.status, usage_persist_outcome, has_dirty_usage, has_failed_dirty_usage)
+}
+
 /// Decide whether an incoming usage snapshot must be skipped as stale, given the local
 /// wall clock `now`. Mirrors `stale_data_usage_persist_reason` in
 /// `crates/ecstore/src/data_usage/mod.rs` — keep the two consistent.

crates/scanner/src/scanner/cycle_state.rs

@@ -1327,6 +1327,13 @@ pub(super) async fn persisted_usage_floor_for_startup(
     let mut floor = PersistedUsageFloor::default();
     let mut found_any = false;
     let mut bootstrap_pending = false;
+    // A valid JSON object without a baseline identity is not a floor and must
+    // never be treated as an empty one.  It can, however, be a partially
+    // written v2 primary left behind during an upgrade.  Keep its epoch as a
+    // fence while looking for a durable companion snapshot; if no companion
+    // is new enough, the caller still fails closed below.
+    let mut invalid_baseline_path: Option<String> = None;
+    let mut invalid_baseline_epoch: Option<u64> = None;
     let update_floor = |floor: &mut PersistedUsageFloor, usage: &DataUsageInfo, path: &str| -> Result<(), ScannerError> {
         floor.leader_epoch = floor.leader_epoch.max(usage.scanner_epoch.unwrap_or_default());
         if let Some(completed_cycle) = usage.scanner_cycle {
@@ -1340,6 +1347,7 @@ pub(super) async fn persisted_usage_floor_for_startup(
     };
     for primary_path in [DATA_USAGE_OBJ_NAME_PATH.as_str(), LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()] {
         let backup_path = format!("{primary_path}.bkp");
+        let is_v2_path = primary_path == DATA_USAGE_OBJ_NAME_PATH.as_str();
         let primary_epoch = match read_config_with_revision(storeapi.clone(), primary_path).await {
             Ok((Some(data), _)) => {
                 let usage = serde_json::from_slice::<DataUsageInfo>(&data).map_err(|err| {
@@ -1353,13 +1361,20 @@ pub(super) async fn persisted_usage_floor_for_startup(
                     update_floor(&mut floor, &usage, primary_path)?;
                     None
                 } else if !data_usage_info_has_persisted_baseline_identity(&usage) {
-                    return Err(ScannerError::Other(format!(
-                        "scanner usage floor from {primary_path} has no persisted baseline identity"
-                    )));
+                    invalid_baseline_path.get_or_insert_with(|| primary_path.to_string());
+                    invalid_baseline_epoch = invalid_baseline_epoch.max(usage.scanner_epoch);
+                    None
                 } else {
                     let epoch = usage.scanner_epoch.unwrap_or_default();
-                    update_floor(&mut floor, &usage, primary_path)?;
-                    Some(epoch)
+                    // A legacy snapshot may be structurally valid but older
+                    // than an incomplete v2 snapshot left by a newer leader.
+                    // Do not let that candidate regress the startup floor.
+                    if !is_v2_path && invalid_baseline_epoch.is_some_and(|fenced_epoch| epoch < fenced_epoch) {
+                        None
+                    } else {
+                        update_floor(&mut floor, &usage, primary_path)?;
+                        Some(epoch)
+                    }
                 }
             }
             Ok((None, _)) => None,
@@ -1377,21 +1392,26 @@ pub(super) async fn persisted_usage_floor_for_startup(
                         "scanner usage bootstrap conflicts with a persisted backup".to_string(),
                     ));
                 }
-                any_found = true;
                 let usage = serde_json::from_slice::<DataUsageInfo>(&data).map_err(|err| {
                     ScannerError::Other(format!("failed to decode scanner usage floor from {backup_path}: {err}"))
                 })?;
                 if !data_usage_info_has_persisted_baseline_identity(&usage) {
-                    return Err(ScannerError::Other(format!(
-                        "scanner usage floor from {backup_path} has no persisted baseline identity"
-                    )));
-                }
-                let backup_epoch = usage.scanner_epoch.unwrap_or_default();
-                // A backup write from an older leader may complete after the
-                // primary epoch has been fenced. It must not advance the startup
-                // floor unless its epoch is at least as new as the primary.
-                if primary_epoch.is_none_or(|epoch| backup_epoch >= epoch) {
-                    update_floor(&mut floor, &usage, &backup_path)?;
+                    invalid_baseline_path.get_or_insert_with(|| backup_path.clone());
+                    invalid_baseline_epoch = invalid_baseline_epoch.max(usage.scanner_epoch);
+                    // This is still persisted state, so it must not enable a
+                    // missing-state bootstrap.  Continue to a legacy pair in
+                    // case it contains a complete, fenced snapshot.
+                } else {
+                    let backup_epoch = usage.scanner_epoch.unwrap_or_default();
+                    // A backup write from an older leader may complete after the
+                    // primary epoch has been fenced. It must not advance the startup
+                    // floor unless its epoch is at least as new as the primary.
+                    if primary_epoch.is_none_or(|epoch| backup_epoch >= epoch)
+                        && invalid_baseline_epoch.is_none_or(|epoch| backup_epoch >= epoch)
+                    {
+                        update_floor(&mut floor, &usage, &backup_path)?;
+                        any_found = true;
+                    }
                 }
             }
             Ok((None, _)) => {}
@@ -1413,6 +1433,11 @@ pub(super) async fn persisted_usage_floor_for_startup(
     }
 
     if !found_any && !bootstrap_pending {
+        if let Some(path) = invalid_baseline_path {
+            return Err(ScannerError::Other(format!(
+                "persisted scanner usage floor from {path} has no authoritative baseline or newer valid backup"
+            )));
+        }
         if !allow_missing_for_bootstrap {
             return Err(ScannerError::Other(
                 "persisted scanner usage floor has no authoritative baseline".to_string(),

crates/scanner/src/scanner/leadership.rs

@@ -82,9 +82,30 @@ pub(super) async fn usage_snapshot_for_epoch_fence(
     primary: Option<&[u8]>,
     allow_bootstrap_pending: bool,
 ) -> Result<Option<DataUsageInfo>, ScannerError> {
+    // A partially written v2 primary is not itself a baseline, but a durable
+    // companion may still provide one after an interrupted upgrade. Keep the
+    // primary epoch as a fence while checking those companions; malformed
+    // bytes and bootstrap markers retain their fail-closed behavior.
+    let mut invalid_primary_epoch = None;
     if let Some(primary) = primary {
-        return decode_usage_snapshot_for_epoch_fence(primary, DATA_USAGE_OBJ_NAME_PATH.as_str(), allow_bootstrap_pending)
-            .map(Some);
+        let usage: DataUsageInfo = serde_json::from_slice(primary).map_err(|err| {
+            ScannerError::Other(format!(
+                "failed to decode scanner usage epoch fence from {}: {err}",
+                DATA_USAGE_OBJ_NAME_PATH.as_str()
+            ))
+        })?;
+        if data_usage_info_has_persisted_baseline_identity(&usage)
+            || (allow_bootstrap_pending && data_usage_info_is_bootstrap_pending(&usage))
+        {
+            return Ok(Some(usage));
+        }
+        if data_usage_info_is_bootstrap_pending(&usage) {
+            return Err(ScannerError::Other(format!(
+                "scanner usage epoch fence from {} has no persisted baseline identity",
+                DATA_USAGE_OBJ_NAME_PATH.as_str()
+            )));
+        }
+        invalid_primary_epoch = usage.scanner_epoch;
     }
 
     let backup_path = format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str());
@@ -92,7 +113,10 @@ pub(super) async fn usage_snapshot_for_epoch_fence(
         .await
         .map_err(|err| ScannerError::Other(format!("failed to read scanner usage epoch fence backup: {err}")))?;
     if let Some(backup) = backup.as_deref() {
-        return decode_usage_snapshot_for_epoch_fence(backup, &backup_path, false).map(Some);
+        let usage = decode_usage_snapshot_for_epoch_fence(backup, &backup_path, false)?;
+        if invalid_primary_epoch.is_none_or(|epoch| usage.scanner_epoch.unwrap_or_default() >= epoch) {
+            return Ok(Some(usage));
+        }
     }
 
     for path in [
@@ -103,7 +127,10 @@ pub(super) async fn usage_snapshot_for_epoch_fence(
             .await
             .map_err(|err| ScannerError::Other(format!("failed to read legacy scanner usage epoch fence: {err}")))?;
         if let Some(legacy) = legacy.as_deref() {
-            return decode_usage_snapshot_for_epoch_fence(legacy, &path, false).map(Some);
+            let usage = decode_usage_snapshot_for_epoch_fence(legacy, &path, false)?;
+            if invalid_primary_epoch.is_none_or(|epoch| usage.scanner_epoch.unwrap_or_default() >= epoch) {
+                return Ok(Some(usage));
+            }
         }
     }
     // A missing usage snapshot is an uninitialized state, not an empty

crates/scanner/src/scanner/tests.rs

@@ -352,6 +352,7 @@ struct MemoryConfigStore {
     objects: Mutex<HashMap<String, Vec<u8>>>,
     revisions: Mutex<HashMap<String, u64>>,
     insert_after_gets: Mutex<HashMap<String, Vec<u8>>>,
+    delayed_gets: Mutex<HashMap<String, Duration>>,
     non_regular_objects: Mutex<HashSet<String>>,
     fail_put_number: Mutex<HashMap<String, usize>>,
     object_not_found_put_number: Mutex<HashMap<String, usize>>,
@@ -399,6 +400,9 @@ impl crate::storage_api::scanner_io::ObjectIO for MemoryConfigStore {
         _opts: &ObjectOptions,
     ) -> EcstoreResult<GetObjectReader> {
         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;
@@ -1923,6 +1927,176 @@ async fn scanner_startup_uses_primary_and_backup_usage_floor() {
     assert_eq!(epoch, 11);
 }
 
+#[tokio::test]
+async fn scanner_usage_floor_keeps_valid_primary_when_backup_has_no_identity() {
+    let store = Arc::new(MemoryConfigStore::default());
+    let backup_path = format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str());
+    let mut primary = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0);
+    primary.scanner_epoch = Some(8);
+    primary.scanner_cycle = Some(100);
+    let backup = DataUsageInfo {
+        scanner_epoch: Some(9),
+        scanner_cycle: Some(101),
+        usage_snapshot_complete: false,
+        ..Default::default()
+    };
+
+    for (path, usage) in [(DATA_USAGE_OBJ_NAME_PATH.as_str(), primary), (backup_path.as_str(), backup)] {
+        store.objects.lock().await.insert(
+            memory_config_key(RUSTFS_META_BUCKET, path),
+            serde_json::to_vec(&usage).expect("usage snapshot should encode"),
+        );
+    }
+
+    assert_eq!(
+        persisted_usage_floor(store)
+            .await
+            .expect("valid primary should remain authoritative"),
+        PersistedUsageFloor {
+            next_cycle: 101,
+            leader_epoch: 8,
+        }
+    );
+}
+
+#[tokio::test]
+async fn scanner_usage_floor_recovers_from_incomplete_v2_primary_using_fenced_backup() {
+    let store = Arc::new(MemoryConfigStore::default());
+    let backup_path = format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str());
+
+    // This shape is valid JSON from an interrupted v2 publication, but it is
+    // not a durable baseline because the snapshot is incomplete. It must not
+    // be converted into an empty floor.
+    let primary = DataUsageInfo {
+        scanner_epoch: Some(7),
+        scanner_cycle: Some(100),
+        usage_snapshot_complete: false,
+        ..Default::default()
+    };
+    let mut backup = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0);
+    backup.scanner_epoch = Some(7);
+    backup.scanner_cycle = Some(103);
+
+    for (path, usage) in [(DATA_USAGE_OBJ_NAME_PATH.as_str(), primary), (backup_path.as_str(), backup)] {
+        store.objects.lock().await.insert(
+            memory_config_key(RUSTFS_META_BUCKET, path),
+            serde_json::to_vec(&usage).expect("usage snapshot should encode"),
+        );
+    }
+
+    assert_eq!(
+        persisted_usage_floor(store)
+            .await
+            .expect("valid backup should recover the usage floor"),
+        PersistedUsageFloor {
+            next_cycle: 104,
+            leader_epoch: 7,
+        }
+    );
+}
+
+#[tokio::test]
+async fn scanner_usage_floor_does_not_bootstrap_over_incomplete_v2_primary() {
+    let store = Arc::new(MemoryConfigStore::default());
+    let primary = DataUsageInfo {
+        scanner_epoch: Some(7),
+        scanner_cycle: Some(100),
+        usage_snapshot_complete: false,
+        ..Default::default()
+    };
+    store.objects.lock().await.insert(
+        memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str()),
+        serde_json::to_vec(&primary).expect("usage snapshot should encode"),
+    );
+
+    let err = persisted_usage_floor_for_startup(store, true)
+        .await
+        .expect_err("an existing incomplete primary must remain fail-closed");
+    assert!(err.to_string().contains("no authoritative baseline"));
+}
+
+#[tokio::test]
+async fn scanner_usage_floor_rejects_backup_older_than_incomplete_v2_primary() {
+    let store = Arc::new(MemoryConfigStore::default());
+    let backup_path = format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str());
+    let primary = DataUsageInfo {
+        scanner_epoch: Some(7),
+        scanner_cycle: Some(100),
+        usage_snapshot_complete: false,
+        ..Default::default()
+    };
+    let mut backup = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0);
+    backup.scanner_epoch = Some(6);
+    backup.scanner_cycle = Some(10_000);
+
+    for (path, usage) in [(DATA_USAGE_OBJ_NAME_PATH.as_str(), primary), (backup_path.as_str(), backup)] {
+        store.objects.lock().await.insert(
+            memory_config_key(RUSTFS_META_BUCKET, path),
+            serde_json::to_vec(&usage).expect("usage snapshot should encode"),
+        );
+    }
+
+    let err = persisted_usage_floor_for_startup(store, true)
+        .await
+        .expect_err("an older backup must not cross the incomplete primary epoch fence");
+    assert!(err.to_string().contains("no authoritative baseline"));
+}
+
+#[tokio::test]
+async fn scanner_usage_floor_rejects_older_legacy_primary_after_incomplete_v2_primary() {
+    let store = Arc::new(MemoryConfigStore::default());
+    let primary = DataUsageInfo {
+        scanner_epoch: Some(7),
+        scanner_cycle: Some(100),
+        usage_snapshot_complete: false,
+        ..Default::default()
+    };
+    let mut legacy = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0);
+    legacy.scanner_epoch = Some(6);
+    legacy.scanner_cycle = Some(103);
+    for (path, usage) in [
+        (DATA_USAGE_OBJ_NAME_PATH.as_str(), primary),
+        (LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str(), legacy),
+    ] {
+        store.objects.lock().await.insert(
+            memory_config_key(RUSTFS_META_BUCKET, path),
+            serde_json::to_vec(&usage).expect("usage snapshot should encode"),
+        );
+    }
+
+    let err = persisted_usage_floor_for_startup(store, true)
+        .await
+        .expect_err("an older legacy baseline must not cross the incomplete v2 epoch fence");
+    assert!(err.to_string().contains("no authoritative baseline"));
+}
+
+#[tokio::test]
+async fn scanner_leadership_fencing_recovers_incomplete_v2_primary_from_backup() {
+    let store = Arc::new(MemoryConfigStore::default());
+    let backup_path = format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str());
+    let primary = serde_json::to_vec(&DataUsageInfo {
+        scanner_epoch: Some(7),
+        scanner_cycle: Some(100),
+        usage_snapshot_complete: false,
+        ..Default::default()
+    })
+    .expect("incomplete usage snapshot should encode");
+    let mut backup = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0);
+    backup.scanner_epoch = Some(7);
+    backup.scanner_cycle = Some(103);
+    store.objects.lock().await.insert(
+        memory_config_key(RUSTFS_META_BUCKET, &backup_path),
+        serde_json::to_vec(&backup).expect("backup usage snapshot should encode"),
+    );
+
+    let recovered = usage_snapshot_for_epoch_fence(store, Some(&primary), false)
+        .await
+        .expect("a valid backup should provide the fencing baseline")
+        .expect("the fencing baseline should be present");
+    assert_eq!(recovered.scanner_epoch, Some(7));
+    assert_eq!(recovered.scanner_cycle, Some(103));
+}
+
 #[tokio::test]
 async fn scanner_usage_floor_ignores_older_backup_after_primary_epoch_fence() {
     let store = Arc::new(MemoryConfigStore::default());
@@ -3182,6 +3356,79 @@ async fn test_observational_usage_defers_when_authoritative_baseline_is_missing(
     );
 }
 
+#[tokio::test]
+async fn test_observational_usage_uses_fenced_backup_when_v2_primary_has_no_identity() {
+    let store = Arc::new(MemoryConfigStore::default());
+    let primary = DataUsageInfo {
+        scanner_epoch: Some(7),
+        scanner_cycle: Some(100),
+        usage_snapshot_complete: false,
+        ..Default::default()
+    };
+    let mut backup = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0);
+    backup.scanner_epoch = Some(7);
+    backup.scanner_cycle = Some(103);
+    store.objects.lock().await.insert(
+        memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str()),
+        serde_json::to_vec(&primary).expect("incomplete primary should encode"),
+    );
+    store.objects.lock().await.insert(
+        memory_config_key(RUSTFS_META_BUCKET, &format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str())),
+        serde_json::to_vec(&backup).expect("backup baseline should encode"),
+    );
+
+    let (sender, receiver) = mpsc::channel(1);
+    let mut observation = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(20)), 1);
+    observation.usage_snapshot_converged = Some(false);
+    sender.send(observation).await.expect("observation should enqueue");
+    drop(sender);
+
+    let outcome = store_data_usage_in_backend_with_outcome_for_epoch_and_baseline_and_route_probe(
+        CancellationToken::new(),
+        store.clone(),
+        receiver,
+        None,
+        None,
+        || async { false },
+    )
+    .await;
+
+    assert_eq!(outcome, DataUsagePersistOutcome::Saved);
+    let observed = read_config(store, DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str())
+        .await
+        .expect("observational snapshot should be persisted");
+    let observed = serde_json::from_slice::<DataUsageInfo>(&observed).expect("observational snapshot should decode");
+    assert_eq!(observed.usage_snapshot_authoritative_baseline, Some(backup.snapshot_identity()));
+}
+
+#[tokio::test]
+async fn usage_baseline_does_not_fall_back_to_older_legacy_snapshot() {
+    let store = Arc::new(MemoryConfigStore::default());
+    let primary = DataUsageInfo {
+        scanner_epoch: Some(7),
+        scanner_cycle: Some(100),
+        usage_snapshot_complete: false,
+        ..Default::default()
+    };
+    let mut legacy = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0);
+    legacy.scanner_epoch = Some(6);
+    legacy.scanner_cycle = Some(103);
+    let primary_data = serde_json::to_vec(&primary).expect("incomplete primary should encode");
+    store.objects.lock().await.insert(
+        memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str()),
+        primary_data.clone(),
+    );
+    store.objects.lock().await.insert(
+        memory_config_key(RUSTFS_META_BUCKET, LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()),
+        serde_json::to_vec(&legacy).expect("legacy baseline should encode"),
+    );
+
+    let baseline = read_data_usage_persist_baseline(store)
+        .await
+        .expect("baseline inspection should complete");
+    assert_eq!(baseline.data.as_deref(), Some(primary_data.as_slice()));
+}
+
 #[tokio::test]
 #[serial]
 async fn test_usage_route_barrier_precedes_durable_reconciliation() {
@@ -3251,6 +3498,85 @@ async fn coordinator_does_not_put_after_remote_generation_flip() {
     assert_eq!(store.put_counts.lock().await.get(&key), None);
 }
 
+#[tokio::test]
+async fn coordinator_classifies_an_expired_publication_lease() {
+    let store = Arc::new(MemoryConfigStore::default());
+    let (sender, receiver) = mpsc::channel(1);
+    sender
+        .send(complete_usage_with_bucket_count(
+            Some(std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(20)),
+            1,
+        ))
+        .await
+        .expect("usage snapshot should enqueue");
+    drop(sender);
+
+    let expired = std::time::Instant::now()
+        .checked_sub(std::time::Duration::from_secs(1))
+        .expect("test instant should support a one-second subtraction");
+    let outcome =
+        store_data_usage_in_backend_with_outcome_for_epoch_and_baseline_and_route_probe_for_publication_epoch_and_lease_fence(
+            CancellationToken::new(),
+            store.clone(),
+            receiver,
+            None,
+            Some(DataUsagePersistBaseline {
+                data: None,
+                revision: DataUsageCacheRevision::Missing,
+            }),
+            ScannerPublicationFence::new(None, Some(expired), None),
+            || async { false },
+        )
+        .await;
+
+    assert_eq!(
+        outcome,
+        DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded)
+    );
+    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() {
@@ -4302,6 +4628,8 @@ fn scanner_cycle_cache_floor_stays_pending_during_deferred_usage_publication() {
     for reason in [
         ScannerCycleDeferReason::DataMovement,
         ScannerCycleDeferReason::ActivityBaselineUnavailable,
+        ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded,
+        ScannerCycleDeferReason::PublicationLeaseReleaseFailed,
     ] {
         let deferred = DataUsagePersistOutcome::Deferred(reason);
         assert_eq!(
@@ -4368,6 +4696,26 @@ fn finalizing_a_deferred_usage_save_keeps_dirty_work_pending() {
     crate::scanner_io::clear_dirty_usage_bucket("photos");
 }
 
+#[test]
+#[serial]
+fn finalizing_post_scan_observation_advances_partially_without_dirty_ack() {
+    crate::scanner_io::clear_dirty_usage_bucket("photos");
+    crate::scanner_io::record_dirty_usage_bucket("photos");
+    let dirty_snapshot = crate::scanner_io::dirty_usage_buckets_for_tests();
+    let observed = crate::scanner_io::ScannerCycleResult::new(
+        ScannerCycleStatus::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable),
+        Some(dirty_snapshot),
+    )
+    .with_observational_snapshot_published(true);
+
+    let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(observed, DataUsagePersistOutcome::Saved);
+
+    assert_eq!(outcome, ScannerCycleOutcome::Partial);
+    assert!(acknowledgements.is_empty());
+    assert!(crate::scanner_io::dirty_usage_buckets_pending());
+    crate::scanner_io::clear_dirty_usage_bucket("photos");
+}
+
 #[tokio::test]
 async fn scanner_cycle_keeps_remote_pending_acknowledgement() {
     let pending = remote_dirty_usage_acknowledgement_pending(7, 1, std::future::ready(Ok::<bool, std::io::Error>(true))).await;
@@ -4451,6 +4799,18 @@ fn data_usage_persist_wait_covers_cache_retries_and_backup() {
     crate::runtime_config::refresh_scanner_runtime_config_for_tests();
 }
 
+#[test]
+fn scanner_publication_lease_deadline_reason_remains_distinct() {
+    assert_eq!(
+        ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded.as_str(),
+        "publication_lease_deadline_exceeded"
+    );
+    assert_eq!(
+        ScannerCycleDeferReason::PublicationLeaseReleaseFailed.as_str(),
+        "publication_lease_release_failed"
+    );
+}
+
 #[tokio::test]
 async fn data_usage_persist_wait_aborts_when_scanner_is_cancelled() {
     let ctx = CancellationToken::new();
@@ -4480,6 +4840,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();

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,12 +37,100 @@ fn remote_lease_expired(deadline: Option<std::time::Instant>) -> 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<std::time::Instant>,
+) -> 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<Bytes>,
     pub(super) revision: DataUsageCacheRevision,
 }
 
+/// Read the bytes used as the baseline for a usage publication while keeping
+/// the v2 primary revision as the CAS fence. During an interrupted upgrade the
+/// primary can be valid JSON without a baseline identity; in that case a
+/// same-or-newer durable companion may still be used, but an older legacy
+/// snapshot must not cross the primary's epoch fence.
+pub(super) async fn read_data_usage_persist_baseline(
+    storeapi: Arc<impl ScannerObjectIO>,
+) -> Result<DataUsagePersistBaseline, EcstoreError> {
+    let (primary, revision) = read_config_with_revision(storeapi.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str()).await?;
+    let Some(primary) = primary else {
+        for path in [
+            format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str()),
+            LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str().to_string(),
+            format!("{}.bkp", LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()),
+        ] {
+            let (candidate, _) = read_config_with_revision(storeapi.clone(), &path).await?;
+            let Some(candidate) = candidate else {
+                continue;
+            };
+            let Ok(usage) = serde_json::from_slice::<DataUsageInfo>(&candidate) else {
+                continue;
+            };
+            if data_usage_info_has_persisted_baseline_identity(&usage) {
+                return Ok(DataUsagePersistBaseline {
+                    data: Some(Bytes::from(candidate)),
+                    revision,
+                });
+            }
+        }
+        return Ok(DataUsagePersistBaseline { data: None, revision });
+    };
+
+    let Ok(primary_info) = serde_json::from_slice::<DataUsageInfo>(&primary) else {
+        // Preserve the original bytes and revision. A completed scan may
+        // replace the invalid primary under this CAS fence; an observation
+        // will still reject it below because it has no verifiable identity.
+        return Ok(DataUsagePersistBaseline {
+            data: Some(Bytes::from(primary)),
+            revision,
+        });
+    };
+    if data_usage_info_has_persisted_baseline_identity(&primary_info) || data_usage_info_is_bootstrap_pending(&primary_info) {
+        return Ok(DataUsagePersistBaseline {
+            data: Some(Bytes::from(primary)),
+            revision,
+        });
+    }
+
+    let invalid_primary_epoch = primary_info.scanner_epoch;
+    for path in [
+        format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str()),
+        LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str().to_string(),
+        format!("{}.bkp", LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()),
+    ] {
+        let (candidate, _) = read_config_with_revision(storeapi.clone(), &path).await?;
+        let Some(candidate) = candidate else {
+            continue;
+        };
+        let Ok(usage) = serde_json::from_slice::<DataUsageInfo>(&candidate) else {
+            continue;
+        };
+        let candidate_epoch = usage.scanner_epoch.unwrap_or_default();
+        if data_usage_info_has_persisted_baseline_identity(&usage)
+            && invalid_primary_epoch.is_none_or(|epoch| candidate_epoch >= epoch)
+        {
+            return Ok(DataUsagePersistBaseline {
+                data: Some(Bytes::from(candidate)),
+                revision,
+            });
+        }
+    }
+
+    Ok(DataUsagePersistBaseline {
+        data: Some(Bytes::from(primary)),
+        revision,
+    })
+}
+
 /// Short-lived publication inputs captured for one usage persistence attempt.
 /// Keeping the movement epoch, lease deadline, and target fence together makes
 /// it explicit that they are one proof rather than independent options.
@@ -48,6 +139,8 @@ pub(super) struct ScannerPublicationFence {
     pub(super) expected_publication_epoch: Option<u64>,
     pub(super) remote_lease_deadline: Option<std::time::Instant>,
     pub(super) scanner_publication_lease_fence: Option<String>,
+    pub(super) remote_lease_tokens: Vec<Uuid>,
+    pub(super) lease_release_safe: Arc<AtomicBool>,
 }
 
 impl ScannerPublicationFence {
@@ -60,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<Uuid>) -> Self {
+        self.remote_lease_tokens = remote_lease_tokens;
+        self
+    }
+
+    pub(super) fn with_lease_release_flag(mut self, lease_release_safe: Arc<AtomicBool>) -> Self {
+        self.lease_release_safe = lease_release_safe;
+        self
+    }
 }
 
 #[derive(Debug)]
@@ -212,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;
@@ -225,7 +332,7 @@ where
             data_usage_info.scanner_epoch = Some(leader_epoch);
         }
         if remote_lease_expired(remote_lease_deadline) {
-            outcome = DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable);
+            outcome = DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded);
             break 'updates;
         }
         if let Some(expected_epoch) = expected_publication_epoch
@@ -281,8 +388,8 @@ where
             publication_epoch = Some(read_epoch);
             let authoritative_data = match next_baseline.as_ref() {
                 Some(baseline) => baseline.data.clone(),
-                None => match read_config_with_revision(storeapi.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str()).await {
-                    Ok((data, _)) => data.map(Bytes::from),
+                None => match read_data_usage_persist_baseline(storeapi.clone()).await {
+                    Ok(baseline) => baseline.data,
                     Err(err) => {
                         error!(
                             target: "rustfs::scanner",
@@ -497,30 +604,59 @@ where
                 break DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement);
             }
             if remote_lease_expired(remote_lease_deadline) {
-                break DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable);
+                break DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded);
             }
 
             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::ActivityBaselineUnavailable);
+                    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();
 
@@ -618,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 {
@@ -641,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 {
@@ -683,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 {
@@ -700,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();
@@ -739,6 +883,8 @@ async fn cleanup_observed_data_usage_snapshot_for_epoch_and_lease(
     expected_publication_epoch: Option<u64>,
     remote_lease_deadline: Option<std::time::Instant>,
     scanner_publication_lease_fence: Option<&str>,
+    remote_lease_tokens: &[Uuid],
+    lease_release_safe: Arc<AtomicBool>,
 ) -> bool {
     if remote_lease_expired(remote_lease_deadline) {
         return false;
@@ -807,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(),
@@ -826,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(

crates/scanner/src/scanner_io.rs

@@ -269,6 +269,10 @@ fn should_publish_usage_snapshot(status: ScannerCycleStatus) -> bool {
     matches!(status, ScannerCycleStatus::Complete | ScannerCycleStatus::Superseded)
 }
 
+fn should_publish_observational_snapshot(status: ScannerCycleStatus) -> bool {
+    matches!(status, ScannerCycleStatus::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable))
+}
+
 fn prepare_usage_snapshot_for_publication(
     status: ScannerCycleStatus,
     mut data_usage_info: DataUsageInfo,
@@ -570,6 +574,13 @@ pub(crate) async fn scanner_set_disk_inventory(set: &SetDisks) -> Vec<Arc<Disk>>
 pub(crate) enum ScannerCycleDeferReason {
     ActivityBaselineUnavailable,
     DataMovement,
+    /// A granted lease's absolute deadline cannot cover the persistence
+    /// operation. This can occur even when the configured budget fits the
+    /// nominal TTL because lease acquisition consumed part of the window.
+    PublicationLeaseDeadlineExceeded,
+    /// A remote lease could not be released after the persistence attempt.
+    /// Keep the cycle deferred because the peer may still admit movement.
+    PublicationLeaseReleaseFailed,
 }
 
 impl ScannerCycleDeferReason {
@@ -577,6 +588,8 @@ impl ScannerCycleDeferReason {
         match self {
             Self::ActivityBaselineUnavailable => "activity_baseline_unavailable",
             Self::DataMovement => "data_movement",
+            Self::PublicationLeaseDeadlineExceeded => "publication_lease_deadline_exceeded",
+            Self::PublicationLeaseReleaseFailed => "publication_lease_release_failed",
         }
     }
 }
@@ -611,6 +624,7 @@ fn scanner_activity_preflight(
 pub(crate) struct ScannerCycleResult {
     pub(crate) status: ScannerCycleStatus,
     publication_epoch: Option<u64>,
+    observational_snapshot_published: bool,
     dirty_usage_clear: Option<DirtyUsageBuckets>,
     remote_dirty_usage_acknowledgements: Vec<crate::scanner::ScannerDirtyUsageAcknowledgement>,
     remote_publication_lease_targets: Vec<(String, String, u64)>,
@@ -624,6 +638,7 @@ impl ScannerCycleResult {
         Self {
             status,
             publication_epoch: None,
+            observational_snapshot_published: false,
             dirty_usage_clear,
             remote_dirty_usage_acknowledgements: Vec::new(),
             remote_publication_lease_targets: Vec::new(),
@@ -642,6 +657,15 @@ impl ScannerCycleResult {
         self.publication_epoch
     }
 
+    pub(crate) fn with_observational_snapshot_published(mut self, published: bool) -> Self {
+        self.observational_snapshot_published = published;
+        self
+    }
+
+    pub(crate) fn has_observational_snapshot(&self) -> bool {
+        self.observational_snapshot_published
+    }
+
     fn with_failed_dirty_usage(mut self, failed_dirty_usage: bool) -> Self {
         self.failed_dirty_usage = failed_dirty_usage;
         self

crates/scanner/src/scanner_io/guards.rs

@@ -286,6 +286,16 @@ pub(super) fn scanner_task_join_error(stage: &str, err: tokio::task::JoinError)
 mod tests {
     use super::*;
     use rustfs_scanner_contracts::metrics::{ScannerWorkSource, global_metrics};
+    use tokio::sync::oneshot;
+
+    fn active_bucket_drive_count(source: ScannerWorkSource, bucket: &str, drive: &str) -> u64 {
+        global_metrics()
+            .scanner_runtime_details_report()
+            .active_bucket_drive_scans
+            .into_iter()
+            .find(|active| active.source == source.as_str() && active.bucket == bucket && active.drive == drive)
+            .map_or(0, |active| active.count)
+    }
 
     #[test]
     fn bucket_drive_failure_guard_retires_active_scan_on_drop() {
@@ -305,4 +315,85 @@ mod tests {
                 .any(|active| active.source == source.as_str() && active.bucket == bucket && active.drive == drive)
         );
     }
+
+    #[tokio::test]
+    async fn bucket_drive_failure_guard_retires_active_scan_after_cancellation() {
+        let source = ScannerWorkSource::Usage;
+        let bucket = "__guard_cancel_lifecycle_test__";
+        let drive = "/__guard_cancel_lifecycle_test__";
+        global_metrics().record_scan_bucket_drive_start(source, bucket, drive);
+
+        let cancellation = CancellationToken::new();
+        let worker_cancellation = cancellation.clone();
+        let worker = tokio::spawn(async move {
+            let mut guard = BucketDriveFailureGuard::new(source, bucket, drive);
+            worker_cancellation.cancelled().await;
+            guard.mark_not_failed();
+        });
+
+        cancellation.cancel();
+        worker.await.expect("cancelled scanner worker should finish");
+
+        assert_eq!(active_bucket_drive_count(source, bucket, drive), 0);
+    }
+
+    #[tokio::test]
+    async fn bucket_drive_failure_guard_retires_active_scan_when_worker_is_aborted() {
+        let source = ScannerWorkSource::Bitrot;
+        let bucket = "__guard_abort_lifecycle_test__";
+        let drive = "/__guard_abort_lifecycle_test__";
+        global_metrics().record_scan_bucket_drive_start(source, bucket, drive);
+
+        let (started_sender, started_receiver) = oneshot::channel();
+        let worker = tokio::spawn(async move {
+            let _guard = BucketDriveFailureGuard::new(source, bucket, drive);
+            started_sender.send(()).expect("test should observe worker start");
+            std::future::pending::<()>().await;
+        });
+        started_receiver.await.expect("scanner worker should start");
+        assert_eq!(active_bucket_drive_count(source, bucket, drive), 1);
+
+        worker.abort();
+        worker.await.expect_err("aborted scanner worker should report cancellation");
+
+        assert_eq!(active_bucket_drive_count(source, bucket, drive), 0);
+    }
+
+    #[tokio::test]
+    async fn bucket_drive_failure_guards_track_overlapping_scans_independently() {
+        let source = ScannerWorkSource::Usage;
+        let bucket = "__guard_overlap_lifecycle_test__";
+        let drive = "/__guard_overlap_lifecycle_test__";
+        global_metrics().record_scan_bucket_drive_start(source, bucket, drive);
+        global_metrics().record_scan_bucket_drive_start(source, bucket, drive);
+
+        let (first_release_sender, first_release_receiver) = oneshot::channel();
+        let (second_release_sender, second_release_receiver) = oneshot::channel();
+        let (first_started_sender, first_started_receiver) = oneshot::channel();
+        let (second_started_sender, second_started_receiver) = oneshot::channel();
+        let first = tokio::spawn(async move {
+            let _guard = BucketDriveFailureGuard::new(source, bucket, drive);
+            first_started_sender.send(()).expect("test should observe first worker start");
+            first_release_receiver.await.expect("first worker should be released");
+        });
+        let second = tokio::spawn(async move {
+            let _guard = BucketDriveFailureGuard::new(source, bucket, drive);
+            second_started_sender
+                .send(())
+                .expect("test should observe second worker start");
+            second_release_receiver.await.expect("second worker should be released");
+        });
+
+        first_started_receiver.await.expect("first scanner worker should start");
+        second_started_receiver.await.expect("second scanner worker should start");
+        assert_eq!(active_bucket_drive_count(source, bucket, drive), 2);
+
+        first_release_sender.send(()).expect("first worker should be released");
+        first.await.expect("first scanner worker should finish");
+        assert_eq!(active_bucket_drive_count(source, bucket, drive), 1);
+
+        second_release_sender.send(()).expect("second worker should be released");
+        second.await.expect("second scanner worker should finish");
+        assert_eq!(active_bucket_drive_count(source, bucket, drive), 0);
+    }
 }

crates/scanner/src/scanner_io/io_cycle.rs

@@ -162,18 +162,18 @@ impl ScannerIOCycle for ECStore {
                 dirty_usage_status,
                 activity_status,
             );
-            if !publish_usage_snapshot(
-                &updates,
-                status,
-                DataUsageInfo {
-                    last_update: Some(SystemTime::now()),
-                    scanner_cycle: Some(want_cycle),
-                    usage_snapshot_complete: true,
-                    ..Default::default()
-                },
-            )
-            .await?
-            {
+            let empty_usage = DataUsageInfo {
+                last_update: Some(SystemTime::now()),
+                scanner_cycle: Some(want_cycle),
+                usage_snapshot_complete: true,
+                ..Default::default()
+            };
+            let observational_snapshot_published = if should_publish_observational_snapshot(status) {
+                publish_observational_snapshot(&updates, empty_usage).await?
+            } else {
+                publish_usage_snapshot(&updates, status, empty_usage).await?
+            };
+            if !observational_snapshot_published {
                 return Ok(ScannerCycleResult::new(status, None).with_publication_epoch(publication_epoch));
             }
             if status == ScannerCycleStatus::Complete {
@@ -188,6 +188,7 @@ impl ScannerIOCycle for ECStore {
             };
             return Ok(ScannerCycleResult::new(status, dirty_usage_clear)
                 .with_publication_epoch(publication_epoch)
+                .with_observational_snapshot_published(observational_snapshot_published)
                 .with_remote_publication_lease_targets(remote_publication_lease_targets)
                 .with_remote_dirty_usage_acknowledgements(remote_dirty_usage_acknowledgements));
         }
@@ -437,13 +438,19 @@ impl ScannerIOCycle for ECStore {
             dirty_usage_status,
             activity_status,
         );
-        if let Some((data_usage_info, _)) = completed_usage {
-            publish_usage_snapshot(&updates, cycle_status, data_usage_info).await?;
+        let observational_snapshot_published = if let Some((data_usage_info, _)) = completed_usage {
+            if should_publish_observational_snapshot(cycle_status) {
+                publish_observational_snapshot(&updates, data_usage_info).await?
+            } else {
+                publish_usage_snapshot(&updates, cycle_status, data_usage_info).await?
+            }
         } else if !ctx.is_cancelled()
             && let Some((data_usage_info, _)) = observational_usage
         {
-            publish_observational_snapshot(&updates, data_usage_info).await?;
-        }
+            publish_observational_snapshot(&updates, data_usage_info).await?
+        } else {
+            false
+        };
         let dirty_usage_clear = should_clear_dirty_usage_snapshot(
             result.is_ok(),
             structurally_complete_snapshot,
@@ -463,6 +470,7 @@ impl ScannerIOCycle for ECStore {
         };
         Ok(ScannerCycleResult::new(cycle_status, dirty_usage_clear)
             .with_publication_epoch(publication_epoch)
+            .with_observational_snapshot_published(observational_snapshot_published)
             .with_remote_publication_lease_targets(remote_publication_lease_targets)
             .with_remote_dirty_usage_acknowledgements(remote_dirty_usage_acknowledgements)
             .with_failed_dirty_usage(!failed_buckets.is_empty())

crates/scanner/src/scanner_io/tests.rs

@@ -909,6 +909,47 @@ async fn structurally_complete_superseded_cycles_publish_without_claiming_conver
     );
 }
 
+#[tokio::test]
+async fn post_scan_activity_failure_retains_complete_usage_as_observation() {
+    let (updates, mut receiver) = mpsc::channel(1);
+    let status = ScannerCycleStatus::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable);
+
+    assert!(should_publish_observational_snapshot(status));
+    assert!(
+        publish_observational_snapshot(
+            &updates,
+            DataUsageInfo {
+                last_update: Some(SystemTime::now()),
+                scanner_cycle: Some(7),
+                objects_total_count: 3,
+                objects_total_size: 12,
+                usage_snapshot_complete: true,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("post-scan activity failure should retain an observation")
+    );
+
+    let observed = receiver.recv().await.expect("observational update should be queued");
+    assert!(!observed.usage_snapshot_complete);
+    assert!(observed.usage_snapshot_partial);
+    assert_eq!(observed.usage_snapshot_converged, Some(false));
+    assert_eq!(observed.objects_total_count, 3);
+    assert_eq!(observed.objects_total_size, 12);
+}
+
+#[test]
+fn only_unverified_activity_allows_post_scan_observation() {
+    assert!(should_publish_observational_snapshot(ScannerCycleStatus::Deferred(
+        ScannerCycleDeferReason::ActivityBaselineUnavailable
+    )));
+    assert!(!should_publish_observational_snapshot(ScannerCycleStatus::Deferred(
+        ScannerCycleDeferReason::DataMovement
+    )));
+    assert!(!should_publish_observational_snapshot(ScannerCycleStatus::Incomplete));
+}
+
 #[test]
 fn scanner_cycle_fails_closed_for_namespace_disappearance() {
     for activity_status in [ScannerCycleActivityStatus::Changed, ScannerCycleActivityStatus::Unchanged] {

crates/scanner/src/storage_api.rs

@@ -92,7 +92,9 @@ pub(crate) use rustfs_ecstore::api::event::{EventArgs as EcstoreEventArgs, send_
 pub(crate) use rustfs_ecstore::api::layout::{
     EndpointServerPools as EcstoreEndpointServerPools, Endpoints as EcstoreEndpoints, PoolEndpoints as EcstorePoolEndpoints,
 };
-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,
@@ -106,9 +108,9 @@ pub(crate) use rustfs_ecstore::api::runtime::{
     setup_is_erasure_sd as ecstore_is_erasure_sd,
 };
 pub(crate) use rustfs_ecstore::api::set_disk::SetDisks as EcstoreSetDisks;
+pub(crate) use rustfs_ecstore::api::storage::ECStore as EcstoreStore;
 #[cfg(test)]
 pub(crate) use rustfs_ecstore::api::storage::init_local_disks_with_instance_ctx as ecstore_init_local_disks_with_instance_ctx;
-pub(crate) use rustfs_ecstore::api::storage::{ECStore as EcstoreStore, SCANNER_PUBLICATION_LEASE_TTL_MS};
 use rustfs_storage_api as storage_contracts;
 
 pub(crate) mod owner {
@@ -125,7 +127,7 @@ pub(crate) mod owner {
         EcstoreNsScannerOpenRequest, EcstoreObjectLockConfiguration, EcstoreObjectOpts, EcstoreReplicationConfigurationExt,
         EcstoreReplicationScannerBridge, EcstoreResultType, EcstoreScanGuard, EcstoreSetDisks, EcstoreStorageError, EcstoreStore,
         EcstoreVersioningApi, EcstoreVersioningConfiguration, SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY,
-        SCANNER_PUBLICATION_LEASE_TTL_MS, ScannerReplicationHealObject, ScannerReplicationHealResult,
+        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,

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<Arc<dyn Send + Sync>> =
+                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<Arc<dyn Send + Sync>> =
+                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,