fix(heal): retry interrupted internode shard writes

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",

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] {