fix: recover interrupted GETs and heal operations (#7876)

* fix(heal): retry interrupted internode shard writes

* fix(deps): update rustls for RUSTSEC-2026-0285

* fix(heal): retain RPC transport failures across metadata probes

* test(heal): wait for replacement disk and coordinator readiness

* test(connect): validate service-only diagnostic capabilities

* fix(s3): resume GET bodies after erasure read quorum loss

Cargo.lock

@@ -9939,8 +9939,10 @@ dependencies = [
  "rustfs-config",
  "rustfs-ecstore",
  "rustfs-heal-contracts",
+ "rustfs-io-metrics",
  "rustfs-lock",
  "rustfs-madmin",
+ "rustfs-rio",
  "rustfs-storage-api",
  "rustfs-test-utils",
  "rustfs-utils",
@@ -11085,9 +11087,9 @@ dependencies = [
 
 [[package]]
 name = "rustls"
-version = "0.23.44"
+version = "0.23.45"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6725596c3f2c3a0aef021139e145d4eafe314a6623e4680ca83852b2c67ab2ba"
+checksum = "0d41d731c7d2f962d1ccc364cec258de3c0e93b38c2fb3ba97ac74513048d634"
 dependencies = [
  "aws-lc-rs",
  "log",

Cargo.toml

@@ -211,7 +211,7 @@ openidconnect = { default-features = false, version = "4.0" }
 pbkdf2 = "0.13.0"
 p256 = { version = "0.14.0", features = ["ecdsa", "pkcs8"] }
 rsa = { version = "=0.10.0-rc.18" }
-rustls = { default-features = false, version = "0.23.44" }
+rustls = { default-features = false, version = "0.23.45" }
 rustls-native-certs = "0.8"
 rustls-pki-types = "1.15.1"
 x509-parser = "0.18.1"

crates/e2e_test/src/chaos.rs

@@ -374,6 +374,30 @@ pub(crate) fn census_object_version_on_disk(
     })
 }
 
+pub(crate) fn is_cluster_heal_coordination_unavailable(error: &(dyn std::error::Error + Send + Sync)) -> bool {
+    let message = error.to_string();
+    message.contains("500 Internal Server Error") && message.contains("cluster heal coordination unavailable")
+}
+
+/// Wait for the restarted cluster to admit the first root heal request.
+pub(crate) async fn start_root_heal_when_control_ready(
+    heal_url: &str,
+    heal_body: &str,
+    access_key: &str,
+    secret_key: &str,
+) -> ChaosResult<()> {
+    let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(45);
+    loop {
+        match signed_admin_post(heal_url, Some(heal_body), access_key, secret_key).await {
+            Ok(_) => return Ok(()),
+            Err(error) if is_cluster_heal_coordination_unavailable(error.as_ref()) && tokio::time::Instant::now() < deadline => {
+                tokio::time::sleep(std::time::Duration::from_millis(250)).await;
+            }
+            Err(error) => return Err(error),
+        }
+    }
+}
+
 /// `POST` a signed (SigV4, service `s3`) admin request without relying on the
 /// external `awscurl` binary. Mirrors the admin heal calls used by the heal
 /// regression suite.

crates/e2e_test/src/distributed/heal_test.rs

@@ -16,7 +16,7 @@ use super::harness::{
     DistCluster, DistLayout, TestResult, assert_inventory, get_object_bytes, payload_for, put_object, sha256_hex, unique_bucket,
     wait_until,
 };
-use crate::chaos::{VersionShardCensus, census_object_version_on_disk, signed_admin_post};
+use crate::chaos::{VersionShardCensus, census_object_version_on_disk, start_root_heal_when_control_ready};
 use crate::common::init_logging;
 use aws_sdk_s3::Client;
 use aws_sdk_s3::primitives::ByteStream;
@@ -26,7 +26,6 @@ use std::collections::{BTreeMap, HashSet};
 use std::io::{Read, Write};
 use std::path::{Path, PathBuf};
 use std::time::Duration;
-use tokio::time::{Instant, sleep};
 
 const EC84_NODE_COUNT: usize = 3;
 const EC84_DRIVES_PER_NODE: usize = 4;
@@ -34,8 +33,6 @@ const EC84_DATA_BLOCKS: usize = 8;
 const EC84_PARITY_BLOCKS: usize = 4;
 const EC84_TARGET_DRIVE_RESTART_CASE: &str = "ec84-target-drive-restart";
 const EC84_TARGET_DRIVE_RESTART_ORACLE: &str = "ec84-target-drive-restart.json";
-const EC84_HEAL_CONTROL_READY_TIMEOUT: Duration = Duration::from_secs(45);
-const EC84_HEAL_CONTROL_RETRY_DELAY: Duration = Duration::from_millis(250);
 
 #[derive(Clone)]
 struct ExpectedShard {
@@ -214,29 +211,6 @@ fn assert_replaced_drive_empty(drive: &Path, bucket: &str, keys: &[String]) -> T
     Ok(())
 }
 
-fn is_cluster_heal_coordination_unavailable(error: &(dyn std::error::Error + Send + Sync)) -> bool {
-    let message = error.to_string();
-    message.contains("500 Internal Server Error") && message.contains("cluster heal coordination unavailable")
-}
-
-async fn start_ec84_root_heal_when_control_ready(
-    heal_url: &str,
-    heal_body: &str,
-    access_key: &str,
-    secret_key: &str,
-) -> TestResult {
-    let deadline = Instant::now() + EC84_HEAL_CONTROL_READY_TIMEOUT;
-    loop {
-        match signed_admin_post(heal_url, Some(heal_body), access_key, secret_key).await {
-            Ok(_) => return Ok(()),
-            Err(error) if is_cluster_heal_coordination_unavailable(error.as_ref()) && Instant::now() < deadline => {
-                sleep(EC84_HEAL_CONTROL_RETRY_DELAY).await;
-            }
-            Err(error) => return Err(error),
-        }
-    }
-}
-
 async fn put_large_inventory(client: &Client, bucket: &str) -> TestResult<Vec<ExpectedShard>> {
     let mut expected = Vec::new();
     for index in 0..4 {
@@ -330,7 +304,7 @@ async fn three_node_four_drive_ec8_4_root_heal_rebuilds_replaced_drive_after_res
     let heal_body =
         r#"{"recursive":true,"dryRun":false,"remove":false,"recreate":true,"scanMode":2,"updateParity":false,"nolock":false}"#;
     let heal_url = format!("{}/rustfs/admin/v3/heal/{bucket}?forceStart=true", dist.cluster.nodes[0].url);
-    start_ec84_root_heal_when_control_ready(&heal_url, heal_body, &dist.cluster.access_key, &dist.cluster.secret_key).await?;
+    start_root_heal_when_control_ready(&heal_url, heal_body, &dist.cluster.access_key, &dist.cluster.secret_key).await?;
 
     wait_until(
         Duration::from_secs(120),
@@ -394,7 +368,7 @@ async fn three_node_four_drive_ec8_4_root_heal_rebuilds_replaced_drive_after_res
 
 #[cfg(test)]
 mod tests {
-    use super::*;
+    use crate::chaos::is_cluster_heal_coordination_unavailable;
 
     #[test]
     fn cluster_heal_coordination_retry_is_exact() {

crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs

@@ -16,7 +16,9 @@
 
 #[cfg(test)]
 mod tests {
-    use crate::chaos::{VersionShardCensus, census_object_version_on_disk, sha256_hex, signed_admin_post};
+    use crate::chaos::{
+        VersionShardCensus, census_object_version_on_disk, sha256_hex, signed_admin_post, start_root_heal_when_control_ready,
+    };
     use crate::common::{
         ClusterTopology, FAST_DATA_USAGE_SCANNER_ENV, RustFSTestClusterEnvironment, RustFSTestEnvironment, admin_request,
         init_logging, rustfs_binary_path,
@@ -960,19 +962,26 @@ mod tests {
 
         let online_key = "cluster/online-before-replacement.bin";
         let online_body = b"object written while all cluster nodes are online".to_vec();
-        clients[0]
-            .put_object()
-            .bucket(bucket)
-            .key(online_key)
-            .body(ByteStream::from(online_body.clone()))
-            .send()
-            .await?;
-
         let replaced_disk = PathBuf::from(&cluster.nodes[1].data_dir);
-        assert!(
-            object_metadata_exists_on_disk(&replaced_disk, bucket, online_key),
-            "node 1 should contain metadata before disk replacement"
-        );
+        // A quorum write need not include the disk this fixture will replace.
+        // Establish that disk's baseline before testing its reconstruction.
+        timeout(Duration::from_secs(30), async {
+            loop {
+                clients[0]
+                    .put_object()
+                    .bucket(bucket)
+                    .key(online_key)
+                    .body(ByteStream::from(online_body.clone()))
+                    .send()
+                    .await?;
+                if object_metadata_exists_on_disk(&replaced_disk, bucket, online_key) {
+                    return Ok::<_, Box<dyn Error + Send + Sync>>(());
+                }
+                sleep(Duration::from_millis(100)).await;
+            }
+        })
+        .await
+        .map_err(|_| "node 1 did not store the baseline object before disk replacement")??;
 
         cluster.stop_node(1)?;
         std::fs::remove_dir_all(&replaced_disk)?;
@@ -1017,7 +1026,7 @@ mod tests {
 
         let heal_body = r#"{"recursive":true,"dryRun":false,"remove":false,"recreate":true,"scanMode":2,"updateParity":false,"nolock":false}"#;
         let heal_url = format!("{}/rustfs/admin/v3/heal/?forceStart=true", cluster.nodes[0].url);
-        signed_admin_post(&heal_url, Some(heal_body), &cluster.access_key, &cluster.secret_key).await?;
+        start_root_heal_when_control_ready(&heal_url, heal_body, &cluster.access_key, &cluster.secret_key).await?;
 
         let expected_objects = [(online_key, online_body.as_slice()), (outage_key, outage_body.as_slice())];
         let mut remaining_rebuild_keys: HashSet<&str> = expected_objects.iter().map(|(key, _)| *key).collect();

crates/ecstore/src/disk/error.rs

@@ -627,13 +627,18 @@ impl From<tokio::task::JoinError> for DiskError {
 impl Clone for DiskError {
     fn clone(&self) -> Self {
         match self {
-            DiskError::Io(io_error) => DiskError::Io(
-                rustfs_rio::clone_internode_http_io_error(io_error)
-                    .and_then(std::io::Error::into_inner)
-                    // The helper derives a kind from the source; Clone must retain the original outer kind.
-                    .map(|source| std::io::Error::new(io_error.kind(), source))
-                    .unwrap_or_else(|| std::io::Error::new(io_error.kind(), io_error.to_string())),
-            ),
+            DiskError::Io(io_error) => {
+                if let Some(status) = io_error.get_ref().and_then(|source| source.downcast_ref::<RpcStatusError>()) {
+                    return DiskError::Io(io::Error::new(io_error.kind(), RpcStatusError(status.0.clone())));
+                }
+                DiskError::Io(
+                    rustfs_rio::clone_internode_http_io_error(io_error)
+                        .and_then(std::io::Error::into_inner)
+                        // The helper derives a kind from the source; Clone must retain the original outer kind.
+                        .map(|source| std::io::Error::new(io_error.kind(), source))
+                        .unwrap_or_else(|| std::io::Error::new(io_error.kind(), io_error.to_string())),
+                )
+            }
             DiskError::MaxVersionsExceeded => DiskError::MaxVersionsExceeded,
             DiskError::Unexpected => DiskError::Unexpected,
             DiskError::CorruptedFormat => DiskError::CorruptedFormat,

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

@@ -36,6 +36,20 @@ const EVENT_HEAL_OBJECT_RENAME: &str = "heal_object_rename";
 const HEAL_RENAME_INCOMPLETE: &str = "heal rename incomplete";
 const READ_REPAIR_DATA_PHASE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60 * 60);
 
+fn is_unavailable_heal_rpc(error: &DiskError) -> bool {
+    let DiskError::Io(error) = error else {
+        return false;
+    };
+    let Some(status) = crate::cluster::rpc::client::embedded_tonic_status(error) else {
+        return false;
+    };
+    // Tonic can report a locally observed HTTP/2 GOAWAY as Internal. Only a
+    // typed transport source identifies that case; peer-supplied text cannot.
+    status.code() == tonic::Code::Unavailable
+        || (matches!(status.code(), tonic::Code::Internal | tonic::Code::Unknown)
+            && std::error::Error::source(status).is_some_and(|source| source.is::<tonic::transport::Error>()))
+}
+
 fn heal_drive_state_for_error(error: &DiskError) -> DriveState {
     match error {
         DiskError::DiskNotFound | DiskError::RemoteClientUnavailable(_) => DriveState::Offline,
@@ -46,6 +60,7 @@ fn heal_drive_state_for_error(error: &DiskError) -> DriveState {
         | DiskError::PartMissingOrCorrupt
         | DiskError::OutdatedXLMeta => DriveState::Missing,
         DiskError::FileCorrupt => DriveState::Corrupt,
+        _ if is_unavailable_heal_rpc(error) => DriveState::Offline,
         _ => DriveState::Unknown(error.to_string()),
     }
 }
@@ -2683,6 +2698,32 @@ mod heal_result_report_tests {
         );
     }
 
+    #[tokio::test]
+    async fn heal_preserves_rpc_transport_failure_through_metadata_classification() {
+        let transport = tonic::transport::Endpoint::from_static("http://127.0.0.1:0")
+            .connect()
+            .await
+            .expect_err("port zero cannot serve an internode connection");
+        let mut interrupted = tonic::Status::internal("h2 protocol error: http2 error");
+        interrupted.set_source(Arc::new(transport));
+        for (status, offline) in [
+            (interrupted, true),
+            (tonic::Status::unavailable("peer restarting"), true),
+            (tonic::Status::internal("h2 protocol error: http2 error"), false),
+            (tonic::Status::permission_denied("connection reset"), false),
+            (tonic::Status::data_loss("broken pipe"), false),
+        ] {
+            let error = DiskError::from(status);
+            let (_, _, reason) = super::should_heal_object_on_disk(&Some(error), &[], &FileInfo::default(), &FileInfo::default());
+            let reason = reason.expect("metadata probe failure must survive classification");
+            assert_eq!(
+                matches!(super::heal_drive_state_for_error(&reason), DriveState::Offline),
+                offline,
+                "{reason:?}"
+            );
+        }
+    }
+
     #[test]
     fn read_repair_commit_fingerprint_tracks_commit_identity_only() {
         let data_dir = Uuid::parse_str("11111111-1111-1111-1111-111111111111").expect("data dir should parse");

crates/heal/Cargo.toml

@@ -76,6 +76,8 @@ rustfs-config = { workspace = true }
 rustfs-concurrency = { workspace = true }
 rustfs-ecstore = { workspace = true }
 rustfs-lock = { workspace = true }
+rustfs-io-metrics = { workspace = true }
+rustfs-rio = { workspace = true }
 rustfs-storage-api = { workspace = true }
 rustfs-common = { workspace = true }
 rustfs-heal-contracts = { workspace = true }

crates/heal/src/error.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use rustfs_io_metrics::internode_metrics::INTERNODE_OPERATION_PUT_FILE_STREAM;
+use rustfs_rio::{InternodeHttpError, InternodeHttpErrorKind};
 use thiserror::Error;
 
 use super::heal::{DiskError, EcstoreError};
@@ -113,6 +115,7 @@ impl Error {
                     return true;
                 }
                 err.is_quorum_error()
+                    || matches!(err, EcstoreError::Io(error) if is_recoverable_internode_error(error))
                     || matches!(
                         err,
                         EcstoreError::DiskNotFound
@@ -137,10 +140,11 @@ impl Error {
                         | DiskError::FaultyRemoteDisk
                         | DiskError::FaultyDisk
                         | DiskError::RemoteClientUnavailable(_)
-                ) || is_recoverable_heal_error_message(&err.to_string())
+                ) || matches!(err, DiskError::Io(error) if is_recoverable_internode_error(error))
+                    || is_recoverable_heal_error_message(&err.to_string())
             }
             Error::TaskExecutionFailed { message } | Error::Other(message) => is_recoverable_heal_error_message(message),
-            Error::Io(err) => is_recoverable_heal_error_message(&err.to_string()),
+            Error::Io(err) => is_recoverable_internode_error(err) || is_recoverable_heal_error_message(&err.to_string()),
             _ => false,
         }
     }
@@ -158,6 +162,18 @@ impl Error {
     }
 }
 
+fn is_recoverable_internode_error(error: &std::io::Error) -> bool {
+    let Some(error) = error.get_ref().and_then(|source| source.downcast_ref::<InternodeHttpError>()) else {
+        return false;
+    };
+    // A restarting peer can return 500 after admitting a shard upload. Heal
+    // can replay that write within its existing object and task retry budgets.
+    // Other operations retain the transport's normal retry classification.
+    error.kind().is_retryable()
+        || (error.context().operation() == Some(INTERNODE_OPERATION_PUT_FILE_STREAM)
+            && matches!(error.kind(), InternodeHttpErrorKind::HttpStatus(status) if matches!(status.as_u16(), 409 | 500)))
+}
+
 /// 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
@@ -208,6 +224,79 @@ mod tests {
     use super::Error;
     use crate::heal::{DiskError, EcstoreError};
 
+    #[test]
+    fn internode_transport_errors_keep_their_recovery_classification() {
+        use rustfs_rio::{InternodeHttpErrorKind as Kind, new_test_internode_http_io_error};
+        for (kind, expected) in [
+            (Kind::ConnectionReset, true),
+            (Kind::BodyStreamAborted, true),
+            (Kind::DnsResolutionFailed, true),
+            (Kind::HttpStatus(http::StatusCode::INTERNAL_SERVER_ERROR), true),
+            (Kind::HttpStatus(http::StatusCode::CONFLICT), true),
+            (Kind::HttpStatus(http::StatusCode::SERVICE_UNAVAILABLE), true),
+            (Kind::HttpStatus(http::StatusCode::FORBIDDEN), false),
+            (Kind::HttpStatus(http::StatusCode::BAD_REQUEST), false),
+            (Kind::HttpStatus(http::StatusCode::NOT_FOUND), false),
+            (Kind::Unknown, false),
+        ] {
+            for error in [
+                Error::Io(new_test_internode_http_io_error(kind)),
+                Error::Disk(DiskError::from(new_test_internode_http_io_error(kind))),
+                Error::Storage(EcstoreError::from(DiskError::from(new_test_internode_http_io_error(kind)))),
+            ] {
+                assert_eq!(error.is_recoverable_heal(), expected, "{error:?}");
+            }
+        }
+        let text = "internode http status 500 Internal Server Error: PUT /rustfs/rpc/put_file_stream_v1";
+        assert!(!Error::Io(std::io::Error::other(text)).is_recoverable_heal());
+        assert!(!Error::other(text).is_recoverable_heal());
+        for error in [DiskError::FileCorrupt, DiskError::DiskFull, DiskError::FileAccessDenied] {
+            assert!(!Error::Disk(error).is_recoverable_heal());
+        }
+    }
+
+    #[tokio::test]
+    async fn read_http_500_is_not_a_retryable_heal_write() {
+        use tokio::io::{AsyncReadExt, AsyncWriteExt};
+        tokio::time::timeout(std::time::Duration::from_secs(5), async {
+            let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.expect("bind fixture");
+            let address = listener.local_addr().expect("fixture address");
+            let server = tokio::spawn(async move {
+                let (mut stream, _) = listener.accept().await.expect("accept request");
+                let mut request = [0_u8; 4096];
+                let mut read = 0;
+                loop {
+                    let count = stream.read(&mut request[read..]).await.expect("request headers");
+                    assert!(count > 0, "request ended before headers");
+                    read += count;
+                    if request[..read].windows(4).any(|bytes| bytes == b"\r\n\r\n") {
+                        break;
+                    }
+                    assert!(read < request.len(), "request exceeds fixture budget");
+                }
+                stream
+                    .write_all(b"HTTP/1.1 500 Internal Server Error\r\nContent-Length: 0\r\nConnection: close\r\n\r\n")
+                    .await
+                    .expect("write response");
+            });
+            let error = match rustfs_rio::HttpReader::new(
+                format!("http://{address}/rustfs/rpc/read_file_stream"),
+                http::Method::GET,
+                http::HeaderMap::new(),
+                None,
+            )
+            .await
+            {
+                Ok(_) => panic!("HTTP 500 must fail the read"),
+                Err(error) => Error::Storage(EcstoreError::from(DiskError::from(error))),
+            };
+            server.await.expect("fixture completed");
+            assert!(!error.is_recoverable_heal(), "{error:?}");
+        })
+        .await
+        .expect("fixture completed within its budget");
+    }
+
     #[test]
     fn incomplete_target_rename_is_recoverable() {
         let task_error = Error::TaskExecutionFailed {

crates/heal/src/heal/task/tests.rs

@@ -1757,6 +1757,7 @@ enum MockHealObjectOutcome {
     DanglingGraceDeferred,
     UnavailableDrive(DriveState),
     RetryableReadQuorum,
+    InternodeHttp(http::StatusCode),
     RetryableSlowDown,
     PermanentOther(&'static str),
 }
@@ -1891,6 +1892,9 @@ impl HealStorageAPI for MockStorage {
                     ))),
                 )),
                 MockHealObjectOutcome::UnavailableDrive(state) => Ok(unavailable_drive_heal_result(state)),
+                MockHealObjectOutcome::InternodeHttp(status) => Err(Error::Storage(EcstoreError::from(DiskError::from(
+                    rustfs_rio::new_test_internode_http_io_error(rustfs_rio::InternodeHttpErrorKind::HttpStatus(status)),
+                )))),
                 MockHealObjectOutcome::RetryableReadQuorum => Err(Error::Storage(EcstoreError::InsufficientReadQuorum(
                     bucket.to_string(),
                     object.to_string(),
@@ -1937,6 +1941,9 @@ impl HealStorageAPI for MockStorage {
                 MockHealObjectOutcome::ErrOther(message) | MockHealObjectOutcome::PermanentOther(message) => {
                     Err(Error::other(message))
                 }
+                MockHealObjectOutcome::InternodeHttp(status) => Err(Error::Storage(EcstoreError::from(DiskError::from(
+                    rustfs_rio::new_test_internode_http_io_error(rustfs_rio::InternodeHttpErrorKind::HttpStatus(status)),
+                )))),
                 MockHealObjectOutcome::RetryableReadQuorum => Err(Error::Storage(EcstoreError::InsufficientReadQuorum(
                     bucket.to_string(),
                     object.to_string(),
@@ -2736,6 +2743,70 @@ async fn test_recursive_bucket_heal_retries_only_retryable_objects() {
     assert_eq!(progress.objects_failed, 0);
 }
 
+#[tokio::test(start_paused = true)]
+async fn recursive_bucket_heal_retries_interrupted_internode_write() {
+    let storage = Arc::new(MockStorage::default());
+    storage.heal_object_outcomes.lock().unwrap().insert(
+        "object-a".to_string(),
+        VecDeque::from([MockHealObjectOutcome::InternodeHttp(http::StatusCode::INTERNAL_SERVER_ERROR)]),
+    );
+    let task = HealTask::from_request(
+        HealRequest::new(
+            HealType::Bucket {
+                bucket: "bucket-a".to_string(),
+            },
+            HealOptions {
+                recursive: true,
+                timeout: None,
+                ..Default::default()
+            },
+            HealPriority::Normal,
+        ),
+        storage.clone(),
+    );
+    task.heal_bucket("bucket-a").await.expect("interrupted write must recover");
+    assert_eq!(storage.heal_object_calls.lock().unwrap().as_slice(), ["object-a", "object-b", "object-a"]);
+    let progress = task.get_progress().await;
+    assert_eq!((progress.objects_scanned, progress.objects_healed, progress.objects_failed), (2, 2, 0));
+}
+
+#[tokio::test(start_paused = true)]
+async fn recursive_bucket_heal_bounds_internode_retries_and_keeps_auth_failures_terminal() {
+    let storage = Arc::new(MockStorage::default());
+    storage.heal_object_outcomes.lock().unwrap().insert(
+        "object-a".to_string(),
+        (0..4)
+            .map(|_| MockHealObjectOutcome::InternodeHttp(http::StatusCode::INTERNAL_SERVER_ERROR))
+            .collect(),
+    );
+    storage.heal_object_outcomes.lock().unwrap().insert(
+        "object-b".to_string(),
+        VecDeque::from([MockHealObjectOutcome::InternodeHttp(http::StatusCode::FORBIDDEN)]),
+    );
+    let task = HealTask::from_request(
+        HealRequest::new(
+            HealType::Bucket {
+                bucket: "bucket-a".to_string(),
+            },
+            HealOptions {
+                recursive: true,
+                timeout: None,
+                ..Default::default()
+            },
+            HealPriority::Normal,
+        ),
+        storage.clone(),
+    );
+    task.heal_bucket("bucket-a")
+        .await
+        .expect_err("persistent and forbidden writes must fail");
+    let failure = task.take_batch_failure().await.expect("retain failure details");
+    assert_eq!((failure.failed, failure.retryable, failure.permanent), (2, 1, 1));
+    let calls = storage.heal_object_calls.lock().unwrap();
+    assert_eq!(calls.iter().filter(|object| object.as_str() == "object-a").count(), 4);
+    assert_eq!(calls.iter().filter(|object| object.as_str() == "object-b").count(), 1);
+}
+
 #[tokio::test(start_paused = true)]
 async fn recursive_bucket_heal_retries_when_recreate_target_is_unavailable() {
     for state in [DriveState::Offline, DriveState::Faulty] {

rustfs/src/app/object/get.rs

@@ -1946,14 +1946,19 @@ fn get_object_resume_control(ctx: GetObjectResumeContext) -> GetObjectResumeCont
     )
 }
 
-/// Mid-stream errors that mean the pinned object data is gone (rebalance or
-/// decommission removed it after copying the version elsewhere). Only typed
+/// Mid-stream errors that mean the current read lost access to its source
+/// shards, whether through relocation or unavailable peers. Only typed
 /// `StorageError`s qualify; generic I/O errors and string-matched "not enough
 /// disks" failures keep the existing fail-loud behavior.
 fn is_object_relocation_error(err: &std::io::Error) -> bool {
     let Some(inner) = err.get_ref() else { return false };
     match inner.downcast_ref::<StorageError>() {
-        Some(StorageError::FileNotFound | StorageError::ObjectNotFound(..) | StorageError::InsufficientReadQuorum(..)) => true,
+        Some(
+            StorageError::FileNotFound
+            | StorageError::ObjectNotFound(..)
+            | StorageError::InsufficientReadQuorum(..)
+            | StorageError::ErasureReadQuorum,
+        ) => true,
         Some(StorageError::Io(source)) => source.kind() == std::io::ErrorKind::NotFound,
         _ => false,
     }
@@ -7793,6 +7798,7 @@ mod tests {
             StorageError::FileNotFound,
             StorageError::ObjectNotFound("test-bucket".to_string(), "relocated-object".to_string()),
             StorageError::InsufficientReadQuorum("test-bucket".to_string(), "relocated-object".to_string()),
+            StorageError::ErasureReadQuorum,
             StorageError::Io(std::io::Error::new(std::io::ErrorKind::NotFound, "relocated shard disappeared")),
         ] {
             let reopen_count = Arc::new(AtomicUsize::new(0));
@@ -8145,30 +8151,41 @@ mod tests {
     async fn get_object_streaming_reader_non_relocation_error_passes_through() {
         use tokio::io::AsyncReadExt;
 
-        let reopen_count = Arc::new(AtomicUsize::new(0));
-        let control = counting_resume_control(Arc::clone(&reopen_count), |_| {
-            panic!("a non-relocation read error must not reopen");
-        });
-        let mut reader = GetObjectStreamingReader::new(
-            FailAtEndReader::new(b"hello ", Some(std::io::Error::new(std::io::ErrorKind::InvalidData, "corrupt"))),
-            "test-bucket",
-            "corrupt-object",
-            "req-resume-passthrough",
-            None,
-            11,
-            Duration::ZERO,
-            GetObjectBodyLifecycle::disabled(),
-            Some(control),
-        );
-        let mut out = Vec::new();
-        let err = reader
-            .read_to_end(&mut out)
-            .await
-            .expect_err("a non-relocation error must fail the body unchanged");
+        for error in [
+            std::io::Error::new(std::io::ErrorKind::InvalidData, "corrupt"),
+            std::io::Error::other(StorageError::FileCorrupt),
+            std::io::Error::other(StorageError::FileAccessDenied),
+            std::io::Error::other(StorageError::ErasureWriteQuorum),
+            std::io::Error::other("erasure read quorum"),
+        ] {
+            let kind = error.kind();
+            let message = error.to_string();
+            let reopen_count = Arc::new(AtomicUsize::new(0));
+            let control = counting_resume_control(Arc::clone(&reopen_count), |_| {
+                panic!("a non-relocation read error must not reopen");
+            });
+            let mut reader = GetObjectStreamingReader::new(
+                FailAtEndReader::new(b"hello ", Some(error)),
+                "test-bucket",
+                "corrupt-object",
+                "req-resume-passthrough",
+                None,
+                11,
+                Duration::ZERO,
+                GetObjectBodyLifecycle::disabled(),
+                Some(control),
+            );
+            let mut out = Vec::new();
+            let err = reader
+                .read_to_end(&mut out)
+                .await
+                .expect_err("a non-relocation error must fail the body unchanged");
 
-        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
-        assert_eq!(out, b"hello ");
-        assert_eq!(reopen_count.load(Ordering::Relaxed), 0);
+            assert_eq!(err.kind(), kind);
+            assert_eq!(err.to_string(), message);
+            assert_eq!(out, b"hello ");
+            assert_eq!(reopen_count.load(Ordering::Relaxed), 0);
+        }
     }
 
     #[tokio::test]

rustfs/src/config/cli.rs

@@ -1633,49 +1633,10 @@ mod tests {
         Cli, Commands, ConnectCommands, ConnectInventoryCommands, ConnectLicenseCommands, ConnectRelayMaterialKind,
         ConnectReportCommands, InspectCommands, preprocess_args_for_legacy,
     };
-    use crate::connect::CONNECT_DIAGNOSTIC_CAPABILITIES;
     use crate::version;
     use clap::error::ErrorKind;
     use clap::{CommandFactory, Parser};
 
-    #[test]
-    fn advertised_diagnostic_capabilities_have_cli_dispatch() {
-        let expected = [
-            ("performance.client@1", &["performance", "client"][..]),
-            ("performance.drive@1", &["performance", "drive"][..]),
-            ("performance.object@1", &["performance", "object"][..]),
-            ("performance.siteReplication@1", &["performance", "site-replication"][..]),
-            ("logs.capture@1", &["logs"][..]),
-            ("profile.cpu@1", &["profile"][..]),
-            ("profile.memory@1", &["profile"][..]),
-            ("profile.threads@1", &["profile"][..]),
-            ("telemetry.record@1", &["telemetry", "record"][..]),
-            ("telemetry.otlp@1", &["telemetry", "otlp"][..]),
-            ("telemetry.replay@1", &["telemetry", "replay"][..]),
-            ("top.api@1", &["top", "api"][..]),
-            ("top.disk@1", &["top", "disk"][..]),
-            ("top.locks@1", &["top", "locks"][..]),
-            ("top.net@1", &["top", "net"][..]),
-            ("top.rpc@1", &["top", "rpc"][..]),
-            ("inspect.object@1", &["inspect", "object"][..]),
-        ];
-        assert_eq!(
-            CONNECT_DIAGNOSTIC_CAPABILITIES,
-            expected.iter().map(|(capability, _)| *capability).collect::<Vec<_>>()
-        );
-
-        let command = Cli::command();
-        let connect = command.find_subcommand("connect").expect("connect command");
-        for (capability, path) in expected {
-            let mut command = connect;
-            for segment in path {
-                command = command
-                    .find_subcommand(segment)
-                    .unwrap_or_else(|| panic!("{capability} is missing CLI dispatch at {segment}"));
-            }
-        }
-    }
-
     #[test]
     fn preprocess_help_command_displays_top_level_help() {
         let args = preprocess_args_for_legacy(vec!["rustfs".to_string(), "help".to_string()]);

rustfs/src/config/mod.rs

@@ -50,6 +50,8 @@ mod snapshot;
 mod config_test;
 
 // Re-export public types
+#[cfg(test)]
+pub(crate) use cli::Cli;
 pub use cli::ConnectSiteReplicationPerformanceOpts;
 pub use cli::{CommandResult, InfoOpts, InfoType};
 pub use cli::{

rustfs/src/connect/diagnostics/job.rs

@@ -1205,6 +1205,58 @@ mod tests {
         }
     }
 
+    #[test]
+    fn advertised_diagnostic_capabilities_have_execution_paths() {
+        use crate::config::Cli;
+        use clap::CommandFactory;
+
+        let expected = [
+            ("performance.client@1", &["performance", "client"][..]),
+            ("performance.drive@1", &["performance", "drive"][..]),
+            ("performance.network@1", &[][..]),
+            ("performance.object@1", &["performance", "object"][..]),
+            ("performance.siteReplication@1", &["performance", "site-replication"][..]),
+            ("logs.capture@1", &["logs"][..]),
+            ("profile.cpu@1", &["profile"][..]),
+            ("profile.memory@1", &["profile"][..]),
+            ("profile.threads@1", &["profile"][..]),
+            ("telemetry.record@1", &["telemetry", "record"][..]),
+            ("telemetry.otlp@1", &["telemetry", "otlp"][..]),
+            ("telemetry.replay@1", &["telemetry", "replay"][..]),
+            ("top.api@1", &["top", "api"][..]),
+            ("top.disk@1", &["top", "disk"][..]),
+            ("top.locks@1", &["top", "locks"][..]),
+            ("top.net@1", &["top", "net"][..]),
+            ("top.rpc@1", &["top", "rpc"][..]),
+            ("inspect.object@1", &["inspect", "object"][..]),
+        ];
+        assert_eq!(
+            super::super::CONNECT_DIAGNOSTIC_CAPABILITIES,
+            expected.iter().map(|(capability, _)| *capability).collect::<Vec<_>>()
+        );
+
+        let command = Cli::command();
+        let connect = command.find_subcommand("connect").expect("connect command");
+        for (capability, path) in expected {
+            if path.is_empty() {
+                // Network probes use the authenticated service dispatcher and
+                // locally resolved peers, not a standalone CLI command.
+                let mut job = envelope();
+                job.job_type = PERFORMANCE_NETWORK_JOB_TYPE.to_owned();
+                job.required_capabilities = vec![capability.to_owned()];
+                job.schema_version = NETWORK_SCHEMA_VERSION;
+                assert_eq!(job.kind(), Ok(DiagnosticJobKind::PerformanceNetwork));
+                continue;
+            }
+            let mut command = connect;
+            for segment in path {
+                command = command
+                    .find_subcommand(segment)
+                    .unwrap_or_else(|| panic!("{capability} is missing CLI dispatch at {segment}"));
+            }
+        }
+    }
+
     #[test]
     fn accepts_a_bounded_signed_profile_job_for_the_exact_device() {
         let (envelope, signer) = signed();

rustfs/src/connect/diagnostics/mod.rs

@@ -38,7 +38,7 @@ mod trace_replay;
 #[cfg(unix)]
 mod trace_runtime;
 
-/// Exact signed diagnostic producers dispatched by the `rustfs connect` CLI.
+/// Signed diagnostic producers available through the CLI or authenticated service jobs.
 pub const CONNECT_DIAGNOSTIC_CAPABILITIES: &[&str] = &[
     perf_client::CLIENT_CAPABILITY,
     perf_drive::DRIVE_CAPABILITY,

rustfs/tests/connect_heartbeat.rs

@@ -538,6 +538,7 @@ async fn sends_only_l0_fields_and_accepts_additive_response_fields() {
             "inventory.environment@1",
             "performance.client@1",
             "performance.drive@1",
+            "performance.network@1",
             "performance.object@1",
             "performance.siteReplication@1",
             "logs.capture@1",