diff --git a/Cargo.lock b/Cargo.lock index c46d9fed3..75977b16f 100644 --- a/Cargo.lock +++ b/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", diff --git a/crates/heal/Cargo.toml b/crates/heal/Cargo.toml index aabd39a6f..3d101b9c9 100644 --- a/crates/heal/Cargo.toml +++ b/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 } diff --git a/crates/heal/src/error.rs b/crates/heal/src/error.rs index 7fa8d7717..a03d14663 100644 --- a/crates/heal/src/error.rs +++ b/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::()) 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 { diff --git a/crates/heal/src/heal/task/tests.rs b/crates/heal/src/heal/task/tests.rs index 1e93246c8..f184a5d56 100644 --- a/crates/heal/src/heal/task/tests.rs +++ b/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] {