From 61821a6f3e6dcd025f601247eac274b61516dc0b Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Mon, 31 Aug 2026 19:39:47 +0800 Subject: [PATCH] fix(heal): resume remote rebuilds after target restart (#6941) * fix(heal): retry unavailable recreate targets * fix(heal): refresh put-file epochs after target restart * test(e2e): harden heal restart evidence Co-Authored-By: heihutu * test(e2e): cancel competing heal before restart Co-Authored-By: heihutu --------- Co-authored-by: houseme Co-authored-by: heihutu --- .config/e2e-nightly-selection.txt | 2 +- .../src/heal_erasure_disk_rebuild_test.rs | 485 +++++++++++++++++- .../cluster/rpc/internode_data_transport.rs | 302 ++++++++++- crates/ecstore/src/disk/error.rs | 81 ++- crates/ecstore/src/erasure/coding/encode.rs | 7 + crates/ecstore/src/erasure/coding/heal.rs | 132 ++++- crates/ecstore/src/set_disk/ops/heal.rs | 36 +- crates/heal/src/heal/task.rs | 2 +- crates/heal/src/heal/task/heal_bucket.rs | 33 +- crates/heal/src/heal/task/tests.rs | 143 ++++++ docs/testing/e2e-suite-inventory.md | 4 +- 11 files changed, 1169 insertions(+), 58 deletions(-) diff --git a/.config/e2e-nightly-selection.txt b/.config/e2e-nightly-selection.txt index 2ffb159d3..a3a2c14ad 100644 --- a/.config/e2e-nightly-selection.txt +++ b/.config/e2e-nightly-selection.txt @@ -1 +1 @@ -sha256=26003ce03eca11391d1c080491e4f408526717e4b47967b62db08abe1edd189a +sha256=9c2b958035a038ffd5ab98cac5f59a1b8e6a16e141f109ec7fb956afc0f11105 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 dd82512a5..46ea2039f 100644 --- a/crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs +++ b/crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs @@ -16,13 +16,14 @@ #[cfg(test)] mod tests { - use crate::chaos::signed_admin_post; - use crate::common::{RustFSTestClusterEnvironment, RustFSTestEnvironment, init_logging}; + use crate::chaos::{VersionShardCensus, census_object_version_on_disk, signed_admin_post}; + use crate::common::{RustFSTestClusterEnvironment, RustFSTestEnvironment, admin_request, init_logging}; use aws_sdk_s3::primitives::ByteStream; + use http::Method; use std::collections::HashSet; use std::error::Error; use std::path::{Path, PathBuf}; - use tokio::time::{Duration, sleep, timeout}; + use tokio::time::{Duration, Instant, sleep, timeout}; use tracing::info; fn has_file_under(path: &Path) -> bool { @@ -48,6 +49,110 @@ mod tests { disk.join(bucket).join(key).join("xl.meta").is_file() } + // Healing may rewrite non-identity bookkeeping in xl.meta. The census + // therefore compares the canonical selected metadata fields plus every + // physical shard, while the payload seed makes object mix-ups observable. + #[derive(Debug)] + struct PhysicalObjectManifest { + key: String, + payload_seed: u8, + shard_census: VersionShardCensus, + } + + fn deterministic_object_body(len: usize, seed: u8) -> Vec { + let mut value = seed; + std::iter::repeat_with(|| { + value = value.wrapping_mul(31).wrapping_add(17); + value + }) + .take(len) + .collect() + } + + fn matching_manifest_count( + disk: &Path, + bucket: &str, + expected_manifests: &[PhysicalObjectManifest], + ) -> Result> { + let mut matching = 0; + for expected in expected_manifests { + let actual = census_object_version_on_disk(disk, bucket, &expected.key, None)?; + if actual.matches_manifest(&expected.shard_census) { + matching += 1; + } + } + Ok(matching) + } + + fn metadata_count(disk: &Path, bucket: &str, expected_manifests: &[PhysicalObjectManifest]) -> usize { + expected_manifests + .iter() + .filter(|expected| object_metadata_exists_on_disk(disk, bucket, &expected.key)) + .count() + } + + fn heal_task_status_diagnostic(body: &str) -> String { + let Ok(status) = serde_json::from_str::(body) else { + return body.to_string(); + }; + let items = status["items"].as_array(); + let mut unresolved_states = HashSet::new(); + for item in items.into_iter().flatten() { + for drive in item["after"]["drives"].as_array().into_iter().flatten() { + if let Some(state) = drive["state"].as_str() + && state != "ok" + { + unresolved_states.insert(state.to_string()); + } + } + } + let mut unresolved_states = unresolved_states.into_iter().collect::>(); + unresolved_states.sort(); + format!( + "summary={:?}, detail={:?}, item_count={}, unresolved_drive_states={unresolved_states:?}", + status["summary"].as_str(), + status["detail"].as_str(), + items.map_or(0, Vec::len) + ) + } + + fn cluster_heal_is_idle(status: &serde_json::Value) -> bool { + let operations = &status["healOperations"]; + status["clusterStatusComplete"] == serde_json::Value::Bool(true) + && status["state"].as_str() == Some("idle") + && operations["queueLength"].as_u64() == Some(0) + && operations["activeTasks"].as_u64() == Some(0) + && operations["retryingTasks"].as_u64() == Some(0) + } + + fn only_admin_heal_is_active(status: &serde_json::Value) -> bool { + let operations = &status["healOperations"]; + status["clusterStatusComplete"] == serde_json::Value::Bool(true) + && status["state"].as_str() == Some("active") + && operations["queueLength"].as_u64() == Some(0) + && operations["activeTasks"].as_u64() == Some(1) + && operations["retryingTasks"].as_u64() == Some(0) + && operations["activeBySource"]["admin"].as_u64() == Some(1) + } + + async fn replacement_recovery_status( + cluster: &RustFSTestClusterEnvironment, + ) -> Result> { + let (status, body) = admin_request( + &cluster.nodes[0].url, + Method::GET, + "/rustfs/admin/v4/heal/replacement-recovery", + None, + &cluster.access_key, + &cluster.secret_key, + ) + .await?; + if !status.is_success() { + return Err(format!("replacement recovery status failed: {status} {body}").into()); + } + serde_json::from_str(&body).map_err(|err| format!("replacement recovery status is not JSON ({err}): {body}").into()) + } + async fn assert_object_body(env: &RustFSTestEnvironment, bucket: &str, key: &str, expected: &[u8]) { let client = env.create_s3_client(); let response = client @@ -442,6 +547,380 @@ mod tests { .into()) } + // Keep the original unformatted-disk scenario above. This case retains the + // format identity so only the explicit admin task can rebuild missing data. + #[tokio::test(flavor = "multi_thread")] + async fn test_cluster_root_heal_resumes_missing_remote_shards_after_node_restart() -> Result<(), Box> + { + init_logging(); + info!( + event = "heal_restart_started", + component = "e2e_test", + subsystem = "heal", + "Starting root-heal restart test" + ); + + let mut cluster = RustFSTestClusterEnvironment::new(4).await?; + cluster.set_env("RUSTFS_UNSAFE_BYPASS_DISK_CHECK", "true"); + cluster.set_env("RUSTFS_HEAL_ENABLED", "true"); + cluster.set_env("RUSTFS_HEAL_AUTO_HEAL_ENABLE", "false"); + cluster.set_env("RUSTFS_HEAL_MRF_ENABLE", "false"); + cluster.set_env("RUSTFS_SCANNER_ENABLED", "false"); + cluster.set_env("RUSTFS_HEAL_MAX_CONCURRENT_HEALS", "1"); + cluster.set_env("RUSTFS_HEAL_MAX_CONCURRENT_PER_SET", "1"); + cluster.set_env("RUSTFS_HEAL_PAGE_OBJECT_CONCURRENCY", "1"); + cluster.set_env("RUSTFS_HEAL_PAGE_PARALLEL_ENABLE", "false"); + // Keep all storage nodes' Heal runtimes enabled so their disk services + // complete normal registration after restart. Scanner, auto-heal and + // MRF are disabled; the pre-root idle barrier below drains the direct + // outage-object repair before the explicit admin task starts. + let server_rust_log = std::env::var("RUSTFS_HEAL_CHAOS_SERVER_RUST_LOG") + .unwrap_or_else(|_| "rustfs::heal::task=info,rustfs=error".to_string()); + cluster.set_env("RUST_LOG", server_rust_log); + if let Ok(log_dir) = std::env::var("RUSTFS_HEAL_CHAOS_LOG_DIR") { + std::fs::create_dir_all(&log_dir)?; + for node_index in 0..cluster.nodes.len() { + cluster.set_node_capture_log_path(node_index, format!("{log_dir}/node{node_index}.log"))?; + } + } + cluster.start().await?; + let clients = cluster.create_all_clients()?; + + let bucket = "heal-restart-during-rebuild"; + clients[0].create_bucket().bucket(bucket).send().await?; + + let replaced_disk = PathBuf::from(&cluster.nodes[1].data_dir); + let replacement_format_path = replaced_disk.join(".rustfs.sys").join("format.json"); + let replacement_format = std::fs::read(&replacement_format_path).map_err(|err| { + format!("failed to capture target format before replacement wipe at {replacement_format_path:?}: {err}") + })?; + let online_object_count = std::env::var("RUSTFS_HEAL_CHAOS_OBJECT_COUNT") + .ok() + .and_then(|value| value.parse::().ok()) + .unwrap_or(24) + .clamp(8, 64); + let object_size_bytes = std::env::var("RUSTFS_HEAL_CHAOS_OBJECT_SIZE_BYTES") + .ok() + .and_then(|value| value.parse::().ok()) + .unwrap_or(4 * 1024 * 1024) + .clamp(1024 * 1024, 16 * 1024 * 1024); + let mut expected_manifests = Vec::with_capacity(online_object_count); + for index in 0..online_object_count { + let key = format!("cluster/online/object-{index:04}.bin"); + let payload_seed = u8::try_from(index + 1).expect("clamped object count must fit in u8"); + timeout( + Duration::from_secs(30), + clients[0] + .put_object() + .bucket(bucket) + .key(&key) + .body(ByteStream::from(deterministic_object_body(object_size_bytes, payload_seed))) + .send(), + ) + .await??; + let shard_census = census_object_version_on_disk(&replaced_disk, bucket, &key, None)?; + assert!( + shard_census.is_complete(), + "node 1 should hold a complete baseline shard for {key}: {shard_census:?}" + ); + assert!( + !shard_census.expected_part_numbers.is_empty(), + "chaos objects must use physical part shards rather than inline data: {shard_census:?}" + ); + expected_manifests.push(PhysicalObjectManifest { + key, + payload_seed, + shard_census, + }); + } + + cluster.stop_node(1)?; + std::fs::remove_dir_all(&replaced_disk)?; + std::fs::create_dir_all( + replacement_format_path + .parent() + .ok_or("replacement format path has no parent")?, + )?; + std::fs::write(&replacement_format_path, replacement_format)?; + assert!( + replacement_format_path.is_file(), + "replacement target must retain only its preformatted topology identity" + ); + + let outage_key = "cluster/written-while-node-down.bin"; + let outage_payload_seed = 0xf1; + timeout( + Duration::from_secs(30), + clients[2] + .put_object() + .bucket(bucket) + .key(outage_key) + .body(ByteStream::from(deterministic_object_body(object_size_bytes, outage_payload_seed))) + .send(), + ) + .await??; + + let mut outage_peer_erasure_indices = HashSet::new(); + for (node_index, node) in cluster.nodes.iter().enumerate() { + if node_index == 1 { + continue; + } + let census = census_object_version_on_disk(Path::new(&node.data_dir), bucket, outage_key, None)?; + assert!( + census.is_complete(), + "online node {node_index} must hold a complete outage-object shard: {census:?}" + ); + let erasure_index = census + .erasure_index + .ok_or_else(|| format!("online node {node_index} outage-object shard has no erasure index: {census:?}"))?; + assert!( + (1..=cluster.nodes.len()).contains(&erasure_index), + "online node {node_index} outage-object erasure index is out of range: {census:?}" + ); + assert!( + outage_peer_erasure_indices.insert(erasure_index), + "outage-object erasure index {erasure_index} is duplicated across online nodes" + ); + } + assert_eq!( + outage_peer_erasure_indices.len(), + cluster.nodes.len().saturating_sub(1), + "every online node must contribute one unique outage-object erasure index" + ); + let expected_outage_target_erasure_index = (1..=cluster.nodes.len()) + .find(|index| !outage_peer_erasure_indices.contains(index)) + .ok_or("online outage-object shards leave no erasure index for the replacement target")?; + + // The PUT path may have admitted a direct Internal object repair while + // node 1 was offline. Cancel the isolated bucket path before the target + // returns; otherwise it could rebuild the outage object and invalidate + // the explicit-root ownership assertion below. + let cancel_outage_heal_path = format!("/rustfs/admin/v3/heal/{bucket}?forceStop=true"); + let (cancel_status, cancel_body) = admin_request( + &cluster.nodes[0].url, + Method::POST, + &cancel_outage_heal_path, + Some( + r#"{"recursive":true,"dryRun":false,"remove":false,"recreate":true,"scanMode":2,"updateParity":false,"nolock":false}"# + .to_string(), + ), + &cluster.access_key, + &cluster.secret_key, + ) + .await?; + if !cancel_status.is_success() { + return Err(format!("cancel outage heal failed: {cancel_status} {cancel_body}").into()); + } + + cluster.start_node(1).await?; + + let status_url = format!("{}/rustfs/admin/v3/background-heal/status", cluster.nodes[0].url); + let recovery_deadline = Instant::now() + Duration::from_secs(60); + loop { + let status_body = signed_admin_post(&status_url, None, &cluster.access_key, &cluster.secret_key).await?; + assert!( + !status_body.contains("MissingContentLength"), + "background heal status should not fail without an explicit Content-Length: {status_body}" + ); + let recovered: serde_json::Value = serde_json::from_str(&status_body) + .map_err(|err| format!("background heal status is not JSON ({err}): {status_body}"))?; + if cluster_heal_is_idle(&recovered) { + break; + } + if Instant::now() >= recovery_deadline { + return Err(format!("cluster heal operations did not become idle before root heal: {recovered}").into()); + } + sleep(Duration::from_millis(250)).await; + } + assert_eq!( + matching_manifest_count(&replaced_disk, bucket, &expected_manifests)?, + 0, + "non-admin Heal is disabled, so the replacement target must remain empty before the explicit root heal" + ); + assert!( + !census_object_version_on_disk(&replaced_disk, bucket, outage_key, None)?.has_xl_meta, + "the object written during the outage must be absent before the explicit root heal" + ); + let pre_heal_replacement = replacement_recovery_status(&cluster).await?; + assert_eq!( + pre_heal_replacement["cluster"]["records"].as_array().map(Vec::len), + Some(0), + "isolated target must not retain an automatic replacement generation: {pre_heal_replacement}" + ); + + 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); + let heal_start_body = signed_admin_post(&heal_url, Some(heal_body), &cluster.access_key, &cluster.secret_key).await?; + let heal_start: serde_json::Value = serde_json::from_str(&heal_start_body) + .map_err(|err| format!("heal start response is not JSON ({err}): {heal_start_body}"))?; + let client_token = heal_start["clientToken"] + .as_str() + .filter(|token| !token.is_empty()) + .ok_or_else(|| format!("heal start response has no client token: {heal_start}"))?; + let task_status_url = format!("{}/rustfs/admin/v3/heal/?clientToken={client_token}", cluster.nodes[0].url); + + let partial_timeout_secs = std::env::var("RUSTFS_HEAL_CHAOS_PARTIAL_TIMEOUT_SECS") + .ok() + .and_then(|value| value.parse::().ok()) + .unwrap_or(60); + let partial_deadline = Instant::now() + Duration::from_secs(partial_timeout_secs); + let pre_interrupt_status = loop { + let status_body = signed_admin_post(&status_url, None, &cluster.access_key, &cluster.secret_key).await?; + let active_status: serde_json::Value = serde_json::from_str(&status_body) + .map_err(|err| format!("background heal status is not JSON ({err}): {status_body}"))?; + if only_admin_heal_is_active(&active_status) { + break active_status; + } + if Instant::now() >= partial_deadline { + return Err(format!("root heal never became active within {partial_timeout_secs}s: {active_status}").into()); + } + sleep(Duration::from_millis(50)).await; + }; + let partial_count = loop { + let matching = matching_manifest_count(&replaced_disk, bucket, &expected_manifests)?; + if matching > 0 && matching < expected_manifests.len() { + break matching; + } + if matching == expected_manifests.len() { + return Err(format!( + "root heal rebuilt all {} baseline objects before the target could be interrupted", + expected_manifests.len() + ) + .into()); + } + if Instant::now() >= partial_deadline { + return Err(format!( + "root heal made no observable partial progress on the replacement target within {partial_timeout_secs}s" + ) + .into()); + } + sleep(Duration::from_millis(10)).await; + }; + info!( + event = "heal_restart_checkpoint", + component = "e2e_test", + subsystem = "heal", + partial_count, + "Verified unique admin owner before target interruption" + ); + + cluster.stop_node(1)?; + let stopped_count = matching_manifest_count(&replaced_disk, bucket, &expected_manifests)?; + assert!( + stopped_count > 0 && stopped_count < expected_manifests.len(), + "the target must stop after a partial rebuild, observed before stop={partial_count}, after stop={stopped_count}, total={}", + expected_manifests.len() + ); + let unclean_shutdown_marker = replaced_disk.join(".rustfs.sys").join("unclean-shutdown"); + match std::fs::remove_file(&unclean_shutdown_marker) { + Ok(()) => {} + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => { + return Err(format!("failed to isolate unclean recovery marker {unclean_shutdown_marker:?}: {error}").into()); + } + } + cluster.start_node(1).await?; + + let heal_timeout_secs = std::env::var("RUSTFS_HEAL_REPLACED_DISK_TIMEOUT_SECS") + .ok() + .and_then(|value| value.parse::().ok()) + .unwrap_or(180); + let heal_deadline = Instant::now() + Duration::from_secs(heal_timeout_secs); + loop { + if metadata_count(&replaced_disk, bucket, &expected_manifests) == expected_manifests.len() + && object_metadata_exists_on_disk(&replaced_disk, bucket, outage_key) + { + let matching = matching_manifest_count(&replaced_disk, bucket, &expected_manifests)?; + let outage_census = census_object_version_on_disk(&replaced_disk, bucket, outage_key, None)?; + if matching == expected_manifests.len() && outage_census.is_complete() { + break; + } + } + if Instant::now() >= heal_deadline { + let matching = matching_manifest_count(&replaced_disk, bucket, &expected_manifests)?; + let outage_census = census_object_version_on_disk(&replaced_disk, bucket, outage_key, None)?; + let final_status = signed_admin_post(&status_url, None, &cluster.access_key, &cluster.secret_key) + .await + .unwrap_or_else(|err| format!("status request failed: {err}")); + let task_status = match timeout( + Duration::from_secs(5), + signed_admin_post(&task_status_url, None, &cluster.access_key, &cluster.secret_key), + ) + .await + { + Ok(Ok(body)) => heal_task_status_diagnostic(&body), + Ok(Err(err)) => format!("task status request failed: {err}"), + Err(_) => "task status request exceeded 5s diagnostic budget".to_string(), + }; + let replacement_status = match timeout(Duration::from_secs(5), replacement_recovery_status(&cluster)).await { + Ok(Ok(status)) => status.to_string(), + Ok(Err(err)) => format!("replacement status request failed: {err}"), + Err(_) => "replacement status request exceeded 5s diagnostic budget".to_string(), + }; + return Err(format!( + "root heal did not resume after target restart within {heal_timeout_secs}s: baseline={matching}/{}, outage={outage_census:?}, status={final_status}, task_status={task_status}, pre_interrupt_status={pre_interrupt_status}, pre_heal_replacement={pre_heal_replacement}, replacement_status={replacement_status}", + expected_manifests.len() + ) + .into()); + } + sleep(Duration::from_millis(250)).await; + } + + for expected in &expected_manifests { + let actual = census_object_version_on_disk(&replaced_disk, bucket, &expected.key, None)?; + assert!( + actual.matches_manifest(&expected.shard_census), + "rebuilt target shard differs from its baseline for {}: {actual:?}", + expected.key + ); + } + let outage_census = census_object_version_on_disk(&replaced_disk, bucket, outage_key, None)?; + assert!( + outage_census.is_complete(), + "outage object must have a complete target shard: {outage_census:?}" + ); + assert_eq!( + outage_census.erasure_index, + Some(expected_outage_target_erasure_index), + "the outage object must be rebuilt into its own missing erasure slot" + ); + + let target_client = cluster.create_s3_client(1)?; + for expected in &expected_manifests { + let response = target_client.get_object().bucket(bucket).key(&expected.key).send().await?; + let actual = response.body.collect().await?.into_bytes(); + let expected_body = deterministic_object_body(object_size_bytes, expected.payload_seed); + assert_eq!(actual.as_ref(), expected_body.as_slice(), "object body changed for {}", expected.key); + } + let response = target_client.get_object().bucket(bucket).key(outage_key).send().await?; + let actual = response.body.collect().await?.into_bytes(); + let expected_outage_body = deterministic_object_body(object_size_bytes, outage_payload_seed); + assert_eq!(actual.as_ref(), expected_outage_body.as_slice(), "object body changed for {outage_key}"); + + let terminal_deadline = Instant::now() + Duration::from_secs(30); + loop { + let status_body = signed_admin_post(&status_url, None, &cluster.access_key, &cluster.secret_key).await?; + let status: serde_json::Value = serde_json::from_str(&status_body) + .map_err(|err| format!("background heal status is not JSON ({err}): {status_body}"))?; + if cluster_heal_is_idle(&status) { + break; + } + if Instant::now() >= terminal_deadline { + return Err(format!("heal data rebuilt but operations did not converge to terminal idle: {status}").into()); + } + sleep(Duration::from_millis(250)).await; + } + + let task_status_body = signed_admin_post(&task_status_url, None, &cluster.access_key, &cluster.secret_key).await?; + let task_status: serde_json::Value = serde_json::from_str(&task_status_body) + .map_err(|err| format!("heal task status is not JSON ({err}): {task_status_body}"))?; + if task_status["summary"].as_str() != Some("finished") { + return Err(format!("heal data rebuilt but task did not finish successfully: {task_status}").into()); + } + + Ok(()) + } + /// Issue #5850: `background-heal/status` must answer while a peer is down. /// /// Exercises the production path in `read_cluster_heal_status` end to end, diff --git a/crates/ecstore/src/cluster/rpc/internode_data_transport.rs b/crates/ecstore/src/cluster/rpc/internode_data_transport.rs index 6b695954f..e1081ffd3 100644 --- a/crates/ecstore/src/cluster/rpc/internode_data_transport.rs +++ b/crates/ecstore/src/cluster/rpc/internode_data_transport.rs @@ -36,6 +36,7 @@ use rustfs_rio::{ChunkReaderBox, HttpChunkReader, HttpReader, HttpWriter}; use sha2::{Digest, Sha256}; use std::collections::HashMap; use std::future::Future; +use std::io; use std::pin::Pin; use std::sync::{Arc, LazyLock, OnceLock}; use std::task::{Context, Poll}; @@ -105,9 +106,13 @@ struct PutFileCapabilityCacheState { cached: Option, generation: u64, in_flight: Option, + rejected_server_epoch: Option, } -type PutFileCapabilityCacheEntry = Arc>; +// The registry lock is released before taking an entry lock. Entry guards cover +// only cache transitions, never a probe or await; poll-based writers must be +// able to reject an epoch atomically with those transitions. +type PutFileCapabilityCacheEntry = Arc>; static PUT_FILE_CAPABILITY_CACHE: LazyLock>> = LazyLock::new(|| parking_lot::RwLock::new(HashMap::new())); @@ -119,7 +124,7 @@ fn put_file_capability_cache_entry(endpoint: &str) -> PutFileCapabilityCacheEntr PUT_FILE_CAPABILITY_CACHE .write() .entry(endpoint.to_owned()) - .or_insert_with(|| Arc::new(tokio::sync::RwLock::new(PutFileCapabilityCacheState::default()))) + .or_insert_with(|| Arc::new(parking_lot::RwLock::new(PutFileCapabilityCacheState::default()))) .clone() } @@ -134,6 +139,23 @@ fn fresh_put_file_capability(state: Option, now: Instant } } +fn reject_put_file_server_epoch(endpoint: &str, server_epoch: Uuid) { + let entry = PUT_FILE_CAPABILITY_CACHE.read().get(endpoint).cloned(); + if let Some(entry) = entry { + let mut state = entry.write(); + if matches!(state.cached, Some(PutFileCapabilityState::V1 { server_epoch: cached, .. }) if cached == server_epoch) { + state.rejected_server_epoch = Some(server_epoch); + } + } +} + +fn usable_put_file_capability(state: &PutFileCapabilityCacheState, now: Instant) -> Option> { + match fresh_put_file_capability(state.cached, now)? { + Some(server_epoch) if state.rejected_server_epoch == Some(server_epoch) => None, + capability => Some(capability), + } +} + fn put_file_capability_status_is_legacy(status: u16) -> bool { status == 404 } @@ -322,13 +344,14 @@ impl InternodeDataTransport for TcpHttpInternodeDataTransport { async fn open_write(&self, request: WriteStreamRequest) -> Result { let server_epoch = self.put_file_auth_capability(&request.endpoint).await?; - let nonce = server_epoch.map(|_| Uuid::new_v4()); - let url = build_put_file_stream_url(&request, nonce.zip(server_epoch)); + let auth_scope = server_epoch.map(|server_epoch| (Uuid::new_v4(), server_epoch)); + let url = build_put_file_stream_url(&request, auth_scope); + let endpoint = request.endpoint; let mut headers = json_headers(); build_auth_headers(&url, &Method::PUT, &mut headers)?; let writer = HttpWriter::new(url.clone(), Method::PUT, headers).await?; - match nonce { - Some(nonce) => Ok(Box::new(PutFileAuthWriter::new(writer, url, nonce))), + match auth_scope { + Some((nonce, server_epoch)) => Ok(Box::new(PutFileAuthWriter::new(writer, url, nonce, endpoint, server_epoch))), None => Ok(Box::new(writer)), } } @@ -498,15 +521,15 @@ where { let entry = put_file_capability_cache_entry(endpoint); { - let state = entry.read().await; - if let Some(cached) = fresh_put_file_capability(state.cached, Instant::now()) { + let state = entry.read(); + if let Some(cached) = usable_put_file_capability(&state, Instant::now()) { return Ok(cached); } } let flight = { - let mut state = entry.write().await; - if let Some(cached) = fresh_put_file_capability(state.cached, Instant::now()) { + let mut state = entry.write(); + if let Some(cached) = usable_put_file_capability(&state, Instant::now()) { return Ok(cached); } if let Some(flight) = state.in_flight.clone() { @@ -532,7 +555,7 @@ where .await; { - let mut state = entry.write().await; + let mut state = entry.write(); let is_current_flight = state .in_flight .as_ref() @@ -540,6 +563,9 @@ where if is_current_flight { match outcome { Ok(Some(server_epoch)) => { + if state.rejected_server_epoch != Some(*server_epoch) { + state.rejected_server_epoch = None; + } state.cached = Some(PutFileCapabilityState::V1 { server_epoch: *server_epoch, revalidate_after: Instant::now() + PUT_FILE_V1_CAPABILITY_TTL, @@ -630,17 +656,23 @@ struct PutFileAuthWriter { inner: W, url: String, nonce: Uuid, + endpoint: String, + server_epoch: Uuid, + server_epoch_rejected: bool, hasher: Sha256, trailer: Option>, trailer_offset: usize, } impl PutFileAuthWriter { - fn new(inner: W, url: String, nonce: Uuid) -> Self { + fn new(inner: W, url: String, nonce: Uuid, endpoint: String, server_epoch: Uuid) -> Self { Self { inner, url, nonce, + endpoint, + server_epoch, + server_epoch_rejected: false, hasher: Sha256::new(), trailer: None, trailer_offset: 0, @@ -656,6 +688,14 @@ impl PutFileAuthWriter { Ok(()) } + fn reject_server_epoch_on_conflict(&mut self, error: &io::Error) { + if self.server_epoch_rejected || !io_error_has_put_file_epoch_conflict(error) { + return; + } + reject_put_file_server_epoch(&self.endpoint, self.server_epoch); + self.server_epoch_rejected = true; + } + fn poll_write_trailer(&mut self, cx: &mut Context<'_>) -> Poll> where W: AsyncWrite + Unpin, @@ -673,7 +713,10 @@ impl PutFileAuthWriter { ))); } Poll::Ready(Ok(written)) => written, - Poll::Ready(Err(err)) => return Poll::Ready(Err(err)), + Poll::Ready(Err(err)) => { + self.reject_server_epoch_on_conflict(&err); + return Poll::Ready(Err(err)); + } Poll::Pending => return Poll::Pending, }; self.trailer_offset += written; @@ -682,6 +725,15 @@ impl PutFileAuthWriter { } } +fn io_error_has_put_file_epoch_conflict(error: &io::Error) -> bool { + error + .get_ref() + .and_then(|source| source.downcast_ref::()) + .is_some_and( + |error| matches!(error.kind(), rustfs_rio::InternodeHttpErrorKind::HttpStatus(status) if status.as_u16() == 409), + ) +} + impl AsyncWrite for PutFileAuthWriter where W: AsyncWrite + Unpin, @@ -698,12 +750,22 @@ where self.hasher.update(&buf[..written]); Poll::Ready(Ok(written)) } - other => other, + Poll::Ready(Err(err)) => { + self.reject_server_epoch_on_conflict(&err); + Poll::Ready(Err(err)) + } + Poll::Pending => Poll::Pending, } } fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - Pin::new(&mut self.inner).poll_flush(cx) + match Pin::new(&mut self.inner).poll_flush(cx) { + Poll::Ready(Err(err)) => { + self.reject_server_epoch_on_conflict(&err); + Poll::Ready(Err(err)) + } + other => other, + } } fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { @@ -712,7 +774,13 @@ where Poll::Ready(Err(err)) => return Poll::Ready(Err(err)), Poll::Pending => return Poll::Pending, } - Pin::new(&mut self.inner).poll_shutdown(cx) + match Pin::new(&mut self.inner).poll_shutdown(cx) { + Poll::Ready(Err(err)) => { + self.reject_server_epoch_on_conflict(&err); + Poll::Ready(Err(err)) + } + other => other, + } } } @@ -840,7 +908,6 @@ mod tests { loop { let strong_count = entry .read() - .await .in_flight .as_ref() .map(|flight| Arc::strong_count(&flight.outcome)) @@ -858,6 +925,50 @@ mod tests { #[derive(Debug)] struct LegacyTestTransport; + #[derive(Clone, Copy, Debug)] + enum PutFileFailurePhase { + Write, + Flush, + Shutdown, + } + + struct PutFileFailureWriter { + phase: PutFileFailurePhase, + status: reqwest::StatusCode, + } + + impl PutFileFailureWriter { + fn error(&self) -> io::Error { + rustfs_rio::new_test_internode_http_io_error(rustfs_rio::InternodeHttpErrorKind::HttpStatus(self.status)) + } + } + + impl tokio::io::AsyncWrite for PutFileFailureWriter { + fn poll_write(self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &[u8]) -> Poll> { + Poll::Ready(if matches!(self.phase, PutFileFailurePhase::Write) { + Err(self.error()) + } else { + Ok(buf.len()) + }) + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(if matches!(self.phase, PutFileFailurePhase::Flush) { + Err(self.error()) + } else { + Ok(()) + }) + } + + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(if matches!(self.phase, PutFileFailurePhase::Shutdown) { + Err(self.error()) + } else { + Ok(()) + }) + } + } + #[async_trait::async_trait] impl InternodeDataTransport for LegacyTestTransport { async fn open_read(&self, _request: ReadStreamRequest) -> Result { @@ -1048,7 +1159,7 @@ mod tests { let v1_endpoint = format!("http://v1-{}.invalid", Uuid::new_v4()); let v1_entry = put_file_capability_cache_entry(&v1_endpoint); let server_epoch = Uuid::new_v4(); - v1_entry.write().await.cached = Some(PutFileCapabilityState::V1 { + v1_entry.write().cached = Some(PutFileCapabilityState::V1 { server_epoch, revalidate_after: Instant::now() + PUT_FILE_V1_CAPABILITY_TTL, }); @@ -1067,7 +1178,7 @@ mod tests { Some(server_epoch) ); assert!(!cache_probe_called.load(Ordering::SeqCst)); - v1_entry.write().await.cached = Some(PutFileCapabilityState::V1 { + v1_entry.write().cached = Some(PutFileCapabilityState::V1 { server_epoch, revalidate_after: Instant::now(), }); @@ -1086,8 +1197,7 @@ mod tests { let legacy_endpoint = format!("http://legacy-{}.invalid", Uuid::new_v4()); let legacy_entry = put_file_capability_cache_entry(&legacy_endpoint); - legacy_entry.write().await.cached = - Some(PutFileCapabilityState::LegacyUntil(Instant::now() + PUT_FILE_LEGACY_CAPABILITY_TTL)); + legacy_entry.write().cached = Some(PutFileCapabilityState::LegacyUntil(Instant::now() + PUT_FILE_LEGACY_CAPABILITY_TTL)); assert!( transport .put_file_auth_capability(&legacy_endpoint) @@ -1098,7 +1208,7 @@ mod tests { let expired_endpoint = format!("http://expired-legacy-{}.invalid", Uuid::new_v4()); let expired_entry = put_file_capability_cache_entry(&expired_endpoint); - expired_entry.write().await.cached = Some(PutFileCapabilityState::LegacyUntil(Instant::now())); + expired_entry.write().cached = Some(PutFileCapabilityState::LegacyUntil(Instant::now())); let reprobed = std::sync::atomic::AtomicBool::new(false); assert_eq!( resolve_put_file_auth_capability(&expired_endpoint, || async { @@ -1349,7 +1459,7 @@ mod tests { }; probe_started.notified().await; { - let mut state = entry.write().await; + let mut state = entry.write(); state.generation = state.generation.checked_add(1).expect("test generation should advance"); state.cached = Some(PutFileCapabilityState::V1 { server_epoch: newer_epoch, @@ -1362,10 +1472,7 @@ mod tests { task.await.expect("stale task should finish").expect("stale probe result"), Some(stale_epoch) ); - assert_eq!( - fresh_put_file_capability(entry.read().await.cached, Instant::now()), - Some(Some(newer_epoch)) - ); + assert_eq!(fresh_put_file_capability(entry.read().cached, Instant::now()), Some(Some(newer_epoch))); } #[test] @@ -1398,6 +1505,8 @@ mod tests { let _ = rustfs_credentials::set_global_rpc_secret("put-file-auth-writer-test-secret".to_string()); let nonce = Uuid::parse_str("11111111-2222-4333-8444-555555555555").expect("nonce"); + let server_epoch = Uuid::parse_str("aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee").expect("server epoch"); + let endpoint = "http://node1:9000".to_string(); let url = concat!( "http://node1:9000/rustfs/rpc/put_file_stream?disk=disk-a&volume=bucket&path=object%2Fpart.1", "&append=false&size=11&put_file_auth=digest-trailer-v1&put_file_nonce=11111111-2222-4333-8444-555555555555" @@ -1406,7 +1515,7 @@ mod tests { let mut sink = Vec::new(); { - let mut writer = PutFileAuthWriter::new(&mut sink, url.clone(), nonce); + let mut writer = PutFileAuthWriter::new(&mut sink, url.clone(), nonce, endpoint, server_epoch); writer.write_all(b"hello world").await.expect("body write should succeed"); writer.shutdown().await.expect("shutdown should append auth trailer"); let err = writer @@ -1424,6 +1533,143 @@ mod tests { assert_eq!(verified, expected_digest); } + #[tokio::test] + async fn put_file_auth_writer_reprobes_after_server_epoch_conflict() { + use tokio::io::AsyncWriteExt; + + let _ = rustfs_credentials::set_global_rpc_secret("put-file-epoch-conflict-test-secret".to_string()); + for status in [reqwest::StatusCode::CONFLICT, reqwest::StatusCode::BAD_REQUEST] { + for (phase, trailer_write) in [ + (PutFileFailurePhase::Write, false), + (PutFileFailurePhase::Write, true), + (PutFileFailurePhase::Flush, false), + (PutFileFailurePhase::Shutdown, false), + ] { + let endpoint = format!("http://epoch-conflict-{}.invalid", Uuid::new_v4()); + let stale_epoch = Uuid::new_v4(); + let replacement_epoch = Uuid::new_v4(); + resolve_put_file_auth_capability(&endpoint, || async { Ok(Some(stale_epoch)) }) + .await + .expect("initial capability should resolve"); + let mut writer = PutFileAuthWriter::new( + PutFileFailureWriter { phase, status }, + format!("{endpoint}{PUT_FILE_AUTH_STREAM_PATH}"), + Uuid::new_v4(), + endpoint.clone(), + stale_epoch, + ); + let error = match (phase, trailer_write) { + (PutFileFailurePhase::Write, false) => writer.write_all(b"body").await, + (PutFileFailurePhase::Flush, _) => writer.flush().await, + _ => writer.shutdown().await, + } + .expect_err("injected writer error must reach the caller"); + let conflict = status == reqwest::StatusCode::CONFLICT; + assert_eq!(io_error_has_put_file_epoch_conflict(&error), conflict); + + let probe_called = AtomicBool::new(false); + let resolved = resolve_put_file_auth_capability(&endpoint, || async { + probe_called.store(true, Ordering::SeqCst); + Ok(Some(replacement_epoch)) + }) + .await + .expect("capability should remain usable or be reprobed"); + assert_eq!(probe_called.load(Ordering::SeqCst), conflict, "phase={phase:?}, trailer={trailer_write}"); + assert_eq!(resolved, Some(if conflict { replacement_epoch } else { stale_epoch })); + } + } + } + + #[tokio::test] + async fn late_put_file_epoch_rejection_preserves_current_rejection() { + let endpoint = format!("http://late-epoch-conflict-{}.invalid", Uuid::new_v4()); + let old_epoch = Uuid::new_v4(); + let current_epoch = Uuid::new_v4(); + let replacement_epoch = Uuid::new_v4(); + assert_eq!( + resolve_put_file_auth_capability(&endpoint, || async { Ok(Some(old_epoch)) }) + .await + .expect("initial epoch should be cached"), + Some(old_epoch) + ); + reject_put_file_server_epoch(&endpoint, old_epoch); + assert_eq!( + resolve_put_file_auth_capability(&endpoint, || async { Ok(Some(current_epoch)) }) + .await + .expect("first restart should install a new epoch"), + Some(current_epoch) + ); + + reject_put_file_server_epoch(&endpoint, current_epoch); + // A writer opened before the first restart can report its 409 after + // a newer writer has already rejected the second server incarnation. + reject_put_file_server_epoch(&endpoint, old_epoch); + let probe_called = AtomicBool::new(false); + let resolved = resolve_put_file_auth_capability(&endpoint, || async { + probe_called.store(true, Ordering::SeqCst); + Ok(Some(replacement_epoch)) + }) + .await + .expect("late old-epoch rejection must preserve the current rejection"); + + assert!(probe_called.load(Ordering::SeqCst), "known-rejected current epoch must be reprobed"); + assert_eq!(resolved, Some(replacement_epoch)); + } + + #[tokio::test] + async fn put_file_epoch_rejection_is_endpoint_and_epoch_scoped() { + let endpoint = format!("http://scoped-epoch-{}.invalid", Uuid::new_v4()); + let other_endpoint = format!("http://other-epoch-{}.invalid", Uuid::new_v4()); + let current_epoch = Uuid::new_v4(); + for endpoint in [&endpoint, &other_endpoint] { + resolve_put_file_auth_capability(endpoint, || async { Ok(Some(current_epoch)) }) + .await + .expect("initial epoch should resolve"); + } + reject_put_file_server_epoch(&endpoint, Uuid::new_v4()); + assert_eq!( + resolve_put_file_auth_capability(&endpoint, || async { panic!("old writer must not invalidate a new epoch") }) + .await + .expect("new epoch must remain cached"), + Some(current_epoch) + ); + reject_put_file_server_epoch(&endpoint, current_epoch); + assert_eq!( + resolve_put_file_auth_capability(&other_endpoint, || async { panic!("another endpoint must stay cached") }) + .await + .expect("other endpoint must remain cached"), + Some(current_epoch) + ); + } + + #[tokio::test] + async fn put_file_rejected_epoch_survives_failed_stale_and_downgrade_probes() { + let endpoint = format!("http://rejected-probe-{}.invalid", Uuid::new_v4()); + let rejected_epoch = Uuid::new_v4(); + let replacement_epoch = Uuid::new_v4(); + resolve_put_file_auth_capability(&endpoint, || async { Ok(Some(rejected_epoch)) }) + .await + .expect("initial epoch should resolve"); + reject_put_file_server_epoch(&endpoint, rejected_epoch); + let failure = resolve_put_file_auth_capability(&endpoint, || async { Err(Error::other("injected probe failure")) }) + .await + .expect_err("probe failure must be returned"); + assert!(failure.to_string().contains("injected probe failure")); + let downgrade = resolve_put_file_auth_capability(&endpoint, || async { Ok(None) }) + .await + .expect_err("rejection must not unpin authenticated v1"); + assert!(downgrade.to_string().contains("downgrade rejected")); + resolve_put_file_auth_capability(&endpoint, || async { Ok(Some(rejected_epoch)) }) + .await + .expect("a probe racing a restart can still return the old epoch"); + assert_eq!( + resolve_put_file_auth_capability(&endpoint, || async { Ok(Some(replacement_epoch)) }) + .await + .expect("same-epoch probe must not clear known rejection"), + Some(replacement_epoch) + ); + } + #[test] fn walk_dir_url_encodes_disk_ref() { let url = build_walk_dir_url(&WalkDirStreamRequest { diff --git a/crates/ecstore/src/disk/error.rs b/crates/ecstore/src/disk/error.rs index 5f1d1175b..c4c068800 100644 --- a/crates/ecstore/src/disk/error.rs +++ b/crates/ecstore/src/disk/error.rs @@ -12,6 +12,7 @@ // 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 std::error::Error as StdError; use std::hash::{Hash, Hasher}; @@ -229,6 +230,19 @@ fn classify_internode_missing_error(error: &InternodeHttpError) -> Option bool { + error.kind().is_retryable() + || (matches!(error.kind(), InternodeHttpErrorKind::HttpStatus(status) if status.as_u16() == 409) + && error.context().operation() == Some(INTERNODE_OPERATION_PUT_FILE_STREAM)) +} + +fn io_error_contains_retryable_internode_write(error: &io::Error) -> bool { + error + .get_ref() + .and_then(|source| source.downcast_ref::()) + .is_some_and(internode_write_error_is_retryable) +} + /// Wrap a terminal shard-read failure without changing its typed /// classification. Timeout-like disk errors retain `TimedOut`; other errors /// retain their inner I/O kind or use `Other` when no more specific kind exists. @@ -336,10 +350,7 @@ impl DiskError { pub fn is_retryable_internode_write_failure(&self) -> bool { match self { - DiskError::Io(io_error) => io_error - .get_ref() - .and_then(|source| source.downcast_ref::()) - .is_some_and(|err| err.kind().is_retryable()), + DiskError::Io(io_error) => io_error_contains_retryable_internode_write(io_error), _ => false, } } @@ -1240,6 +1251,68 @@ mod tests { assert!(!DiskError::FileNotFound.is_internode_http_status(429)); } + #[test] + fn test_put_file_server_epoch_conflict_is_retryable_write_failure() { + let conflict = DiskError::from(rustfs_rio::new_test_internode_http_io_error( + rustfs_rio::InternodeHttpErrorKind::HttpStatus(http::StatusCode::CONFLICT), + )); + let bad_request = DiskError::from(rustfs_rio::new_test_internode_http_io_error( + rustfs_rio::InternodeHttpErrorKind::HttpStatus(http::StatusCode::BAD_REQUEST), + )); + + assert!(conflict.is_retryable_internode_write_failure()); + assert!(!bad_request.is_retryable_internode_write_failure()); + } + + #[tokio::test] + async fn read_stream_conflict_is_not_a_retryable_put_file_failure() { + 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 isolated HTTP fixture"); + let address = listener.local_addr().expect("fixture address"); + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.expect("accept read request"); + let mut request = [0_u8; 4096]; + let mut read = 0; + loop { + let count = stream.read(&mut request[read..]).await.expect("read HTTP request"); + assert!(count > 0, "request ended before its complete headers"); + read += count; + if request[..read].windows(4).any(|bytes| bytes == b"\r\n\r\n") { + break; + } + assert!(read < request.len(), "fixture request headers exceed their budget"); + } + stream + .write_all(b"HTTP/1.1 409 Conflict\r\nContent-Length: 0\r\nConnection: close\r\n\r\n") + .await + .expect("send typed conflict 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 409 must fail the read"), + Err(error) => DiskError::from(error), + }; + server.await.expect("fixture task should complete"); + assert!(error.is_internode_http_status(409)); + assert!( + !error.is_retryable_internode_write_failure(), + "read-operation 409 must not trigger put-file retry" + ); + }) + .await + .expect("isolated read-conflict test must finish within its budget"); + } + #[test] fn test_internode_missing_errors_preserve_disk_error_types() { let file_missing = DiskError::from(rustfs_rio::new_test_remote_file_not_found_http_io_error()); diff --git a/crates/ecstore/src/erasure/coding/encode.rs b/crates/ecstore/src/erasure/coding/encode.rs index a910ccb3f..7872499e4 100644 --- a/crates/ecstore/src/erasure/coding/encode.rs +++ b/crates/ecstore/src/erasure/coding/encode.rs @@ -321,6 +321,13 @@ impl<'a> MultiWriter<'a> { } } + pub(super) fn take_retryable_internode_write_failure(&mut self) -> Option { + self.errs + .iter_mut() + .find(|error| error.as_ref().is_some_and(Error::is_retryable_internode_write_failure)) + .and_then(Option::take) + } + /// Effective budget for one shard operation: the smaller of the per-shard /// stall timeout and the time remaining until the object's absolute cap. /// Returns `None` when neither deadline is configured (wait indefinitely). diff --git a/crates/ecstore/src/erasure/coding/heal.rs b/crates/ecstore/src/erasure/coding/heal.rs index 8055965e4..8539d9fd8 100644 --- a/crates/ecstore/src/erasure/coding/heal.rs +++ b/crates/ecstore/src/erasure/coding/heal.rs @@ -108,6 +108,13 @@ where (shards, errs) } +fn heal_writer_failure(writers: &mut MultiWriter<'_>, error: io::Error) -> Error { + writers + .take_retryable_internode_write_failure() + .map(|error| Error::RemoteClientUnavailable(error.to_string())) + .unwrap_or_else(|| error.into()) +} + impl super::Erasure { pub async fn heal( &self, @@ -202,10 +209,14 @@ impl super::Erasure { .map(|s| Bytes::from(s.unwrap_or_default())) .collect::>(); - writers.write(shards).await?; + if let Err(error) = writers.write(shards).await { + return Err(heal_writer_failure(&mut writers, error)); + } } - writers.shutdown().await?; + if let Err(error) = writers.shutdown().await { + return Err(heal_writer_failure(&mut writers, error)); + } Ok(()) } } @@ -246,6 +257,35 @@ mod tests { } } + struct InternodeFailureWriter { + fail_on_write: bool, + status: http::StatusCode, + } + + impl InternodeFailureWriter { + fn error(&self) -> io::Error { + rustfs_rio::new_test_internode_http_io_error(rustfs_rio::InternodeHttpErrorKind::HttpStatus(self.status)) + } + } + + impl AsyncWrite for InternodeFailureWriter { + fn poll_write(self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &[u8]) -> Poll> { + Poll::Ready(if self.fail_on_write { + Err(self.error()) + } else { + Ok(buf.len()) + }) + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Err(self.error())) + } + } + struct PendingReader; impl AsyncRead for PendingReader { @@ -331,6 +371,94 @@ mod tests { assert!(writers.iter().all(Option::is_some)); } + #[tokio::test] + async fn heal_maps_put_file_epoch_conflict_to_retryable_remote_unavailable() { + for status in [http::StatusCode::CONFLICT, http::StatusCode::BAD_REQUEST] { + for (fail_on_write, data) in [ + (false, b"".as_slice()), + (false, b"payload".as_slice()), + (true, b"payload".as_slice()), + ] { + let erasure = Erasure::new(2, 1, 64); + let encoded = erasure.encode_data(data).expect("source shards should encode"); + let readers = encoded + .iter() + .enumerate() + .map(|(index, shard)| { + (index < erasure.data_shards).then(|| { + BitrotReader::new(Cursor::new(shard.to_vec()), erasure.shard_size(), HashAlgorithm::None, false) + }) + }) + .collect::>(); + let mut writers = (0..erasure.total_shard_count()) + .map(|index| { + (index == erasure.data_shards).then(|| { + BitrotWriterWrapper::new( + CustomWriter::new_tokio_writer(InternodeFailureWriter { fail_on_write, status }), + erasure.shard_size(), + HashAlgorithm::None, + ) + }) + }) + .collect::>(); + let error = erasure + .heal(&mut writers, readers, data.len(), &[]) + .await + .expect_err("failed sole target must not satisfy heal write quorum"); + assert_eq!( + matches!(error, Error::RemoteClientUnavailable(_)), + status == http::StatusCode::CONFLICT, + "status={status}, fail_on_write={fail_on_write}, len={}, error={error:?}", + data.len() + ); + assert!(writers.iter().all(Option::is_none), "failed target must not be committed"); + } + } + } + + #[tokio::test] + async fn heal_epoch_conflict_does_not_abort_healthy_target() { + for fail_on_write in [false, true] { + let erasure = Erasure::new(2, 2, 64); + let data = b"healthy target must retain exact reconstructed bytes"; + let encoded = erasure.encode_data(data).expect("source shards should encode"); + let readers = encoded + .iter() + .enumerate() + .map(|(index, shard)| { + (index < erasure.data_shards) + .then(|| BitrotReader::new(Cursor::new(shard.to_vec()), erasure.shard_size(), HashAlgorithm::None, false)) + }) + .collect::>(); + let mut writers = vec![ + None, + None, + Some(BitrotWriterWrapper::new( + CustomWriter::new_tokio_writer(InternodeFailureWriter { + fail_on_write, + status: http::StatusCode::CONFLICT, + }), + erasure.shard_size(), + HashAlgorithm::None, + )), + Some(inline_writer(erasure.shard_size())), + ]; + erasure + .heal(&mut writers, readers, data.len(), &[]) + .await + .expect("one healthy target must still satisfy the existing heal quorum"); + assert!(writers[2].is_none(), "conflicting target must be dropped"); + assert_eq!( + writers[3] + .take() + .expect("healthy target remains") + .into_inline_data() + .expect("inline target data"), + encoded[3].to_vec() + ); + } + } + #[tokio::test] async fn heal_reconstructs_missing_parity_shard() { let erasure = Erasure::new(2, 2, 64); diff --git a/crates/ecstore/src/set_disk/ops/heal.rs b/crates/ecstore/src/set_disk/ops/heal.rs index fb6b1dab5..11df66bc2 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 heal_drive_state_for_error(error: &DiskError) -> DriveState { + match error { + DiskError::DiskNotFound | DiskError::RemoteClientUnavailable(_) => DriveState::Offline, + DiskError::FaultyDisk | DiskError::FaultyRemoteDisk => DriveState::Faulty, + DiskError::FileNotFound + | DiskError::FileVersionNotFound + | DiskError::VolumeNotFound + | DiskError::PartMissingOrCorrupt + | DiskError::OutdatedXLMeta => DriveState::Missing, + DiskError::FileCorrupt => DriveState::Corrupt, + _ => DriveState::Unknown(error.to_string()), + } +} + #[cfg(test)] static HEAL_RENAME_FAILURES: std::sync::Mutex> = std::sync::Mutex::new(Vec::new()); @@ -892,16 +906,7 @@ impl SetDisks { } let drive_state = match reason { - Some(err) => match err { - DiskError::DiskNotFound => DriveState::Offline.to_string(), - DiskError::FileNotFound - | DiskError::FileVersionNotFound - | DiskError::VolumeNotFound - | DiskError::PartMissingOrCorrupt - | DiskError::OutdatedXLMeta => DriveState::Missing.to_string(), - DiskError::FileCorrupt => DriveState::Corrupt.to_string(), - _ => DriveState::Unknown(err.to_string()).to_string(), - }, + Some(err) => heal_drive_state_for_error(&err).to_string(), None => DriveState::Ok.to_string(), }; result.before.drives.push(HealDriveInfo { @@ -2673,6 +2678,17 @@ mod heal_result_report_tests { assert!(!super::metadata_less_part_file("xl.meta")); } + #[test] + fn unavailable_heal_errors_use_stable_drive_states() { + for error in [DiskError::FaultyDisk, DiskError::FaultyRemoteDisk] { + assert_eq!(super::heal_drive_state_for_error(&error).to_string(), DriveState::Faulty.to_string()); + } + assert_eq!( + super::heal_drive_state_for_error(&DiskError::RemoteClientUnavailable("peer restarting".to_string())).to_string(), + DriveState::Offline.to_string() + ); + } + #[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/src/heal/task.rs b/crates/heal/src/heal/task.rs index b39d92162..c95770834 100644 --- a/crates/heal/src/heal/task.rs +++ b/crates/heal/src/heal/task.rs @@ -24,7 +24,7 @@ use crate::heal::{ use crate::{Error, Result}; use metrics::{counter, histogram}; use rustfs_common::trace_bus::{TraceEvent, TraceFunc, TraceKind, trace_emit}; -use rustfs_heal_contracts::heal_channel::{HealOpts, HealRequestSource, HealScanMode}; +use rustfs_heal_contracts::heal_channel::{DriveState, HealOpts, HealRequestSource, HealScanMode}; use rustfs_madmin::heal_commands::HealResultItem; use rustfs_utils::path::SLASH_SEPARATOR; use serde::{Deserialize, Serialize}; diff --git a/crates/heal/src/heal/task/heal_bucket.rs b/crates/heal/src/heal/task/heal_bucket.rs index 37459cac2..d44df549e 100644 --- a/crates/heal/src/heal/task/heal_bucket.rs +++ b/crates/heal/src/heal/task/heal_bucket.rs @@ -16,6 +16,22 @@ use super::*; use crate::heal::progress::{add_bytes, increment_counter, stable_generation}; use crate::heal::utils::format_set_disk_id; +fn unavailable_recreate_error(result: &HealResultItem, opts: &HealOpts) -> Option { + if opts.dry_run || !opts.recreate { + return None; + } + + let mut offline = false; + for drive in &result.after.drives { + if drive.state == DriveState::Faulty.to_str() { + return Some(Error::Disk(DiskError::FaultyDisk)); + } + offline |= drive.state == DriveState::Offline.to_str(); + } + + offline.then_some(Error::Disk(DiskError::DiskNotFound)) +} + impl HealTask { pub(super) async fn heal_bucket(&self, bucket: &str) -> Result<()> { debug!( @@ -335,13 +351,16 @@ impl HealTask { ) .await { - Ok((result, None)) => { - telemetry_unknown |= !increment_counter(&mut healed); - telemetry_unknown |= - !add_bytes(&mut bytes, u64::try_from(result.object_size).unwrap_or(u64::MAX)); - self.record_result_item(result).await; - None - } + Ok((result, None)) => match unavailable_recreate_error(&result, &heal_opts) { + Some(error) => Some(error), + None => { + telemetry_unknown |= !increment_counter(&mut healed); + telemetry_unknown |= + !add_bytes(&mut bytes, u64::try_from(result.object_size).unwrap_or(u64::MAX)); + self.record_result_item(result).await; + None + } + }, Ok((_, Some(err))) if is_missing_object_dir_heal_result(object, &err) => { telemetry_unknown |= !increment_counter(&mut healed); debug!( diff --git a/crates/heal/src/heal/task/tests.rs b/crates/heal/src/heal/task/tests.rs index 430ad8fa9..1878ca7ea 100644 --- a/crates/heal/src/heal/task/tests.rs +++ b/crates/heal/src/heal/task/tests.rs @@ -706,11 +706,28 @@ enum MockHealObjectOutcome { OkWithOtherError(&'static str), ErrOther(&'static str), DanglingGraceDeferred, + UnavailableDrive(DriveState), RetryableReadQuorum, RetryableSlowDown, PermanentOther(&'static str), } +fn unavailable_drive_heal_result(state: DriveState) -> (HealResultItem, Option) { + ( + HealResultItem { + after: Infos { + drives: vec![HealDriveInfo { + endpoint: "remote-target".to_string(), + state: state.to_string(), + ..Default::default() + }], + }, + ..Default::default() + }, + None, + ) +} + #[derive(Clone, Copy)] enum MockObjectExists { Exists(bool), @@ -813,6 +830,7 @@ impl HealStorageAPI for MockStorage { "dangling object deletion deferred by heal grace window; retry_after_secs=3599; grace_secs=3600", ))), )), + MockHealObjectOutcome::UnavailableDrive(state) => Ok(unavailable_drive_heal_result(state)), MockHealObjectOutcome::RetryableReadQuorum => Err(Error::Storage(EcstoreError::InsufficientReadQuorum( bucket.to_string(), object.to_string(), @@ -833,6 +851,7 @@ impl HealStorageAPI for MockStorage { "dangling object deletion deferred by heal grace window; retry_after_secs=3599; grace_secs=3600", ))), )), + MockHealObjectOutcome::UnavailableDrive(state) => Ok(unavailable_drive_heal_result(state)), MockHealObjectOutcome::OkWithOtherError(message) => Ok((HealResultItem::default(), Some(Error::other(message)))), MockHealObjectOutcome::ErrOther(message) | MockHealObjectOutcome::PermanentOther(message) => { Err(Error::other(message)) @@ -1449,6 +1468,46 @@ 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_when_recreate_target_is_unavailable() { + for state in [DriveState::Offline, DriveState::Faulty] { + let state_name = state.to_string(); + let storage = Arc::new(MockStorage::default()); + storage + .heal_object_outcomes + .lock() + .unwrap() + .insert("object-a".to_string(), VecDeque::from([MockHealObjectOutcome::UnavailableDrive(state)])); + let request = HealRequest::new( + HealType::Bucket { + bucket: "bucket-a".to_string(), + }, + HealOptions { + recursive: true, + recreate_missing: true, + timeout: None, + ..Default::default() + }, + HealPriority::Normal, + ); + let task = HealTask::from_request(request, storage.clone()); + + task.heal_bucket("bucket-a") + .await + .expect("an unavailable recreate target should be retried after it returns"); + + assert_eq!( + storage.heal_object_calls.lock().unwrap().as_slice(), + ["object-a".to_string(), "object-b".to_string(), "object-a".to_string()], + "unexpected calls for unavailable state {state_name}" + ); + let progress = task.get_progress().await; + assert_eq!(progress.objects_scanned, 2, "unexpected scanned count for state {state_name}"); + assert_eq!(progress.objects_healed, 2, "unexpected healed count for state {state_name}"); + assert_eq!(progress.objects_failed, 0, "unexpected failed count for state {state_name}"); + } +} + #[tokio::test(start_paused = true)] async fn recursive_bucket_heal_skips_dangling_delete_grace_without_batch_failure() { let storage = Arc::new(MockStorage::default()); @@ -1486,6 +1545,90 @@ async fn recursive_bucket_heal_skips_dangling_delete_grace_without_batch_failure assert_eq!(progress.skipped_objects, 1); } +#[tokio::test(start_paused = true)] +async fn recursive_bucket_heal_preserves_non_recreate_and_non_availability_results() { + for (dry_run, recreate_missing, state) in [ + (true, true, DriveState::Offline), + (true, true, DriveState::Faulty), + (false, false, DriveState::Offline), + (false, false, DriveState::Faulty), + (false, true, DriveState::Ok), + (false, true, DriveState::Missing), + (false, true, DriveState::Corrupt), + (false, true, DriveState::PermissionDenied), + (false, true, DriveState::Unknown("other failure".to_string())), + ] { + let storage = Arc::new(MockStorage::default()); + storage + .heal_object_outcomes + .lock() + .expect("test outcome lock") + .insert("object-a".to_string(), VecDeque::from([MockHealObjectOutcome::UnavailableDrive(state)])); + let task = HealTask::from_request( + HealRequest::new( + HealType::Bucket { + bucket: "bucket-a".to_string(), + }, + HealOptions { + recursive: true, + dry_run, + recreate_missing, + timeout: None, + ..Default::default() + }, + HealPriority::Normal, + ), + storage.clone(), + ); + task.heal_bucket("bucket-a") + .await + .expect("unchanged best-effort result should not schedule an availability retry"); + assert_eq!( + storage.heal_object_calls.lock().expect("test call lock").as_slice(), + ["object-a".to_string(), "object-b".to_string()] + ); + } +} + +#[tokio::test(start_paused = true)] +async fn recursive_bucket_heal_exhausts_unavailable_target_without_rescanning_healthy_objects() { + let storage = Arc::new(MockStorage::default()); + storage.heal_object_outcomes.lock().expect("test outcome lock").insert( + "object-a".to_string(), + (0..4) + .map(|_| MockHealObjectOutcome::UnavailableDrive(DriveState::Faulty)) + .collect(), + ); + let task = HealTask::from_request( + HealRequest::new( + HealType::Bucket { + bucket: "bucket-a".to_string(), + }, + HealOptions { + recursive: true, + recreate_missing: true, + timeout: None, + ..Default::default() + }, + HealPriority::Normal, + ), + storage.clone(), + ); + let error = task + .heal_bucket("bucket-a") + .await + .expect_err("persistent unavailable target must not report success"); + assert!(matches!(error, Error::TaskExecutionFailed { .. })); + let failure = task + .take_batch_failure() + .await + .expect("exhausted availability failure should be retained"); + assert_eq!((failure.failed, failure.retryable, failure.permanent), (1, 1, 0)); + let calls = storage.heal_object_calls.lock().expect("test call lock"); + 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 test_recursive_bucket_heal_reports_typed_exhausted_and_permanent_failures() { let storage = Arc::new(MockStorage::default()); diff --git a/docs/testing/e2e-suite-inventory.md b/docs/testing/e2e-suite-inventory.md index c7e4938ef..8a52be5af 100644 --- a/docs/testing/e2e-suite-inventory.md +++ b/docs/testing/e2e-suite-inventory.md @@ -57,7 +57,7 @@ | group_delete_test | 4 | | | head_object_consistency_test | 1 | ✅ | | head_object_range_test | 1 | ✅ | -| heal_erasure_disk_rebuild_test | 4 | 🌙 | +| heal_erasure_disk_rebuild_test | 5 | 🌙 | | inline_fast_path_cluster_test | 16 | | | internode_rpc_signature_e2e_test | 5 | | | kms | 50 | | @@ -103,4 +103,4 @@ | tls_hot_reload_test | 1 | ✅ | | version_id_regression_test | 10 | ✅ | -**Total listed: 621 tests across 86 modules · PR smoke: 165 tests / 36 modules · merge/main full: 495 tests / 77 modules · nightly replication: 56 tests · nightly cluster faults: 31 tests / 7 modules · nightly protocols: 16 tests** · updated 2026-08-31. +**Total listed: 622 tests across 86 modules · PR smoke: 165 tests / 36 modules · merge/main full: 495 tests / 77 modules · nightly replication: 56 tests · nightly cluster faults: 32 tests / 7 modules · nightly protocols: 16 tests** · updated 2026-08-31.