fix(scanner): own fenced publication mutations

Remove the static lease budget rejection and carry an effective absolute deadline into storage-owned publication scopes. Keep remote grants until a scope reaches a safe terminal state.

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

crates/ecstore/src/object_api/types.rs

@@ -420,6 +420,7 @@ struct ScannerPublicationCommitScopeInner {
     /// 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.
@@ -453,6 +454,25 @@ impl ScannerPublicationCommitScope {
         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,
@@ -462,6 +482,7 @@ impl ScannerPublicationCommitScope {
                 state: AtomicU8::new(SCANNER_PUBLICATION_SCOPE_ADMITTED),
                 completed: Notify::new(),
                 movement_permit: Mutex::new(Some(movement_permit)),
+                lease_release_safe,
             }),
         }
     }
@@ -537,21 +558,20 @@ impl ScannerPublicationCommitScope {
     }
 
     pub fn mark_aborted_before_commit(&self) -> bool {
-        for expected in [SCANNER_PUBLICATION_SCOPE_ADMITTED, SCANNER_PUBLICATION_SCOPE_IN_FLIGHT] {
-            if self
-                .inner
-                .state
-                .compare_exchange(
-                    expected,
-                    SCANNER_PUBLICATION_SCOPE_ABORTED_BEFORE_COMMIT,
-                    Ordering::AcqRel,
-                    Ordering::Acquire,
-                )
-                .is_ok()
-            {
-                self.inner.completed.notify_waiters();
-                return true;
-            }
+        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
     }
@@ -565,7 +585,12 @@ impl ScannerPublicationCommitScope {
             .state
             .compare_exchange(SCANNER_PUBLICATION_SCOPE_IN_FLIGHT, terminal.as_u8(), Ordering::AcqRel, Ordering::Acquire)
             .is_ok()
-            .then(|| self.inner.completed.notify_waiters())
+            .then(|| {
+                if terminal.permits_lease_release() {
+                    self.inner.lease_release_safe.store(true, Ordering::Release);
+                }
+                self.inner.completed.notify_waiters()
+            })
             .is_some()
     }
 
@@ -595,6 +620,14 @@ impl ScannerPublicationCommitScope {
     }
 }
 
+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

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

@@ -3394,7 +3394,12 @@ 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 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_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;
@@ -3491,7 +3496,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 +3507,21 @@ 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 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,
@@ -3540,6 +3559,13 @@ impl SetDisks {
                     ),
                 )
                 .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;

crates/ecstore/src/store/mod.rs

@@ -544,6 +544,30 @@ impl ECStore {
         ))
     }
 
+    /// 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.
@@ -1521,6 +1545,10 @@ mod tests {
             .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,