refactor(ecstore): make store-to-disk error narrowing named and fallible (#6626)

refactor(ecstore): make store-to-disk error narrowing a named fallible operation

Backlog#1845 step 4. The blanket impl From<StorageError> for DiskError let ? silently push store-only errors (locks, buckets, quotas) across the disk boundary into DiskError::other, where the rendered message fragments reduce_errs quorum buckets. Same story for the blanket From<StorageError> for rustfs_filemeta::Error and its other() catch-all.

Both impls are replaced by named, fallible methods: StorageError::narrow_to_disk() and StorageError::narrow_to_filemeta(). Variants with an identity on the far side map across unchanged - including the two documented lossy collapses (SlowDown -> TooManyOpenFiles, StorageFull -> DiskFull) that the round-trip tests pin - and everything else returns Err(self) so the call site decides what crossing the boundary means. Removing the impls let the compiler enumerate every conversion site; the census that scoped this issue had found 5, the compiler found 33.

Call sites keep their existing behavior: the io identity bridge and the generic sites fold Err into the io-backed other() exactly as the old catch-all did (identity still recoverable by downcast), listing paths use one shared to_filemeta_err helper, and the two sites that relied on the SlowDown collapse now construct DiskError::TooManyOpenFiles directly so the loss is visible where it happens. No behavior change intended anywhere; the io::Error bridge itself is untouched by design.

Ref rustfs/backlog#1845

crates/ecstore/src/set_disk/ops/heal.rs

@@ -474,13 +474,15 @@ impl SetDisks {
         // Bound, not `_`: this guard must live to the end of the scope. A bare
         // `_` would drop it here and release the namespace write lock.
         let _write_lock_guard = if !opts.no_lock {
-            let ns_lock = self.new_ns_lock(bucket, object).await?;
-            Some(
-                ns_lock
-                    .get_write_lock(get_lock_acquire_timeout())
-                    .await
-                    .map_err(|e| self.map_namespace_lock_error(bucket, object, "write", e))?,
-            )
+            let ns_lock = self
+                .new_ns_lock(bucket, object)
+                .await
+                .map_err(|e| e.narrow_to_disk().unwrap_or_else(DiskError::other))?;
+            Some(ns_lock.get_write_lock(get_lock_acquire_timeout()).await.map_err(|e| {
+                self.map_namespace_lock_error(bucket, object, "write", e)
+                    .narrow_to_disk()
+                    .unwrap_or_else(DiskError::other)
+            })?)
         } else {
             None
         };
@@ -2200,13 +2202,15 @@ impl crate::storage_api_contracts::heal::HealOperations for SetDisks {
         opts: &HealOpts,
     ) -> Result<(HealResultItem, Option<Error>)> {
         let _write_lock_guard = if !opts.no_lock {
-            let ns_lock = self.new_ns_lock(bucket, object).await?;
-            Some(
-                ns_lock
-                    .get_write_lock(get_lock_acquire_timeout())
-                    .await
-                    .map_err(|e| self.map_namespace_lock_error(bucket, object, "write", e))?,
-            )
+            let ns_lock = self
+                .new_ns_lock(bucket, object)
+                .await
+                .map_err(|e| e.narrow_to_disk().unwrap_or_else(DiskError::other))?;
+            Some(ns_lock.get_write_lock(get_lock_acquire_timeout()).await.map_err(|e| {
+                self.map_namespace_lock_error(bucket, object, "write", e)
+                    .narrow_to_disk()
+                    .unwrap_or_else(DiskError::other)
+            })?)
         } else {
             None
         };
@@ -2306,13 +2310,15 @@ impl crate::storage_api_contracts::heal::HealOperations for SetDisks {
     async fn check_abandoned_parts(&self, bucket: &str, object: &str, opts: &HealOpts) -> Result<()> {
         let started_at = std::time::Instant::now();
         let _write_lock_guard = if !opts.no_lock {
-            let ns_lock = self.new_ns_lock(bucket, object).await?;
-            Some(
-                ns_lock
-                    .get_write_lock(get_lock_acquire_timeout())
-                    .await
-                    .map_err(|e| self.map_namespace_lock_error(bucket, object, "write", e))?,
-            )
+            let ns_lock = self
+                .new_ns_lock(bucket, object)
+                .await
+                .map_err(|e| e.narrow_to_disk().unwrap_or_else(DiskError::other))?;
+            Some(ns_lock.get_write_lock(get_lock_acquire_timeout()).await.map_err(|e| {
+                self.map_namespace_lock_error(bucket, object, "write", e)
+                    .narrow_to_disk()
+                    .unwrap_or_else(DiskError::other)
+            })?)
         } else {
             None
         };

crates/ecstore/src/set_disk/ops/object.rs

@@ -2205,7 +2205,7 @@ impl SetDisks {
         }
         let current = read_object_transaction_epoch_fence(self, bucket, object)
             .await
-            .map_err(DiskError::from)?;
+            .map_err(|e| e.narrow_to_disk().unwrap_or_else(DiskError::other))?;
         let disks = self.get_disks_internal().await;
         let mut removed = 0usize;