refactor(heal): classify recoverability typed-first with documented fallback (#6629)

refactor(heal): classify recoverability typed-first with documented needle fallback

Backlog#1845 step 6. Heal's retry decision leaned on substring matching of rendered messages; the typed information available in the error values now takes priority:

- Lock failures classify by LockError's own taxonomy instead of the blanket Lock(_) => recoverable: fatal variants (ResourceNotFound / PermissionDenied / Configuration) are terminal since retrying cannot fix them, while contention and transport variants (Timeout, Network, Internal, AlreadyLocked, QuorumNotReached, InsufficientNodes, ...) stay recoverable exactly as before.
- DiskError::RemoteClientUnavailable and its StorageError twin (typed in #6619) join the typed recoverable lists, so client-acquisition failures no longer depend on which needle happens to appear in the detail.
- task.rs is_transient_lock_or_timeout_error consults LockError::is_retryable / QuorumNotReached and the typed Timeout variants before falling back to needles.
- The substring list is demoted to a documented fallback: every needle now carries a producer census comment naming what emits it, with the shrink-only rule stated (delete the needle when its producer becomes typed end-to-end). heal rename incomplete remains the one needle with no typed producer.

heal gains a direct rustfs-lock dependency (already transitive via ecstore) to name LockError variants.

New tests pin each typed source: contention/transport lock variants recoverable, fatal lock variants terminal, RemoteClientUnavailable recoverable with a detail that avoids every needle. All existing recoverability tests stay green.

Ref rustfs/backlog#1845

Cargo.lock

@@ -9664,6 +9664,7 @@ dependencies = [
  "rustfs-concurrency",
  "rustfs-config",
  "rustfs-ecstore",
+ "rustfs-lock",
  "rustfs-madmin",
  "rustfs-storage-api",
  "rustfs-test-utils",

crates/heal/Cargo.toml

@@ -74,6 +74,7 @@ hotpath.workspace = true
 rustfs-config = { workspace = true }
 rustfs-concurrency = { workspace = true }
 rustfs-ecstore = { workspace = true }
+rustfs-lock = { workspace = true }
 rustfs-storage-api = { workspace = true }
 rustfs-common = { workspace = true }
 rustfs-madmin = { workspace = true }

crates/heal/src/error.rs

@@ -80,10 +80,23 @@ impl Error {
     }
 
     /// Whether a heal operation can be retried without changing its inputs.
+    ///
+    /// Typed-first (backlog#1845): the primary classification reads error
+    /// variants and typed helpers (`is_quorum_error`, `LockError::is_fatal`);
+    /// the substring fallback in [`is_recoverable_heal_error_message`] only
+    /// catches errors whose typed identity was destroyed upstream.
     pub(crate) fn is_recoverable_heal(&self) -> bool {
         match self {
             Error::TaskCancelled | Error::TaskTimeout => false,
             Error::TransientSkip { .. } => true,
+            // Lock failures classify by LockError's own taxonomy: only the
+            // fatal variants (ResourceNotFound / PermissionDenied /
+            // Configuration) are terminal - retrying cannot fix them - while
+            // contention and transport variants (Timeout, Network, Internal,
+            // AlreadyLocked, QuorumNotReached, InsufficientNodes, ...) stay
+            // recoverable, as the previous blanket `Lock(_) => true` treated
+            // them.
+            Error::Storage(EcstoreError::Lock(lock_err)) => !lock_err.is_fatal(),
             Error::Storage(err) => {
                 err.is_quorum_error()
                     || matches!(
@@ -92,7 +105,7 @@ impl Error {
                             | EcstoreError::VolumeNotFound
                             | EcstoreError::SlowDown
                             | EcstoreError::OperationCanceled
-                            | EcstoreError::Lock(_)
+                            | EcstoreError::RemoteClientUnavailable(_)
                     )
                     || is_recoverable_heal_error_message(&err.to_string())
             }
@@ -106,6 +119,7 @@ impl Error {
                         | DiskError::SourceStalled
                         | DiskError::FaultyRemoteDisk
                         | DiskError::FaultyDisk
+                        | DiskError::RemoteClientUnavailable(_)
                 ) || is_recoverable_heal_error_message(&err.to_string())
             }
             Error::TaskExecutionFailed { message } | Error::Other(message) => is_recoverable_heal_error_message(message),
@@ -115,20 +129,39 @@ impl Error {
     }
 }
 
+/// Documented substring fallback for errors that reach heal with their typed
+/// identity destroyed (stringified through `TaskExecutionFailed`/`Other`, or
+/// boxed into `Io`). Every needle is annotated with the producer that emits
+/// it; when a producer becomes typed end-to-end, delete its needle here
+/// (backlog#1845 - this list only shrinks).
 fn is_recoverable_heal_error_message(error: &str) -> bool {
     let error = error.to_ascii_lowercase();
     [
+        // set_disk/ops/locking.rs ns_loc lock failures rendered into messages.
         "failed to acquire read lock",
+        // ecstore cluster/rpc/remote_locker.rs "Lock acquisition failed on
+        // remote server" and lock/src local lock responses.
         "lock acquisition failed",
+        // lock/src/distributed_lock.rs + client/local.rs LockResponse::failure.
         "lock acquisition timeout",
+        // remote_locker.rs RPC deadline wrapper.
         "remote lock rpc timed out",
+        // tokio/tonic deadline rendering, reaches heal via stringified RPC errors.
         "deadline has elapsed",
+        // generic io/tonic timeout rendering.
         "timed out",
+        // tonic transport failure rendering.
         "transport error",
+        // LockError::Network display prefix.
         "network error",
+        // io::Error ConnectionRefused rendering.
         "connection refused",
+        // StorageError::OperationCanceled rendered through task messages.
         "operation canceled",
+        // LockError::QuorumNotReached display, when stringified before typing.
         "quorum not reached",
+        // set_disk/ops/heal.rs HEAL_RENAME_INCOMPLETE - the one needle with no
+        // typed variant yet; the producer formats it into a plain message.
         "heal rename incomplete",
     ]
     .iter()
@@ -170,4 +203,69 @@ mod tests {
     fn task_timeout_is_terminal() {
         assert!(!Error::TaskTimeout.is_recoverable_heal());
     }
+
+    #[test]
+    fn lock_contention_and_transport_variants_stay_recoverable() {
+        use rustfs_lock::LockError;
+        for lock_err in [
+            LockError::Timeout {
+                resource: "bucket/object".to_string(),
+                timeout: std::time::Duration::from_secs(5),
+            },
+            LockError::Network {
+                message: "peer unreachable".to_string(),
+                source: Box::new(std::io::Error::other("reset")),
+            },
+            LockError::Internal {
+                message: "channel busy".to_string(),
+            },
+            LockError::AlreadyLocked {
+                resource: "bucket/object".to_string(),
+                owner: "node-2".to_string(),
+            },
+            LockError::QuorumNotReached {
+                required: 3,
+                achieved: 1,
+            },
+            LockError::InsufficientNodes {
+                required: 3,
+                available: 1,
+            },
+        ] {
+            assert!(
+                Error::Storage(EcstoreError::Lock(lock_err)).is_recoverable_heal(),
+                "contention/transport lock failures must stay retryable"
+            );
+        }
+    }
+
+    #[test]
+    fn fatal_lock_variants_are_terminal() {
+        use rustfs_lock::LockError;
+        for lock_err in [
+            LockError::ResourceNotFound {
+                resource: "bucket/object".to_string(),
+            },
+            LockError::PermissionDenied {
+                reason: "acl".to_string(),
+            },
+            LockError::Configuration {
+                message: "bad quorum config".to_string(),
+            },
+        ] {
+            assert!(
+                !Error::Storage(EcstoreError::Lock(lock_err)).is_recoverable_heal(),
+                "fatal lock failures cannot be fixed by retrying"
+            );
+        }
+    }
+
+    #[test]
+    fn remote_client_unavailable_is_recoverable_via_typed_variant() {
+        // The detail deliberately avoids every substring needle: the typed
+        // variant alone must classify these as retryable.
+        let detail = "auth interceptor rebuild".to_string();
+        assert!(Error::Disk(DiskError::RemoteClientUnavailable(detail.clone())).is_recoverable_heal());
+        assert!(Error::Storage(EcstoreError::RemoteClientUnavailable(detail)).is_recoverable_heal());
+    }
 }

crates/heal/src/heal/task.rs

@@ -662,6 +662,15 @@ impl HealTask {
     }
 
     fn is_transient_lock_or_timeout_error(err: &Error) -> bool {
+        // Typed-first (backlog#1845): trust the lock taxonomy and timeout
+        // variants before falling back to message needles for errors whose
+        // typed identity was stringified upstream.
+        if let Error::Storage(EcstoreError::Lock(lock_err)) = err {
+            return lock_err.is_retryable() || matches!(lock_err, rustfs_lock::LockError::QuorumNotReached { .. });
+        }
+        if matches!(err, Error::Disk(DiskError::Timeout) | Error::Storage(EcstoreError::Timeout)) {
+            return true;
+        }
         let message = err.to_string().to_ascii_lowercase();
         message.contains("lock acquisition timeout")
             || message.contains("lock acquisition failed")