fix(heal): fence format writes during transitions

crates/ecstore/src/core/sets.rs

@@ -985,14 +985,11 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for Sets {
     }
 }
 
-#[async_trait::async_trait]
+impl Sets {
-impl crate::storage_api_contracts::heal::HealOperations for Sets {
+    pub(crate) async fn heal_format_with_fence<F>(&self, dry_run: bool, fence_lost: F) -> Result<(HealResultItem, Option<Error>)>
-    type Error = Error;
+    where
-    type HealResultItem = HealResultItem;
+        F: Fn() -> bool + Send + Sync,
-    type HealOptions = HealOpts;
+    {
-
-    #[tracing::instrument(skip(self))]
-    async fn heal_format(&self, dry_run: bool) -> Result<(HealResultItem, Option<Error>)> {
         let (disks, init_errs) = init_storage_disks_with_errors(
             &self.endpoints.endpoints,
             &DiskOption {
@@ -1065,6 +1062,9 @@ impl crate::storage_api_contracts::heal::HealOperations for Sets {
             // Save new formats `format.json` on unformatted disks.
             for (index, (fm, disk)) in tmp_new_formats.iter_mut().zip(disks.iter()).enumerate() {
                 if fm.is_some() && disk.is_some() {
+                    if fence_lost() {
+                        return Ok((res, Some(StorageError::SlowDown)));
+                    }
                     if let Err(err) = save_format_file(disk, fm).await {
                         if let Some(disk) = disk.as_ref() {
                             let _ = disk.close().await;
@@ -1098,6 +1098,18 @@ impl crate::storage_api_contracts::heal::HealOperations for Sets {
         }
         Ok((res, None))
     }
+}
+
+#[async_trait::async_trait]
+impl crate::storage_api_contracts::heal::HealOperations for Sets {
+    type Error = Error;
+    type HealResultItem = HealResultItem;
+    type HealOptions = HealOpts;
+
+    #[tracing::instrument(skip(self))]
+    async fn heal_format(&self, dry_run: bool) -> Result<(HealResultItem, Option<Error>)> {
+        self.heal_format_with_fence(dry_run, || false).await
+    }
     #[tracing::instrument(skip(self))]
     async fn heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result<HealResultItem> {
         let mut result = HealResultItem {

crates/ecstore/src/store/heal.rs

@@ -187,7 +187,8 @@ impl ECStore {
                 continue;
             }
 
-            let (mut result, err) = pool.heal_format(dry_run).await?;
+            let fence_lost = || pool_guard.is_lock_lost() || rebalance_guard.is_lock_lost();
+            let (mut result, err) = pool.heal_format_with_fence(dry_run, fence_lost).await?;
             if let Some(err) = err {
                 match err {
                     StorageError::NoHealRequired => {
@@ -203,8 +204,8 @@ impl ECStore {
             r.before.drives.append(&mut result.before.drives);
             r.after.drives.append(&mut result.after.drives);
 
-            // Sets::heal_format cannot observe this guard before each disk write;
+            // A lease can be lost after the final write; fail closed before
-            // fail closed after the call if the lease was lost during format IO.
+            // reporting the pool as successfully healed.
             if pool_guard.is_lock_lost() || rebalance_guard.is_lock_lost() {
                 first_error.get_or_insert(heal_format_fence_lost_error());
                 break;