From 628663f468f998b6858624fdf975053fc0018cc8 Mon Sep 17 00:00:00 2001 From: cxymds Date: Fri, 16 Oct 2026 12:08:29 +0800 Subject: [PATCH] fix(heal): recover replacements after pool removal (#8600) * fix(heal): recover replacements after pool removal * test(e2e): gate upgrade probes on writable baseline nodes * test(e2e): register baseline readiness regression in smoke digest --- .config/e2e-smoke-selection.txt | 2 +- .../src/upgrade_compatibility_test.rs | 131 +++++- crates/ecstore/src/disk/local.rs | 1 + .../src/disk/local/replacement_lease.rs | 83 ++++ crates/ecstore/src/disk/mod.rs | 8 + crates/ecstore/src/services/rebalance/meta.rs | 25 + crates/ecstore/src/store/heal.rs | 350 +++++++++++++- crates/heal/src/heal/resume.rs | 10 + crates/heal/src/heal/resume/handoff.rs | 106 ++++- crates/heal/src/heal/resume/handoff/tests.rs | 445 +++++++++++++++++- crates/heal/src/heal/storage.rs | 15 + crates/heal/tests/heal_integration_test.rs | 174 +++++++ crates/heal/tests/storage_api.rs | 1 + docs/architecture/compat-cleanup-register.md | 2 + 14 files changed, 1336 insertions(+), 17 deletions(-) diff --git a/.config/e2e-smoke-selection.txt b/.config/e2e-smoke-selection.txt index f76127334..9430a4d40 100644 --- a/.config/e2e-smoke-selection.txt +++ b/.config/e2e-smoke-selection.txt @@ -1 +1 @@ -sha256=0996063fb393d38b5b2d43260b555a01bf9316ecab01cd03d5258c83388bb33b +sha256=ce16185f87c6cd30aa0891ebd22be3352aedc10040d48cc5b00652d72f7be3b5 diff --git a/crates/e2e_test/src/upgrade_compatibility_test.rs b/crates/e2e_test/src/upgrade_compatibility_test.rs index cae6199ff..5eb1fd031 100644 --- a/crates/e2e_test/src/upgrade_compatibility_test.rs +++ b/crates/e2e_test/src/upgrade_compatibility_test.rs @@ -777,6 +777,18 @@ async fn wait_for_upgrade_writes( Ok(()) } +fn previous_baseline_writers(clients: &[Client], restarted_node: usize) -> &[Client] { + // Node 0 has persisted the seed. With restart order [1, 2, 3, 0], only + // this prefix has also cleared rc.5's startup write fence. Every RRS + // probe still needs all drives, including those of not-yet-restarted peers. + let ready_end = if restarted_node == 0 { + clients.len() + } else { + restarted_node + 1 + }; + &clients[..ready_end] +} + async fn prepare_previous_release_baseline(cluster: &mut RustFSTestClusterEnvironment, previous_binary: &Path) -> TestResult { let clients: Vec<_> = cluster.create_all_clients()?.iter().map(upgrade_probe_client).collect(); // rc.5 can latch a non-elected node's write fence while its first startup @@ -787,8 +799,9 @@ async fn prepare_previous_release_baseline(cluster: &mut RustFSTestClusterEnviro // A node that just came back is still replaying its startup traffic against // every peer. Killing the next node during that burst gives the fresh node // two failures in a row and it drops that peer's drive to Offline, where - // only the slow recovery probe can bring it back. Let every node re-admit - // every drive before the next kill and before any current binary starts. + // only the slow recovery probe can bring it back. Let every writable node + // re-admit every drive before the next kill. Nodes still awaiting their + // restart may retain the startup write fence and cannot coordinate a PUT. for node in [1, 2, 3, 0] { cluster.stop_node(node)?; cluster.start_node_from_binary(node, previous_binary).await?; @@ -800,7 +813,13 @@ async fn prepare_previous_release_baseline(cluster: &mut RustFSTestClusterEnviro LISTING_CONVERGENCE_TIMEOUT, ) .await?; - wait_for_drive_readmission(&clients, MIXED_BUCKET, &context, DRIVE_READMISSION_TIMEOUT).await?; + wait_for_drive_readmission( + previous_baseline_writers(&clients, node), + MIXED_BUCKET, + &context, + DRIVE_READMISSION_TIMEOUT, + ) + .await?; } wait_for_upgrade_write_readiness(&clients, MIXED_BUCKET, "previous-baseline", LISTING_CONVERGENCE_TIMEOUT).await?; @@ -872,6 +891,112 @@ async fn parity_zero_write_needs_every_drive() -> TestResult { mod upgrade_write_readiness_tests { use super::*; use crate::fake_s3_target::FaultAction; + use aws_sdk_s3::config::interceptors::BeforeTransmitInterceptorContextRef; + use aws_sdk_s3::config::{ConfigBag, Intercept, RuntimeComponents}; + + #[derive(Debug)] + struct RequireAllDrivesProbe; + + impl Intercept for RequireAllDrivesProbe { + fn name(&self) -> &'static str { + "require-all-drives-probe" + } + + fn read_before_transmit( + &self, + context: &BeforeTransmitInterceptorContextRef<'_>, + _runtime_components: &RuntimeComponents, + _cfg: &mut ConfigBag, + ) -> Result<(), BoxError> { + assert_eq!(context.request().method(), "PUT"); + assert_eq!( + context.request().headers().get("x-amz-storage-class"), + Some("REDUCED_REDUNDANCY"), + "every admitted coordinator must use the all-drive storage class" + ); + Ok(()) + } + } + + #[tokio::test] + async fn previous_baseline_probes_only_writable_coordinators_until_all_restart() -> TestResult { + let mut targets = Vec::new(); + let mut clients = Vec::new(); + for _ in 0..MIXED_NODE_COUNT { + let target = FakeS3Target::start().await?; + target.create_bucket(MIXED_BUCKET); + clients.push(Client::from_conf( + fake_source_client(&target) + .config() + .to_builder() + .interceptor(RequireAllDrivesProbe) + .build(), + )); + targets.push(target); + } + for target in &targets[2..] { + target.inject(FakeTargetOperation::PutObject, FaultAction::Status(StatusCode::INTERNAL_SERVER_ERROR), 10); + } + + // Reproduce the original all-coordinator gate: node 0 and restarted + // node 1 can write, but unrestarted node 2 still returns a latched 500. + let error = wait_for_drive_readmission(&clients, MIXED_BUCKET, "premature", Duration::from_secs(5)) + .await + .expect_err("an unrestarted coordinator must retain its startup write failure"); + assert!(error.to_string().contains("node 2 drive re-admission failed"), "{error}"); + let failed = targets[2].requests(); + assert_eq!(failed.len(), 1, "the startup 500 must fail without retrying"); + assert_eq!(failed[0].fault, Some(FaultAction::Status(StatusCode::INTERNAL_SERVER_ERROR))); + for target in &targets { + target.take_requests(); + } + + for (restarted_node, expected_writers) in [(1, 2), (2, 3), (3, 4), (0, 4)] { + // A restart clears only this process's startup write fence. + targets[restarted_node].clear_faults(); + let phase = format!("previous-restart-{restarted_node}"); + wait_for_drive_readmission( + previous_baseline_writers(&clients, restarted_node), + MIXED_BUCKET, + &phase, + Duration::from_secs(5), + ) + .await?; + for (node, target) in targets.iter().enumerate() { + let key = format!(".upgrade-readiness/{phase}/all-drives/node-{node}"); + let admitted = node < expected_writers; + assert_eq!( + target.count_requests(FakeTargetOperation::PutObject, &key), + usize::from(admitted), + "{phase}: only writable node {node} may coordinate an all-drive probe" + ); + assert_eq!(target.has_object(MIXED_BUCKET, &key), admitted); + assert_eq!(target.take_requests().len(), usize::from(admitted)); + } + } + + // Once every node is admitted, its permanent errors must still fail + // the full gate immediately rather than being skipped or retried. + let phase = "previous-ready-error"; + let key = format!(".upgrade-readiness/{phase}/all-drives/node-3"); + targets[3].inject_for_key( + FakeTargetOperation::PutObject, + &key, + FaultAction::Status(StatusCode::INTERNAL_SERVER_ERROR), + 1, + ); + let error = + wait_for_drive_readmission(previous_baseline_writers(&clients, 0), MIXED_BUCKET, phase, Duration::from_secs(5)) + .await + .expect_err("a recovered coordinator's 500 must still fail the final gate"); + assert!(error.to_string().contains("node 3 drive re-admission failed"), "{error}"); + assert_eq!(targets[3].count_requests(FakeTargetOperation::PutObject, &key), 1); + assert!(!targets[3].has_object(MIXED_BUCKET, &key)); + for target in targets { + target.shutdown().await; + } + Ok(()) + } #[tokio::test] async fn waits_for_each_writer_after_metadata_is_ready() -> TestResult { diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index b7c7ff210..5cc0e5e83 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -38,6 +38,7 @@ mod uring_read_chunks; #[path = "uring_result_budget.rs"] mod uring_result_budget; pub use replacement_lease::ReplacementExecutionLease; +pub(crate) use replacement_lease::is_blank_target as is_blank_replacement_target; use crate::crash_inject::{self, CrashPoint}; use crate::data_usage::local_snapshot::ensure_data_usage_layout; diff --git a/crates/ecstore/src/disk/local/replacement_lease.rs b/crates/ecstore/src/disk/local/replacement_lease.rs index 2a94654d7..3864c490f 100644 --- a/crates/ecstore/src/disk/local/replacement_lease.rs +++ b/crates/ecstore/src/disk/local/replacement_lease.rs @@ -26,6 +26,42 @@ pub struct ReplacementExecutionLease { // file while it is held would allow a second owner to lock a different inode. const EXECUTION_LOCK_FILE: &str = ".rustfs-replacement.lock"; +/// A compatibility retry may only admit a disk that has not received data. +/// The caller retains the mount descriptor and the replacement executor lease. +pub(crate) async fn is_blank_target(root: PathBuf) -> Result { + tokio::task::spawn_blocking(move || { + let mut directories = vec![(root, 0usize)]; + let mut entries_left = 64usize; + while let Some((directory, depth)) = directories.pop() { + for entry in std::fs::read_dir(directory)? { + let entry = entry?; + let Some(remaining) = entries_left.checked_sub(1) else { + return Ok(false); + }; + entries_left = remaining; + let kind = entry.file_type()?; + let name = entry.file_name(); + if depth == 0 && name == EXECUTION_LOCK_FILE && kind.is_file() { + if entry.metadata()?.len() != 0 { + return Ok(false); + } + } else if depth == 1 && name == crate::disk::HEALING_MARKER_PATH && kind.is_file() { + if entry.metadata()?.len() > 256 { + return Ok(false); + } + } else if kind.is_dir() && depth < 4 && (depth > 0 || name == crate::disk::RUSTFS_META_BUCKET) { + directories.push((entry.path(), depth + 1)); + } else { + return Ok(false); + } + } + } + Ok(true) + }) + .await + .map_err(DiskError::from)? +} + pub(super) async fn acquire(root: PathBuf) -> Result> { #[cfg(unix)] { @@ -56,6 +92,53 @@ pub(super) async fn acquire(root: PathBuf) -> Result Err(DiskError::other("replacement execution requires a local target")), } } + + pub(crate) async fn is_blank_replacement_target(&self) -> Result { + let Some(root) = self.replacement_mount_lease_root() else { + return Ok(false); + }; + let blank = local::is_blank_replacement_target(root).await?; + Ok(blank && self.has_replacement_mount_lease()) + } } pub async fn new_disk(ep: &Endpoint, opt: &DiskOption) -> Result { diff --git a/crates/ecstore/src/services/rebalance/meta.rs b/crates/ecstore/src/services/rebalance/meta.rs index 8983f40cd..54f5efe8e 100644 --- a/crates/ecstore/src/services/rebalance/meta.rs +++ b/crates/ecstore/src/services/rebalance/meta.rs @@ -71,6 +71,31 @@ impl From for RebalStatus { } impl RebalanceMeta { + /// A historical run has no authority over the current deployment's pool + /// indices. Stop timestamps alone do not prove that its workers stopped. + pub(crate) fn has_terminal_topology(&self) -> bool { + !self.id.is_empty() + && !self.pool_stats.is_empty() + && self.pool_stats.iter().all(|pool| { + if pool.info.stopping + || !matches!(pool.info.status, RebalStatus::Completed | RebalStatus::Stopped | RebalStatus::Failed) + && (pool.participating || pool.info.status != RebalStatus::None) + { + return false; + } + let Some(message) = pool.info.last_error.as_deref() else { + return true; + }; + if !is_rebalance_stop_propagation_error(Some(message)) { + return true; + } + decode_rebalance_stop_propagation_record(message).is_some_and(|record| { + !record.has_failures() + && (record.stop_attempt_at.or(self.stopped_at).is_none() || record.terminal_reload_attempt_at.is_some()) + }) + }) + } + pub fn new() -> Self { Self::default() } diff --git a/crates/ecstore/src/store/heal.rs b/crates/ecstore/src/store/heal.rs index 943f6be34..46500632f 100644 --- a/crates/ecstore/src/store/heal.rs +++ b/crates/ecstore/src/store/heal.rs @@ -117,6 +117,47 @@ enum HealFormatPoolSkip { Retryable, } +#[derive(Debug)] +enum HealFormatRebalance { + Missing, + Current(RebalanceMeta), + Historical, +} + +impl HealFormatRebalance { + fn current(&self) -> Option<&RebalanceMeta> { + match self { + Self::Current(meta) => Some(meta), + Self::Missing | Self::Historical => None, + } + } +} + +#[derive(Debug, thiserror::Error)] +#[error( + "format repair blocked: rebalance metadata describes {stored_pools} pools but the deployment has {current_pools}; the historical run is not confirmed terminal" +)] +struct HealFormatRebalanceTopologyMismatch { + stored_pools: usize, + current_pools: usize, +} + +fn classify_heal_format_rebalance(meta: Option, current_pools: usize) -> Result { + let Some(meta) = meta else { + return Ok(HealFormatRebalance::Missing); + }; + if meta.pool_stats.len() == current_pools { + return Ok(HealFormatRebalance::Current(meta)); + } + if meta.has_terminal_topology() { + return Ok(HealFormatRebalance::Historical); + } + Err(Error::other(HealFormatRebalanceTopologyMismatch { + stored_pools: meta.pool_stats.len(), + current_pools, + })) +} + fn classify_heal_format_pool( pool_idx: usize, pool_cmd_line: &str, @@ -487,7 +528,7 @@ impl ECStore { NamespaceLockGuard, NamespaceLockGuard, PoolMeta, - Option, + HealFormatRebalance, )> { let metadata_pool = self .pools @@ -543,12 +584,9 @@ impl ECStore { Err(err) => return Err(err), }; - if rebalance_meta - .as_ref() - .is_some_and(|meta| meta.pool_stats.len() != self.pools.len()) - { - return Err(heal_format_fence_lost_error()); - } + // Keep historical status on disk, but never use its obsolete indices + // to gate the current pools. The current pool identity remains fenced. + let rebalance_meta = classify_heal_format_rebalance(rebalance_meta, self.pools.len())?; write_state.ensure_write_safe("heal format fence failed")?; if pool_guard.is_lock_lost() || rebalance_guard.is_lock_lost() { @@ -751,7 +789,8 @@ impl ECStore { first_error.get_or_insert(heal_format_fence_lost_error()); break; } - if let Some(skip) = classify_heal_format_pool(pool_idx, &pool.endpoints.cmd_line, &pool_meta, rebalance_meta.as_ref()) + if let Some(skip) = + classify_heal_format_pool(pool_idx, &pool.endpoints.cmd_line, &pool_meta, rebalance_meta.current()) { if matches!(skip, HealFormatPoolSkip::Completed) { count_completed += 1; @@ -825,6 +864,74 @@ impl ECStore { Ok((r, None)) } + /// Read-only admission for one legacy pre-format recovery handoff. The + /// caller holds the target execution leases and validates their owners. + pub async fn replacement_preformat_retry_is_safe( + &self, + pool_index: usize, + set_index: usize, + targets: &[String], + ) -> Result { + let pool = self + .pools + .get(pool_index) + .ok_or_else(|| invalid_heal_pool_index(pool_index, self.pools.len()))?; + let Some(set) = pool.disk_set.get(set_index) else { + return Err(StorageError::PreconditionFailed); + }; + let (write_state, pool_guard, rebalance_guard, pool_meta, rebalance_meta) = self.acquire_heal_format_fence().await?; + if !matches!(rebalance_meta, HealFormatRebalance::Historical) + || classify_heal_format_pool(pool_index, &pool.endpoints.cmd_line, &pool_meta, None).is_some() + || targets.is_empty() + { + return Ok(false); + } + + let mut slots = BTreeSet::new(); + let mut target_disks = Vec::with_capacity(targets.len()); + for target in targets { + let Some((slot, endpoint)) = set + .set_endpoints + .iter() + .enumerate() + .find(|(_, endpoint)| endpoint.to_string() == *target) + else { + return Ok(false); + }; + if !endpoint.is_local || !slots.insert(slot) { + return Ok(false); + } + // Open the current mounted target; a cached disk handle can still + // refer to the previous disk at the same endpoint. + let target_disk = disk::new_disk( + endpoint, + &disk::DiskOption { + cleanup: false, + health_check: false, + }, + ) + .await?; + if !target_disk.is_blank_replacement_target().await? { + return Ok(false); + } + target_disks.push(target_disk); + } + + let (_, error) = set + .heal_replacement_format_with_fence(true, targets, || pool_guard.is_lock_lost() || rebalance_guard.is_lock_lost()) + .await?; + write_state.ensure_write_safe("pre-format recovery admission fence failed")?; + if pool_guard.is_lock_lost() || rebalance_guard.is_lock_lost() { + return Err(heal_format_fence_lost_error()); + } + for target_disk in target_disks { + if !target_disk.is_blank_replacement_target().await? { + return Ok(false); + } + } + Ok(error.is_none()) + } + #[instrument(skip(self, targets), fields(pool_index, set_index, target_count = targets.len()))] pub async fn heal_replacement_format( &self, @@ -846,7 +953,8 @@ impl ECStore { })?; let (mut write_state, pool_guard, rebalance_guard, pool_meta, rebalance_meta) = self.acquire_heal_format_fence().await?; - if let Some(skip) = classify_heal_format_pool(pool_index, &pool.endpoints.cmd_line, &pool_meta, rebalance_meta.as_ref()) { + if let Some(skip) = classify_heal_format_pool(pool_index, &pool.endpoints.cmd_line, &pool_meta, rebalance_meta.current()) + { return Ok((HealResultItem::default(), Some(heal_format_pool_skip_error(skip)))); } @@ -2137,9 +2245,13 @@ mod tests { } async fn multi_pool_heal_store() -> (tempfile::TempDir, Arc, CancellationToken) { + heal_store_with_pool_count(2).await + } + + async fn heal_store_with_pool_count(pool_count: usize) -> (tempfile::TempDir, Arc, CancellationToken) { let temp_dir = tempfile::tempdir().expect("multi-pool heal test directory should be created"); let mut pool_endpoints = Vec::new(); - for pool_index in 0..2 { + for pool_index in 0..pool_count { let mut endpoints = Vec::new(); for disk_index in 0..4 { let disk_path = temp_dir.path().join(format!("pool{pool_index}-disk{disk_index}")); @@ -2456,6 +2568,224 @@ mod tests { ); } + fn completed_two_pool_rebalance() -> RebalanceMeta { + RebalanceMeta { + id: Uuid::new_v4().to_string(), + pool_stats: vec![ + RebalanceStats { + participating: true, + info: RebalanceInfo { + status: RebalStatus::Completed, + ..Default::default() + }, + ..Default::default() + }, + RebalanceStats::default(), + ], + ..Default::default() + } + } + + #[test] + fn historical_rebalance_topology_requires_every_pool_to_be_terminal() { + use crate::services::rebalance::{RebalanceStopPropagationRecord, encode_rebalance_stop_propagation_record}; + + assert!(matches!( + classify_heal_format_rebalance(Some(completed_two_pool_rebalance()), 1), + Ok(HealFormatRebalance::Historical) + )); + // Expansion has the same historical-index problem as removal. + assert!(matches!( + classify_heal_format_rebalance(Some(completed_two_pool_rebalance()), 3), + Ok(HealFormatRebalance::Historical) + )); + assert!(matches!(classify_heal_format_rebalance(None, 1), Ok(HealFormatRebalance::Missing))); + for status in [RebalStatus::Completed, RebalStatus::Stopped, RebalStatus::Failed] { + let mut meta = completed_two_pool_rebalance(); + meta.pool_stats[0].info.status = status; + assert!(meta.has_terminal_topology(), "terminal status {status:?}"); + } + + let mut invalid = Vec::new(); + for index in 0..2 { + let mut active = completed_two_pool_rebalance(); + active.stopped_at = Some(OffsetDateTime::UNIX_EPOCH); + active.pool_stats[index].info.status = RebalStatus::Started; + invalid.push(active); + let mut stopping = completed_two_pool_rebalance(); + stopping.pool_stats[index].info.stopping = true; + invalid.push(stopping); + } + let mut unknown_participant = completed_two_pool_rebalance(); + unknown_participant.pool_stats[0].info.status = RebalStatus::None; + invalid.push(unknown_participant); + let mut no_id = completed_two_pool_rebalance(); + no_id.id.clear(); + invalid.push(no_id); + invalid.push(RebalanceMeta::default()); + for record in [ + RebalanceStopPropagationRecord { + stop_attempt_at: Some(OffsetDateTime::UNIX_EPOCH), + ..Default::default() + }, + RebalanceStopPropagationRecord { + stop_failures: vec!["peer".to_string()], + ..Default::default() + }, + RebalanceStopPropagationRecord { + terminal_reload_attempt_at: Some(OffsetDateTime::UNIX_EPOCH), + terminal_reload_failures: vec!["peer".to_string()], + ..Default::default() + }, + ] { + let mut meta = completed_two_pool_rebalance(); + meta.pool_stats[1].info.last_error = Some(encode_rebalance_stop_propagation_record(&record)); + invalid.push(meta); + } + let mut malformed = completed_two_pool_rebalance(); + malformed.pool_stats[0].info.last_error = Some("rebalance stop propagation incomplete: invalid".to_string()); + invalid.push(malformed); + for meta in invalid { + let error = classify_heal_format_rebalance(Some(meta), 1).expect_err("uncertain history must block repair"); + assert!(error.to_string().contains("not confirmed terminal")); + assert!(!error.to_string().contains("request rate")); + } + let mut acknowledged = completed_two_pool_rebalance(); + acknowledged.pool_stats[0].info.last_error = + Some(encode_rebalance_stop_propagation_record(&RebalanceStopPropagationRecord { + stop_attempt_at: Some(OffsetDateTime::UNIX_EPOCH), + terminal_reload_attempt_at: Some(OffsetDateTime::UNIX_EPOCH), + ..Default::default() + })); + assert!(acknowledged.has_terminal_topology()); + } + + #[tokio::test] + #[serial_test::serial] + async fn replacement_format_after_historical_pool_removal() { + let (temp_dir, store, shutdown) = heal_store_with_pool_count(1).await; + let mut meta = completed_two_pool_rebalance(); + meta.save(store.pools[0].clone()) + .await + .expect("persist historical two-pool run"); + let (before, _) = read_config_no_lock_preserve_empty_with_metadata(store.pools[0].clone(), REBAL_META_NAME) + .await + .expect("read historical run before repair"); + let target = remove_heal_test_format(&temp_dir, &store, 0, 3).await; + + // Even a worker in a removed pool must fence every current format write. + meta.pool_stats[1].participating = true; + meta.pool_stats[1].info.status = RebalStatus::Started; + meta.save(store.pools[0].clone()) + .await + .expect("persist active historical pool"); + let err = store + .heal_replacement_format(false, 0, 0, std::slice::from_ref(&target)) + .await + .expect_err("removed active source must block format repair"); + assert!(err.to_string().contains("2 pools but the deployment has 1")); + assert_heal_test_format_missing(&temp_dir, 0, 3).await; + meta.pool_stats[1] = RebalanceStats::default(); + meta.save(store.pools[0].clone()) + .await + .expect("persist terminal history again"); + + let (result, error) = store + .heal_replacement_format(false, 0, 0, std::slice::from_ref(&target)) + .await + .expect("terminal historical topology must allow target format repair"); + assert!(error.is_none(), "target repair: {error:?}"); + assert!(result.after.drives.iter().all(|drive| !drive.uuid.is_empty())); + let disk = store.pools[0].disk_set[0].disks.read().await[3] + .clone() + .expect("replacement disk"); + let repaired = load_format_erasure(&disk, true).await.expect("read repaired target format"); + assert!(!repaired.erasure.this.is_nil()); + + remove_heal_test_format(&temp_dir, &store, 0, 3).await; + let (_, error) = store + .handle_heal_format(false) + .await + .expect("full format heal uses the same historical fence"); + assert!(error.is_none(), "full format repair: {error:?}"); + assert!(heal_test_format_path(&temp_dir, 0, 3).exists()); + let (after, _) = read_config_no_lock_preserve_empty_with_metadata(store.pools[0].clone(), REBAL_META_NAME) + .await + .expect("historical status remains readable"); + assert_eq!(before, after, "format repair must not rewrite historical rebalance status"); + shutdown.cancel(); + } + + #[tokio::test] + #[serial_test::serial] + async fn preformat_retry_requires_historical_topology_and_a_blank_target() { + let (temp_dir, store, shutdown) = heal_store_with_pool_count(1).await; + completed_two_pool_rebalance() + .save(store.pools[0].clone()) + .await + .expect("persist historical two-pool run"); + let target = remove_heal_test_format(&temp_dir, &store, 0, 3).await; + assert!( + !store + .replacement_preformat_retry_is_safe(0, 0, std::slice::from_ref(&target)) + .await + .expect("inspect target with old metadata"), + "missing format alone must not admit a disk containing old metadata" + ); + let target_path = temp_dir.path().join("pool0-disk3"); + for entry in std::fs::read_dir(&target_path).expect("list target fixture") { + let entry = entry.expect("target fixture entry"); + if entry.file_type().expect("fixture entry type").is_dir() { + std::fs::remove_dir_all(entry.path()).expect("clear replacement fixture directory"); + } else { + std::fs::remove_file(entry.path()).expect("clear replacement fixture file"); + } + } + let target_disk = store.pools[0].disk_set[0].disks.read().await[3] + .clone() + .expect("replacement disk"); + if !cfg!(target_os = "linux") { + assert!(target_disk.acquire_replacement_execution_lease().await.is_err()); + assert!( + !store + .replacement_preformat_retry_is_safe(0, 0, &[target]) + .await + .expect("platforms without pinned mount identity must reject compatibility recovery") + ); + shutdown.cancel(); + return; + } + let _lease = target_disk + .acquire_replacement_execution_lease() + .await + .expect("own replacement executor"); + assert!( + store + .replacement_preformat_retry_is_safe(0, 0, std::slice::from_ref(&target)) + .await + .expect("inspect blank target") + ); + assert_heal_test_format_missing(&temp_dir, 0, 3).await; + std::fs::create_dir(target_path.join("user-bucket")).expect("simulate residual user bucket"); + assert!( + !store + .replacement_preformat_retry_is_safe(0, 0, std::slice::from_ref(&target)) + .await + .expect("inspect nonblank target") + ); + std::fs::remove_dir(target_path.join("user-bucket")).expect("remove fixture bucket"); + delete_config(store.pools[0].clone(), REBAL_META_NAME) + .await + .expect("remove historical run"); + assert!( + !store + .replacement_preformat_retry_is_safe(0, 0, &[target]) + .await + .expect("ordinary replacement has no compatibility authority") + ); + shutdown.cancel(); + } + #[tokio::test] #[serial_test::serial] async fn replacement_format_heal_coexists_with_capacity_read_fence() { diff --git a/crates/heal/src/heal/resume.rs b/crates/heal/src/heal/resume.rs index 20bc4339d..575922faf 100644 --- a/crates/heal/src/heal/resume.rs +++ b/crates/heal/src/heal/resume.rs @@ -368,6 +368,11 @@ pub struct ResumeState { /// and may receive one bounded compatibility re-arm. #[serde(default)] pub replacement_legacy_retry_compatibility_done: bool, + /// New generations cannot consume the one-shot compatibility handoff for + /// a pre-format intent stranded by historical rebalance topology. Missing + /// fields in older schema-seven snapshots remain eligible for validation. + #[serde(default)] + pub replacement_preformat_handoff_retry_done: bool, /// The current pass was rewound because the replacement mount was not /// ready. Task-level error persistence must not turn this readiness wait /// back into a consumed retry attempt. @@ -455,6 +460,7 @@ impl ResumeState { replacement_legacy_import: None, replacement_legacy_successor: None, replacement_legacy_retry_compatibility_done: false, + replacement_preformat_handoff_retry_done: false, replacement_retry_waiting_for_target: false, dangling_delete_retry_not_before: None, start_time: SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs(), @@ -500,6 +506,7 @@ impl ResumeState { // New generations already use readiness-aware persistence and must // never enter the one-shot compatibility path for legacy snapshots. state.replacement_legacy_retry_compatibility_done = true; + state.replacement_preformat_handoff_retry_done = true; state } @@ -1272,6 +1279,9 @@ impl ResumeManager { if state.max_retries == 0 { return Ok(false); } + if self.rearm_preformat_handoff_retry_if_needed(storage).await? { + return Ok(true); + } let legacy_rearm = replacement_retry_is_legacy_transient_skip(&state); let ready = storage .replacement_targets_ready_for_retry( diff --git a/crates/heal/src/heal/resume/handoff.rs b/crates/heal/src/heal/resume/handoff.rs index 47f3a0bdf..001cc71aa 100644 --- a/crates/heal/src/heal/resume/handoff.rs +++ b/crates/heal/src/heal/resume/handoff.rs @@ -89,6 +89,65 @@ pub(super) fn validate_lineage(task_id: &str, set_disk_id: &str, lineage: &[Repl } impl ResumeManager { + // RUSTFS_COMPAT_TODO(8565): recover pre-format intents stranded by historical pool topology. Remove after supported upgrades no longer admit snapshots lacking replacement_preformat_handoff_retry_done and retained affected intents are resolved. + pub(super) async fn rearm_preformat_handoff_retry_if_needed( + &self, + storage: &dyn crate::heal::storage::HealStorageAPI, + ) -> Result { + let state = self.get_state().await; + if !preformat_handoff_retry_candidate(&state) { + return Ok(false); + } + let identities = storage.replacement_target_identities(&state.replacement_targets).await?; + if identities == state.replacement_target_identities { + return Ok(false); + } + // Lock order: target execution leases, the store's format fence during + // the probe, then the intent CAS after the probe releases its fence. + let execution = storage.replacement_execution(&state.replacement_targets).await?; + if execution.identities() != identities.as_slice() { + return Err(Error::ReplacementTargetNotReady( + "target changed before pre-format recovery admission".to_string(), + )); + } + // A cleared summary does not prove an unstarted pass. Any checkpoint, + // including an empty/corrupt file or an isolated checkpoint marker, + // prevents this compatibility admission. + for suffix in [RESUME_CHECKPOINT_FILE, RESUME_CHECKPOINT_BLOCKED_FILE] { + let path = Path::new(BUCKET_META_PREFIX).join(format!("{}_{suffix}", state.task_id)); + match self.disk.read_all(RUSTFS_META_BUCKET, path_to_str(&path)?).await { + Ok(_) => return Ok(false), + Err(DiskError::FileNotFound) => {} + Err(error) => return Err(error.into()), + } + } + if Self::replacement_completion_proof_if_present(self.disk.clone(), &state.task_id) + .await? + .is_some() + || !storage + .replacement_preformat_retry_is_safe(&state.set_disk_id, &state.replacement_targets) + .await? + { + return Ok(false); + } + let buckets = storage.list_buckets().await?.into_iter().map(|bucket| bucket.name).collect(); + if storage + .replacement_target_identities(&state.replacement_targets) + .await? + .as_slice() + != execution.identities() + { + return Err(Error::ReplacementTargetNotReady( + "target changed during pre-format recovery admission".to_string(), + )); + } + // Retain the execution leases through validation and the Prepared CAS. + // Budget admission, its one-shot marker, and its reserved successor are + // one durable decision, so a crash cannot grant an unbound retry. + self.prepare_replacement_handoff_inner(&execution, buckets, true).await?; + Ok(true) + } + /// Shared startup/scanner decision. A changed mount starts a full scan in /// one durable successor; an unfinished transfer always reuses that UUID. pub(crate) async fn resolve_replacement_recovery( @@ -184,11 +243,23 @@ impl ResumeManager { /// Select a successor only after every old executor is fenced by the target /// execution leases. The durable predecessor is the sole handoff authority. pub(crate) async fn prepare_replacement_handoff( + &self, + execution: &ReplacementExecution, + buckets: Vec, + ) -> Result { + self.prepare_replacement_handoff_inner(execution, buckets, false).await + } + + async fn prepare_replacement_handoff_inner( &self, execution: &ReplacementExecution, mut buckets: Vec, + preformat_retry: bool, ) -> Result { let state = self.get_state().await; + if preformat_retry && !preformat_handoff_retry_candidate(&state) { + return Err(replacement_recovery_conflict("pre-format recovery eligibility changed")); + } if let Some(handoff) = &state.replacement_handoff { Self::validate_handoff(handoff, &state)?; let raw = Self::read_state_file(&self.disk, &state.task_id, self.state_file).await?; @@ -219,7 +290,7 @@ impl ResumeManager { "replacement generation is not eligible for an online handoff", )); } - if state.retry_count.saturating_add(1) >= state.max_retries { + if !preformat_retry && state.retry_count.saturating_add(1) >= state.max_retries { return Err(Error::ReplacementRetryBudgetExhausted); } if state.replacement_phase == ReplacementPhase::OwnershipPending { @@ -280,6 +351,12 @@ impl ResumeManager { if current.replacement_revision != state.replacement_revision { return Err(replacement_recovery_conflict("replacement changed during handoff preparation")); } + if preformat_retry { + current.replacement_preformat_handoff_retry_done = true; + // One handoff consumes one attempt; its successor must still + // have exactly one actual rebuild attempt under the same cap. + current.retry_count = current.max_retries - 2; + } current.replacement_handoff = Some(ReplacementHandoff { phase: ReplacementHandoffPhase::Prepared, link, @@ -378,6 +455,8 @@ impl ResumeManager { link.targets.clone(), ); state.replacement_phase = ReplacementPhase::OwnershipPending; + // The constructor marks this compatibility grant consumed for every + // successor, including descendants of a recovered older generation. state.replacement_predecessor = Some(predecessor.task_id.clone()); state.retry_count = predecessor .retry_count @@ -493,6 +572,31 @@ impl ResumeManager { } } +fn preformat_handoff_retry_candidate(state: &ResumeState) -> bool { + replacement_retry_is_exhausted_active(state) + && state.replacement_generation.as_deref() == Some(state.task_id.as_str()) + && !state.replacement_preformat_handoff_retry_done + && state.replacement_phase == ReplacementPhase::Intent + && state.replacement_execution_protocol == 1 + && state.max_retries >= 2 + && state.replacement_predecessor.is_none() + && state.replacement_handoff.is_none() + && state.replacement_lineage.is_empty() + && state.replacement_legacy_import.is_none() + && state.replacement_legacy_successor.is_none() + && state.resume_cursor.is_none() + && state.current_bucket.is_none() + && state.current_object.is_none() + && state.completed_buckets.is_empty() + && state.processed_objects == 0 + && state.successful_objects == 0 + && state.failed_objects == 0 + && state.skipped_objects == 0 + && state.skipped_new_versions == 0 + && state.skipped_ilm_expired == 0 + && state.processed_bytes == 0 +} + fn successor_can_be_rebound(state: &ResumeState, predecessor: &str) -> bool { state.replacement_predecessor.as_deref() == Some(predecessor) && successor_has_no_progress(state) } diff --git a/crates/heal/src/heal/resume/handoff/tests.rs b/crates/heal/src/heal/resume/handoff/tests.rs index a6aaf9442..a358a0579 100644 --- a/crates/heal/src/heal/resume/handoff/tests.rs +++ b/crates/heal/src/heal/resume/handoff/tests.rs @@ -13,10 +13,451 @@ // limitations under the License. use super::*; -use crate::heal::{HEALING_MARKER_PATH, resume::tests::schema_test_disk}; -use std::sync::LazyLock; +use crate::heal::{ + HEALING_MARKER_PATH, HealObjectInfo, + resume::tests::schema_test_disk, + storage::{HealListItem, HealStorageAPI}, + storage_api::storage::{BucketInfo, DiskSetSelector}, +}; +use rustfs_heal_contracts::heal_channel::HealOpts; +use rustfs_madmin::heal_commands::HealResultItem; +use std::sync::{ + LazyLock, + atomic::{AtomicUsize, Ordering}, +}; static FAILURES: LazyLock>> = LazyLock::new(|| Mutex::new(HashMap::new())); + +struct PreformatRecoveryStorage { + execution: Arc, + historical_topology: bool, + probe_calls: AtomicUsize, + identity_calls: AtomicUsize, + change_identity_on_call: usize, +} + +#[async_trait::async_trait] +impl HealStorageAPI for PreformatRecoveryStorage { + async fn replacement_target_identities(&self, _: &[String]) -> Result> { + let mut identities = self.execution.identities().to_vec(); + if self.identity_calls.fetch_add(1, Ordering::SeqCst) == self.change_identity_on_call { + identities[0].filesystem_identity = "another-mount:fs:root".to_string(); + } + Ok(identities) + } + + async fn replacement_execution(&self, _: &[String]) -> Result> { + Ok(self.execution.clone()) + } + + async fn replacement_targets_ready_for_retry( + &self, + _: &str, + _: &[String], + expected: &[ReplacementTargetIdentity], + ) -> Result { + if expected != self.execution.identities() { + return Ok(false); + } + for disk in self.execution.test_disks() { + if !matches!(disk.get_disk_id().await, Ok(Some(id)) if !id.is_nil()) { + return Ok(false); + } + } + Ok(true) + } + + async fn replacement_preformat_retry_is_safe(&self, _: &str, _: &[String]) -> Result { + self.probe_calls.fetch_add(1, Ordering::SeqCst); + if !self.historical_topology { + return Ok(false); + } + for disk in self.execution.test_disks() { + match disk.read_all(RUSTFS_META_BUCKET, "format.json").await { + Err(DiskError::FileNotFound | DiskError::VolumeNotFound) => {} + Ok(_) => return Ok(false), + Err(error) => return Err(error.into()), + } + } + Ok(true) + } + + async fn list_buckets(&self) -> Result> { + Ok(Vec::new()) + } + + async fn get_object_meta(&self, _: &str, _: &str) -> Result> { + Err(Error::other("pre-format recovery must not inspect objects")) + } + + async fn ec_decode_rebuild(&self, _: &str, _: &str) -> Result> { + Err(Error::other("pre-format recovery must not rebuild objects")) + } + + async fn get_bucket_info(&self, _: &str) -> Result> { + Err(Error::other("pre-format recovery must not inspect buckets")) + } + + async fn list_buckets_for_admin_heal(&self, _: Option) -> Result> { + Err(Error::other("pre-format recovery must not start admin discovery")) + } + + async fn object_exists(&self, _: &str, _: &str) -> Result { + Err(Error::other("pre-format recovery must not inspect objects")) + } + + async fn heal_object(&self, _: &str, _: &str, _: Option<&str>, _: &HealOpts) -> Result<(HealResultItem, Option)> { + Err(Error::other("pre-format recovery must not heal objects")) + } + + async fn heal_bucket(&self, _: &str, _: &HealOpts) -> Result { + Err(Error::other("pre-format recovery must not heal buckets")) + } + + async fn heal_format(&self, _: bool) -> Result<(HealResultItem, Option)> { + Err(Error::other("pre-format recovery must not write formats")) + } + + async fn list_objects_for_heal_page( + &self, + _: &str, + _: &str, + _: Option<&str>, + _: bool, + ) -> Result<(Vec, Option, bool)> { + Err(Error::other("pre-format recovery must not traverse objects")) + } + + async fn get_disk_for_resume(&self, _: &str) -> Result { + Err(Error::other("pre-format recovery must retain its durable anchor")) + } +} + +async fn preformat_recovery_fixture() -> (Vec, ResumeManager, PreformatRecoveryStorage) { + let (dirs, manager, execution) = fixture(1).await; + execution.test_disks()[0] + .delete(RUSTFS_META_BUCKET, HEALING_MARKER_PATH, Default::default()) + .await + .expect("remove the fixture marker from the blank replacement"); + { + let mut state = manager.state.write().await; + state.retry_count = state.max_retries; + state.replacement_preformat_handoff_retry_done = false; + state.replacement_retry_waiting_for_target = true; + } + manager.save_state_strict().await.expect("persist exhausted older intent"); + ( + dirs, + manager, + PreformatRecoveryStorage { + execution, + historical_topology: true, + probe_calls: AtomicUsize::new(0), + identity_calls: AtomicUsize::new(0), + change_identity_on_call: usize::MAX, + }, + ) +} + +#[tokio::test] +async fn preformat_recovery_same_mount_reuses_existing_readiness_rearm() { + let (_dirs, manager, mut storage) = preformat_recovery_fixture().await; + let initial = manager.get_state().await; + storage.execution = + ReplacementExecution::for_test(storage.execution.test_disks().to_vec(), initial.replacement_target_identities.clone()); + assert!(matches!( + storage.execution.test_disks()[0].get_disk_id().await, + Err(DiskError::UnformattedDisk) + )); + assert!( + manager + .rearm_replacement_recovery_if_needed(&storage) + .await + .expect("rearm blank same mount") + ); + let resumed = manager + .resolve_replacement_recovery(&storage) + .await + .expect("resume the same generation"); + assert_eq!(resumed.task_id, initial.task_id); + assert_eq!(resumed.retry_count, resumed.max_retries - 1); + assert!(resumed.replacement_handoff.is_none()); + assert!(!resumed.replacement_preformat_handoff_retry_done); + assert_eq!(storage.probe_calls.load(Ordering::SeqCst), 0); +} + +#[tokio::test] +async fn preformat_recovery_reserves_one_handoff_and_one_rebuild() { + let (_dirs, manager, mut storage) = preformat_recovery_fixture().await; + assert!( + manager + .rearm_replacement_recovery_if_needed(&storage) + .await + .expect("admit historical pre-format recovery") + ); + let prepared = manager.get_state().await; + assert_eq!(prepared.replacement_phase, ReplacementPhase::HandoffPending); + assert!(prepared.replacement_preformat_handoff_retry_done); + let child_state = manager + .resolve_replacement_recovery(&storage) + .await + .expect("resolve the reserved successor"); + assert_eq!(child_state.retry_count, child_state.max_retries - 1); + assert!(child_state.replacement_preformat_handoff_retry_done); + assert_eq!(child_state.replacement_lineage.len(), 1); + let child = ResumeManager::load_replacement_intent(manager.disk.clone(), &child_state.task_id) + .await + .expect("reload the admitted successor"); + // Unlike a generic schema-test disk, an admitted replacement has an exact + // set/slot binding that get_disk_id validates against its persisted format. + let mut endpoint = storage.execution.test_disks()[0].endpoint(); + endpoint.set_pool_index(0); + endpoint.set_set_index(0); + endpoint.set_disk_index(0); + let disk = crate::heal::new_disk( + &endpoint, + &crate::heal::DiskOption { + cleanup: false, + health_check: false, + }, + ) + .await + .expect("open the replacement with its real erasure-set slot"); + storage.execution = ReplacementExecution::for_test(vec![disk.clone()], storage.execution.identities().to_vec()); + let disk_id = Uuid::new_v4(); + let format = serde_json::json!({ + "version": "1", "format": "xl", "id": Uuid::new_v4(), + "xl": {"version": "3", "this": disk_id, "sets": [[disk_id]], "distributionAlgo": "SIPMOD+PARITY"} + }); + storage.execution.test_disks()[0] + .write_all( + RUSTFS_META_BUCKET, + "format.json", + serde_json::to_vec(&format).expect("serialize target format").into(), + ) + .await + .expect("format the admitted target before rebuilding"); + crate::heal::storage_api::EcstoreDiskAPI::set_disk_id(disk.as_ref(), Some(disk_id)) + .await + .expect("publish the formatted disk identity as save_format_file does"); + assert_eq!(disk.get_disk_id().await.expect("formatted replacement must be ready"), Some(disk_id)); + child + .acquire_replacement_markers(&storage.execution) + .await + .expect("commit successor ownership"); + child + .mark_replacement_rebuilding(storage.execution.identities().to_vec()) + .await + .expect("admit actual rebuilding"); + child + .record_replacement_failure(&Error::other("actual rebuild failed"), child_state.max_retries) + .await + .expect("exhaust final attempt"); + assert!( + !child + .rearm_replacement_recovery_if_needed(&storage) + .await + .expect("never regrant the successor") + ); + let reloaded = ResumeManager::load_replacement_intent(manager.disk.clone(), &prepared.task_id) + .await + .expect("reload predecessor"); + assert!( + !reloaded + .rearm_preformat_handoff_retry_if_needed(&storage) + .await + .expect("never regrant the predecessor") + ); + assert_eq!(storage.probe_calls.load(Ordering::SeqCst), 1); +} + +#[tokio::test] +async fn preformat_recovery_replays_prepared_and_published_crashes_without_new_credit() { + for boundary in ["prepared", "successor_published"] { + let (_dirs, manager, storage) = preformat_recovery_fixture().await; + let source = manager.get_state().await.task_id; + FAILURES + .lock() + .expect("failure injection lock") + .insert(source.clone(), boundary); + assert!(manager.rearm_replacement_recovery_if_needed(&storage).await.is_err()); + let reloaded = ResumeManager::load_replacement_intent(manager.disk.clone(), &source) + .await + .expect("reload crashed grant"); + let prepared = reloaded.get_state().await; + assert!(prepared.replacement_preformat_handoff_retry_done); + let successor = prepared + .replacement_handoff + .expect("grant and successor are atomic") + .link + .successor; + assert!( + !reloaded + .rearm_preformat_handoff_retry_if_needed(&storage) + .await + .expect("grant already used") + ); + let recovered = reloaded + .resolve_replacement_recovery(&storage) + .await + .expect("replay the reserved handoff"); + assert_eq!(recovered.task_id, successor); + assert_eq!(recovered.retry_count, recovered.max_retries - 1); + assert_eq!(storage.probe_calls.load(Ordering::SeqCst), 1); + } +} + +#[tokio::test] +async fn preformat_recovery_rejects_progress_checkpoints_formats_and_unrelated_topology() { + for rejection in [ + "progress", + "checkpoint", + "blocked", + "completion_proof", + "format", + "topology", + "new_generation", + "single_retry", + ] { + let (_dirs, manager, mut storage) = preformat_recovery_fixture().await; + let source = manager.get_state().await.task_id; + match rejection { + "progress" => manager.state.write().await.processed_objects = 1, + "checkpoint" | "blocked" => { + let suffix = if rejection == "checkpoint" { + RESUME_CHECKPOINT_FILE + } else { + RESUME_CHECKPOINT_BLOCKED_FILE + }; + manager + .disk + .write_all(RUSTFS_META_BUCKET, &format!("{BUCKET_META_PREFIX}/{source}_{suffix}"), Vec::new().into()) + .await + .expect("persist checkpoint evidence, including an empty file"); + } + "format" => storage.execution.test_disks()[0] + .write_all(RUSTFS_META_BUCKET, "format.json", b"existing format".to_vec().into()) + .await + .expect("persist actual format evidence"), + "completion_proof" => { + let proof = super::super::replacement::ReplacementCompletionProof::from_state(&manager.get_state().await, 1) + .expect("construct a prior completion proof"); + manager + .disk + .write_all( + RUSTFS_META_BUCKET, + path_to_str(&replacement_completion_proof_path(&source)).expect("proof path"), + serde_json::to_vec(&proof).expect("serialize completion proof").into(), + ) + .await + .expect("persist existing completion proof"); + } + "topology" => storage.historical_topology = false, + "new_generation" => manager.state.write().await.replacement_preformat_handoff_retry_done = true, + "single_retry" => manager.state.write().await.max_retries = 1, + _ => unreachable!("known test rejection"), + } + manager.save_state_strict().await.expect("persist rejection fixture"); + let before = ResumeManager::read_state_file(&manager.disk, &source, manager.state_file) + .await + .expect("source bytes before admission"); + assert!( + !manager + .rearm_preformat_handoff_retry_if_needed(&storage) + .await + .expect("reject unsafe recovery"), + "{rejection}" + ); + assert_eq!( + ResumeManager::read_state_file(&manager.disk, &source, manager.state_file) + .await + .expect("source bytes after rejection"), + before, + "{rejection}" + ); + } +} + +#[tokio::test] +async fn preformat_recovery_rejects_mount_changes_before_or_during_probe() { + for change_identity_on_call in [0, 1] { + let (_dirs, manager, mut storage) = preformat_recovery_fixture().await; + storage.change_identity_on_call = change_identity_on_call; + let source = manager.get_state().await.task_id; + let before = ResumeManager::read_state_file(&manager.disk, &source, manager.state_file) + .await + .expect("source bytes before mount change"); + assert!(matches!( + manager.rearm_preformat_handoff_retry_if_needed(&storage).await, + Err(Error::ReplacementTargetNotReady(_)) + )); + assert_eq!( + ResumeManager::read_state_file(&manager.disk, &source, manager.state_file) + .await + .expect("source bytes after mount change"), + before + ); + } +} + +#[tokio::test] +async fn preformat_recovery_rejects_unknown_marker_and_stale_revision() { + let (_dirs, manager, storage) = preformat_recovery_fixture().await; + let source = manager.get_state().await.task_id; + storage.execution.test_disks()[0] + .write_all(RUSTFS_META_BUCKET, HEALING_MARKER_PATH, b"unknown".to_vec().into()) + .await + .expect("publish unknown marker owner"); + let before = ResumeManager::read_state_file(&manager.disk, &source, manager.state_file) + .await + .expect("source before owner conflict"); + assert!(matches!( + manager.rearm_preformat_handoff_retry_if_needed(&storage).await, + Err(Error::ReplacementOwnershipConflict(_)) + )); + assert_eq!( + ResumeManager::read_state_file(&manager.disk, &source, manager.state_file) + .await + .expect("source after owner conflict"), + before + ); + storage.execution.test_disks()[0] + .delete(RUSTFS_META_BUCKET, HEALING_MARKER_PATH, Default::default()) + .await + .expect("remove fixture owner conflict"); + let stale = ResumeManager::load_replacement_intent(manager.disk.clone(), &source) + .await + .expect("open stale source"); + manager.save_state_strict().await.expect("advance source revision"); + assert!(stale.rearm_preformat_handoff_retry_if_needed(&storage).await.is_err()); + let unchanged = ResumeManager::load_replacement_intent(manager.disk.clone(), &source) + .await + .expect("reload authoritative source") + .get_state() + .await; + assert!(!unchanged.replacement_preformat_handoff_retry_done); + assert!(unchanged.replacement_handoff.is_none()); +} + +#[test] +fn preformat_recovery_marker_defaults_only_old_snapshots_to_eligible() { + let state = ResumeState::replacement_intent( + Uuid::new_v4().to_string(), + "erasure_set".into(), + "pool_0_set_0".into(), + Vec::new(), + vec!["replacement-0".into()], + vec![identity("replacement-0", "mount:fs:root")], + ); + assert!(state.replacement_preformat_handoff_retry_done); + let mut wire = serde_json::to_value(state).expect("serialize new intent"); + wire.as_object_mut() + .expect("intent object") + .remove("replacement_preformat_handoff_retry_done"); + let old: ResumeState = serde_json::from_value(wire).expect("deserialize older schema-seven intent"); + assert!(!old.replacement_preformat_handoff_retry_done); +} + #[tokio::test] async fn handoff_rebinding_never_resets_an_exhausted_successor_budget() { let (_dirs, parent, execution) = fixture(1).await; diff --git a/crates/heal/src/heal/storage.rs b/crates/heal/src/heal/storage.rs index d67519ff2..fc3e8f3db 100644 --- a/crates/heal/src/heal/storage.rs +++ b/crates/heal/src/heal/storage.rs @@ -736,6 +736,13 @@ pub trait HealStorageAPI: Send + Sync { Ok(true) } + /// Admit a one-shot recovery of an old pre-format intent only when the + /// current format fence permits its set, rebalance has a quiescent older + /// topology, and every target is a blank, unformatted replacement. + async fn replacement_preformat_retry_is_safe(&self, _set_disk_id: &str, _targets: &[String]) -> Result { + Ok(false) + } + async fn replacement_execution(&self, _targets: &[String]) -> Result> { Err(Error::other("replacement execution lease acquisition is unsupported")) } @@ -2109,6 +2116,14 @@ impl HealStorageAPI for ECStoreHealStorage { async fn replacement_execution(&self, targets: &[String]) -> Result> { ReplacementExecution::acquire(targets).await } + + async fn replacement_preformat_retry_is_safe(&self, set_disk_id: &str, targets: &[String]) -> Result { + let (pool_index, set_index) = crate::heal::utils::parse_set_disk_id(set_disk_id)?; + self.ecstore + .replacement_preformat_retry_is_safe(pool_index, set_index, targets) + .await + .map_err(Error::Storage) + } } #[cfg(test)] diff --git a/crates/heal/tests/heal_integration_test.rs b/crates/heal/tests/heal_integration_test.rs index 8e163838c..b9b93460c 100644 --- a/crates/heal/tests/heal_integration_test.rs +++ b/crates/heal/tests/heal_integration_test.rs @@ -702,6 +702,180 @@ mod serial_tests { } } +#[test] +#[serial] +fn historical_rebalance_topology_allows_format_and_object_rebuild() { + const STACK_SIZE: usize = 8 * 1024 * 1024; + std::thread::Builder::new() + .name("historical-rebalance-rebuild".to_owned()) + .stack_size(STACK_SIZE) + .spawn(|| { + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(4) + .thread_stack_size(STACK_SIZE) + .enable_all() + .build() + .expect("historical rebalance recovery runtime"); + runtime.block_on(historical_rebalance_topology_allows_format_and_object_rebuild_case()); + }) + .expect("historical rebalance recovery test thread") + .join() + .expect("historical rebalance recovery test result"); +} + +async fn historical_rebalance_topology_allows_format_and_object_rebuild_case() { + use storage_api::integration::{ + DiskAPI as _, DiskError, DiskOption, RUSTFS_META_BUCKET, RebalStatus, RebalanceMeta, RebalanceStats, new_disk, + }; + + let temp = tempfile::tempdir().expect("isolated historical rebalance fixture"); + let env = rustfs_test_utils::TestECStoreEnv::builder() + .base_dir(temp.path()) + .build() + .await; + let ecstore = env.ecstore; + let disks = env.disk_paths; + let storage = Arc::new(ECStoreHealStorage::new(ecstore.clone())); + assert_eq!(ecstore.pools.len(), 1); + let bucket = "historical-rebalance-rebuild"; + let object = "non-inline-object"; + let expected = non_inline_test_data(); + create_test_bucket(&ecstore, bucket).await; + upload_test_object(&ecstore, bucket, object, &expected).await; + wait_for_object_copies(&disks, bucket, object).await; + + // Model the persisted post-removal state. This test does not execute a + // rebalance worker, decommission lifecycle, or mount-identity handoff. + let mut historical = RebalanceMeta { + id: uuid::Uuid::new_v4().to_string(), + pool_stats: vec![RebalanceStats::default(), RebalanceStats::default()], + ..Default::default() + }; + for pool in &mut historical.pool_stats { + pool.participating = true; + pool.info.status = RebalStatus::Completed; + } + historical + .save_with_opts( + ecstore.pools[0].clone(), + ObjectOptions { + write_completion: WriteCompletion::TailDrained, + ..Default::default() + }, + ) + .await + .expect("persist completed historical two-pool metadata on the current metadata pool"); + let historical_bytes = ecstore.pools[0] + .get_object_reader(RUSTFS_META_BUCKET, "rebalance.bin", None, HeaderMap::new(), &ObjectOptions::default()) + .await + .expect("open persisted historical metadata") + .read_all() + .await + .expect("read persisted historical metadata"); + let old_format: serde_json::Value = serde_json::from_slice( + &fs::read(disks[0].join(RUSTFS_META_BUCKET).join("format.json")) + .await + .expect("read target format"), + ) + .expect("decode target format"); + let target_object = disks[0].join(bucket).join(object); + let old_shard = fs::read(find_part_file(&target_object).expect("original target shard")) + .await + .expect("read original target shard bytes"); + + // Preserve the root inode held by disk I/O, while replacing its contents + // with an empty target, as in the existing fresh-target recovery fixture. + for entry in std::fs::read_dir(&disks[0]).expect("read replacement target root") { + let entry = entry.expect("replacement target entry"); + if entry.file_type().expect("replacement entry type").is_dir() { + fs::remove_dir_all(entry.path()).await.expect("empty target directory"); + } else { + fs::remove_file(entry.path()).await.expect("empty target file"); + } + } + assert!(!disks[0].join(RUSTFS_META_BUCKET).join("format.json").exists()); + assert!(!target_object.exists()); + // A replacement reconnects with a fresh handle. Reusing the old handle + // would retain its cached format bytes despite the emptied directory. + let endpoint = ecstore.pools[0].endpoints.endpoints.as_ref()[0].clone(); + let replacement = new_disk( + &endpoint, + &DiskOption { + cleanup: false, + health_check: false, + }, + ) + .await + .expect("open a fresh replacement disk handle"); + assert!(matches!(replacement.get_disk_id().await, Err(DiskError::UnformattedDisk))); + ecstore.pools[0].get_disks(0).disks.write().await[0] = Some(replacement); + let target = endpoint.to_string(); + let (_, format_error) = storage + .heal_replacement_format(false, 0, 0, std::slice::from_ref(&target)) + .await + .expect("historical topology must admit target-scoped format repair"); + assert!(format_error.is_none(), "target format repair: {format_error:?}"); + let repaired_format: serde_json::Value = serde_json::from_slice( + &fs::read(disks[0].join(RUSTFS_META_BUCKET).join("format.json")) + .await + .expect("read repaired target format"), + ) + .expect("decode repaired target format"); + assert_eq!( + repaired_format["xl"]["this"], old_format["xl"]["this"], + "format repair must retain the original slot UUID" + ); + assert!( + !target_object.exists(), + "format repair alone must not be mistaken for object reconstruction" + ); + + let mut request = HealRequest::new( + HealType::ErasureSet { + buckets: vec![bucket.to_owned()], + set_disk_id: "pool_0_set_0".to_owned(), + }, + HealOptions { + recursive: true, + pool_index: Some(0), + set_index: Some(0), + scan_mode: HealScanMode::Deep, + ..Default::default() + }, + HealPriority::Normal, + ); + request.source = HealRequestSource::Admin; + let task = HealTask::from_request(request, storage); + task.execute() + .await + .expect("rebuild objects after historical-topology format repair"); + let outcome = task.get_outcome().await; + assert_eq!(outcome.execution, HealExecutionOutcome::Completed); + assert_eq!(outcome.coverage, HealTraversalCoverage::Complete); + // Verify target-side physical repair before a GET could perform read repair. + assert!(target_object.join("xl.meta").exists()); + let rebuilt_shard = fs::read(find_part_file(&target_object).expect("rebuilt target shard")) + .await + .expect("read rebuilt target shard bytes"); + assert_eq!(rebuilt_shard, old_shard, "the replacement shard must contain the original encoded bytes"); + let actual = ecstore + .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default()) + .await + .expect("open rebuilt object") + .read_all() + .await + .expect("read rebuilt object bytes"); + assert_eq!(actual, expected, "reconstructed object bytes must be unchanged"); + let retained_history = ecstore.pools[0] + .get_object_reader(RUSTFS_META_BUCKET, "rebalance.bin", None, HeaderMap::new(), &ObjectOptions::default()) + .await + .expect("open retained historical metadata") + .read_all() + .await + .expect("read retained historical metadata"); + assert_eq!(retained_history, historical_bytes, "healing must retain the historical rebalance payload"); +} + #[test] #[serial] fn replacement_delete_recovery_on_fresh_target() { diff --git a/crates/heal/tests/storage_api.rs b/crates/heal/tests/storage_api.rs index dd615cba5..171ad216b 100644 --- a/crates/heal/tests/storage_api.rs +++ b/crates/heal/tests/storage_api.rs @@ -28,6 +28,7 @@ pub(crate) mod integration { pub(crate) use rustfs_ecstore::api::object::{ ObjectOptions, PutObjReader, ShardIntegrityWriteMode, WriteCommitGuard, WriteCompletion, }; + pub(crate) use rustfs_ecstore::api::rebalance::{RebalStatus, RebalanceMeta, RebalanceStats}; pub(crate) use rustfs_ecstore::api::storage::ECStore; pub(crate) use rustfs_storage_api::BucketOperations; pub(crate) use rustfs_storage_api::MakeBucketOptions; diff --git a/docs/architecture/compat-cleanup-register.md b/docs/architecture/compat-cleanup-register.md index 423ca0bd5..b97bf86e4 100644 --- a/docs/architecture/compat-cleanup-register.md +++ b/docs/architecture/compat-cleanup-register.md @@ -11,6 +11,8 @@ ## Open Items +- `8565` historical rebalance replacement recovery: older schema-seven intents can exhaust their retry budget before format repair when retained rebalance metadata describes removed pools. After validating blank targets, terminal historical topology, and ownership, grant one durable handoff and one rebuild attempt. Remove after supported direct upgrades no longer admit snapshots lacking replacement_preformat_handoff_retry_done and all retained affected intents have been resolved. + - `backlog-2253-metadata-observation-adapter` metadata observations: preserve aligned metadata/error slices for existing layout, shard and exact-version PUT, MPU and delete consumers while read/heal scheduling uses typed disk observations. Pending slots acquire empty placeholders only in this adapter. Remove after those remaining consumers use typed observations; delete the adapter and old facade in a separate cleanup. - `connect-894` pending Connect heartbeats: replay the exact preceding producer-capability list, with or without jobs, when upgrading to the memory-service capability. Preserve request ID, sequence, and persisted body rather than inserting the new capability into a retry. Remove after upgrades from the pre-service-memory capability set are unsupported.