From a8302b511c6ed7e2dbceefab61ee7d0d57bf4460 Mon Sep 17 00:00:00 2001 From: Chris Date: Tue, 15 Sep 2026 03:11:18 +0800 Subject: [PATCH] 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 | 6 +- Cargo.toml | 2 +- crates/e2e_test/src/chaos.rs | 24 +++++ crates/e2e_test/src/distributed/heal_test.rs | 32 +------ .../src/heal_erasure_disk_rebuild_test.rs | 37 +++++--- crates/ecstore/src/disk/error.rs | 19 ++-- crates/ecstore/src/set_disk/ops/heal.rs | 41 ++++++++ crates/heal/Cargo.toml | 2 + crates/heal/src/error.rs | 93 ++++++++++++++++++- crates/heal/src/heal/task/tests.rs | 71 ++++++++++++++ rustfs/src/app/object/get.rs | 69 ++++++++------ rustfs/src/config/cli.rs | 39 -------- rustfs/src/config/mod.rs | 2 + rustfs/src/connect/diagnostics/job.rs | 52 +++++++++++ rustfs/src/connect/diagnostics/mod.rs | 2 +- rustfs/tests/connect_heartbeat.rs | 1 + 16 files changed, 371 insertions(+), 121 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index c46d9fed3..03c42c8c6 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", @@ -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", diff --git a/Cargo.toml b/Cargo.toml index 7ca61412c..8eb757686 100644 --- a/Cargo.toml +++ b/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" diff --git a/crates/e2e_test/src/chaos.rs b/crates/e2e_test/src/chaos.rs index f0b2971e9..228a78342 100644 --- a/crates/e2e_test/src/chaos.rs +++ b/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. diff --git a/crates/e2e_test/src/distributed/heal_test.rs b/crates/e2e_test/src/distributed/heal_test.rs index d09c0a32e..9b1d8897b 100644 --- a/crates/e2e_test/src/distributed/heal_test.rs +++ b/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> { 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() { diff --git a/crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs b/crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs index a4df5f4b2..5d663be9f 100644 --- a/crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs +++ b/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>(()); + } + 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(); diff --git a/crates/ecstore/src/disk/error.rs b/crates/ecstore/src/disk/error.rs index 8e551b4af..c6094d8e5 100644 --- a/crates/ecstore/src/disk/error.rs +++ b/crates/ecstore/src/disk/error.rs @@ -627,13 +627,18 @@ impl From 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::()) { + 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, diff --git a/crates/ecstore/src/set_disk/ops/heal.rs b/crates/ecstore/src/set_disk/ops/heal.rs index a983ad8b6..486a6c5cb 100644 --- a/crates/ecstore/src/set_disk/ops/heal.rs +++ b/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::())) +} + 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"); 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] { diff --git a/rustfs/src/app/object/get.rs b/rustfs/src/app/object/get.rs index dd58487e5..7ca48bcda 100644 --- a/rustfs/src/app/object/get.rs +++ b/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::() { - 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] diff --git a/rustfs/src/config/cli.rs b/rustfs/src/config/cli.rs index 9333137a4..2a6f4d3d3 100644 --- a/rustfs/src/config/cli.rs +++ b/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::>() - ); - - 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()]); diff --git a/rustfs/src/config/mod.rs b/rustfs/src/config/mod.rs index 5822a74de..05d66ce52 100644 --- a/rustfs/src/config/mod.rs +++ b/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::{ diff --git a/rustfs/src/connect/diagnostics/job.rs b/rustfs/src/connect/diagnostics/job.rs index 171b052e9..20d0fcb56 100644 --- a/rustfs/src/connect/diagnostics/job.rs +++ b/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::>() + ); + + 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(); diff --git a/rustfs/src/connect/diagnostics/mod.rs b/rustfs/src/connect/diagnostics/mod.rs index b7172ae72..e4ec5241f 100644 --- a/rustfs/src/connect/diagnostics/mod.rs +++ b/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, diff --git a/rustfs/tests/connect_heartbeat.rs b/rustfs/tests/connect_heartbeat.rs index dc2652797..7b5b1034e 100644 --- a/rustfs/tests/connect_heartbeat.rs +++ b/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",