diff --git a/crates/ecstore/src/cluster/rpc/peer_s3_client.rs b/crates/ecstore/src/cluster/rpc/peer_s3_client.rs index 29f5ac986..3c3321384 100644 --- a/crates/ecstore/src/cluster/rpc/peer_s3_client.rs +++ b/crates/ecstore/src/cluster/rpc/peer_s3_client.rs @@ -417,8 +417,10 @@ impl S3PeerSys { let errs = join_all(futures).await; let mut pool_errs = Vec::new(); + let mut all_pools_have_participants = true; for pool_idx in 0..self.pools_count { let per_pool_errs = pool_participant_errors(&self.clients, &errs, pool_idx); + all_pools_have_participants &= !per_pool_errs.is_empty(); pool_errs.push(reduce_pool_write_quorum_errs(&per_pool_errs)); } @@ -451,6 +453,17 @@ impl S3PeerSys { for pool_idx in 0..self.pools_count { let per_pool_errs = pool_participant_errors(&self.clients, &errs, pool_idx); if let Some(pool_err) = reduce_pool_write_quorum_errs(&per_pool_errs) { + // A success/busy split can miss quorum without a majority error. + // Keep that refusal retryable without masking other peer failures. + let pool_err = if pool_err == Error::ErasureWriteQuorum + && all_pools_have_participants + && per_pool_errs.iter().any(|err| err.as_ref() == Some(&Error::TooManyOpenFiles)) + && errs.iter().flatten().all(|err| *err == Error::TooManyOpenFiles) + { + Error::TooManyOpenFiles + } else { + pool_err + }; tracing::error!("heal_bucket per_pool_errs: {per_pool_errs:?}"); tracing::error!("heal_bucket reduce_write_quorum_errs: {pool_err}"); return Err(pool_err); @@ -1810,6 +1823,7 @@ mod tests { #[derive(Debug)] struct TestPeerS3Client { pools: Option>, + heal_bucket_result: Result, make_bucket_result: Result<()>, list_bucket_result: Result>, delete_bucket_result: Result<()>, @@ -1819,7 +1833,7 @@ mod tests { #[async_trait] impl PeerS3Client for TestPeerS3Client { async fn heal_bucket(&self, _bucket: &str, _opts: &HealOpts) -> Result { - unreachable!("not used by quorum tests") + self.heal_bucket_result.clone() } async fn make_bucket(&self, _bucket: &str, _opts: &MakeBucketOptions) -> Result<()> { @@ -1835,8 +1849,11 @@ mod tests { self.delete_bucket_result.clone() } - async fn get_bucket_info(&self, _bucket: &str, _opts: &BucketOptions) -> Result { - unreachable!("not used by quorum tests") + async fn get_bucket_info(&self, bucket: &str, _opts: &BucketOptions) -> Result { + Ok(BucketInfo { + name: bucket.to_string(), + ..Default::default() + }) } fn get_pools(&self) -> Option> { @@ -1848,6 +1865,17 @@ mod tests { test_peer_with_make_bucket(pools, Ok(())) } + fn test_peer_with_heal_bucket(pools: &[usize], heal_bucket_result: Result) -> Client { + Arc::new(Box::new(TestPeerS3Client { + pools: Some(pools.to_vec()), + heal_bucket_result, + make_bucket_result: Ok(()), + list_bucket_result: Ok(Vec::new()), + delete_bucket_result: Ok(()), + make_bucket_calls: Arc::new(AtomicUsize::new(0)), + })) + } + fn test_peer_with_make_bucket(pools: &[usize], make_bucket_result: Result<()>) -> Client { test_peer_with_results(pools, make_bucket_result, Ok(Vec::new())) } @@ -1885,6 +1913,7 @@ mod tests { ) -> Client { Arc::new(Box::new(TestPeerS3Client { pools: Some(pools.to_vec()), + heal_bucket_result: Ok(HealResultItem::default()), make_bucket_result, list_bucket_result, delete_bucket_result, @@ -2634,6 +2663,119 @@ mod tests { assert_eq!(err, Error::VolumeExists); } + #[tokio::test] + async fn test_heal_bucket_preserves_busy_refusal_below_error_quorum() { + for busy_count in 2..=4 { + let peer_sys = S3PeerSys { + clients: (0..4) + .map(|peer| { + test_peer_with_heal_bucket( + &[0], + if peer < busy_count { + Err(Error::TooManyOpenFiles) + } else { + Ok(HealResultItem::default()) + }, + ) + }) + .collect(), + pools_count: 1, + }; + + let error = peer_sys + .heal_bucket("bucket", &HealOpts::default()) + .await + .expect_err("busy peers must not turn a failed heal into success"); + assert_eq!(error, Error::TooManyOpenFiles, "busy peers: {busy_count}"); + } + } + + #[tokio::test] + async fn test_heal_bucket_busy_peer_preserves_success_quorum() { + let peer_sys = S3PeerSys { + clients: vec![ + test_peer_with_heal_bucket(&[0], Err(Error::TooManyOpenFiles)), + test_peer(&[0]), + test_peer(&[0]), + test_peer(&[0]), + ], + pools_count: 1, + }; + + peer_sys + .heal_bucket("bucket", &HealOpts::default()) + .await + .expect("three successful peers still satisfy the four-peer write quorum"); + } + + #[tokio::test] + async fn test_heal_bucket_busy_peers_do_not_hide_other_failures() { + for failure in [Error::FaultyDisk, Error::VolumeNotFound, Error::ErasureWriteQuorum] { + let peer_sys = S3PeerSys { + clients: vec![ + test_peer_with_heal_bucket(&[0], Err(Error::TooManyOpenFiles)), + test_peer_with_heal_bucket(&[0], Err(Error::TooManyOpenFiles)), + test_peer_with_heal_bucket(&[0], Err(failure.clone())), + test_peer(&[0]), + ], + pools_count: 1, + }; + + let error = peer_sys + .heal_bucket("bucket", &HealOpts::default()) + .await + .expect_err("a mixed failure must still fail write quorum"); + assert_eq!(error, Error::ErasureWriteQuorum, "non-busy failure: {failure}"); + } + } + + #[tokio::test] + async fn test_heal_bucket_busy_pool_does_not_hide_another_pool_failure() { + let peer_sys = S3PeerSys { + clients: vec![ + test_peer_with_heal_bucket(&[0], Err(Error::TooManyOpenFiles)), + test_peer_with_heal_bucket(&[0], Err(Error::TooManyOpenFiles)), + test_peer(&[0]), + test_peer(&[0]), + test_peer_with_heal_bucket(&[1], Err(Error::FaultyDisk)), + ], + pools_count: 2, + }; + + let error = peer_sys + .heal_bucket("bucket", &HealOpts::default()) + .await + .expect_err("the busy split must not hide a non-busy failure from another pool"); + assert_eq!(error, Error::ErasureWriteQuorum); + } + + #[tokio::test] + async fn test_heal_bucket_missing_pool_participants_remains_quorum_failure() { + for occupied_pool in 0..2 { + let peer_sys = S3PeerSys { + clients: vec![ + test_peer_with_heal_bucket(&[occupied_pool], Err(Error::TooManyOpenFiles)), + test_peer_with_heal_bucket(&[occupied_pool], Err(Error::TooManyOpenFiles)), + test_peer(&[occupied_pool]), + test_peer(&[occupied_pool]), + ], + pools_count: 2, + }; + + let error = peer_sys + .heal_bucket( + "bucket", + &HealOpts { + recreate: true, + ..Default::default() + }, + ) + .await + .expect_err("a pool without participants cannot be treated as a busy gate"); + assert_eq!(error, Error::ErasureWriteQuorum, "pool with participants: {occupied_pool}"); + } + } + #[test] fn heal_bucket_mode_fails_closed_on_incomplete_topology() { let mut opts = HealOpts::default(); diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index 6739f83ca..cb2e5be0f 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -33554,12 +33554,15 @@ mod pools_tests { #[test] fn test_resolve_decommission_preflight_heal_result_wraps_error_context() { - let err = resolve_decommission_preflight_heal_result::<()>("bucket-a", Err(Error::FaultyDisk)) - .expect_err("heal failure should carry preflight context"); - assert!( - err.to_string() - .contains("decommission preflight heal failed for bucket bucket-a") - ); + for failure in [Error::FaultyDisk, Error::ErasureWriteQuorum] { + let err = resolve_decommission_preflight_heal_result::<()>("bucket-a", Err(failure)) + .expect_err("heal failure should carry preflight context"); + assert!( + err.to_string() + .contains("decommission preflight heal failed for bucket bucket-a") + ); + assert_ne!(err, Error::SlowDown, "non-busy failures must not become a retryable start"); + } } #[test] diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index a69f55cfd..abcc3c65f 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -10501,6 +10501,20 @@ mod tests { ) .await .expect("seed data movement source"); + store.pools[0] + .put_object( + &bucket, + object, + &mut PutObjReader::from_vec(b"newer source body".to_vec()), + &ObjectOptions { + versioned: true, + version_id: Some(uuid::Uuid::new_v4().to_string()), + mod_time: Some(mod_time + time::Duration::SECOND), + ..Default::default() + }, + ) + .await + .expect("retain an unmoved newer source version"); *store.rebalance_meta.write().await = Some(active_rebalance_meta_for_pool(store.pools.len(), 0)); let source_reader = store.pools[0] @@ -10519,6 +10533,8 @@ mod tests { .await .expect("read data movement source"); let source_data_dir = source_reader.object_info.data_dir; + assert!(!source_reader.object_info.is_latest, "the copied version is historical on the source"); + assert_eq!(source_reader.object_info.num_versions, 2); crate::data_movement::migrate_object(store.clone(), 0, bucket.clone(), source_reader, None, "test_read_window") .await .expect("commit data movement target"); @@ -10535,6 +10551,8 @@ mod tests { ) .await .expect("read committed data movement target"); + assert!(target.is_latest, "the target has only the copied historical version"); + assert_eq!(target.num_versions, 1); assert_ne!(source_data_dir, target.data_dir, "the target must use its own data directory"); assert_eq!( rustfs_utils::http::get_consistent_str(&target.user_defined, rustfs_utils::http::SUFFIX_DATA_MOVED), @@ -10555,6 +10573,27 @@ mod tests { .expect("the equal-time migration window must remain readable"); assert_eq!(resolved.data_dir, target.data_dir); + let mut migrated = store + .get_object_reader( + &bucket, + object, + None, + HeaderMap::new(), + &ObjectOptions { + version_id: Some(version.to_string()), + ..Default::default() + }, + ) + .await + .expect("a partial target history must not hide the readable historical version"); + let mut body = Vec::new(); + migrated + .stream + .read_to_end(&mut body) + .await + .expect("read migrated historical body"); + assert_eq!(body, b"source body"); + shutdown.cancel(); } diff --git a/crates/ecstore/src/store/rebalance/support.rs b/crates/ecstore/src/store/rebalance/support.rs index 13be3844c..be2220561 100644 --- a/crates/ecstore/src/store/rebalance/support.rs +++ b/crates/ecstore/src/store/rebalance/support.rs @@ -186,21 +186,18 @@ fn same_user_defined_identity(left: &ObjectInfo, right: &ObjectInfo) -> bool { } } -/// Pool-specific erasure geometry and data-movement rewrites are intentionally -/// excluded. The selected winner still carries the chosen pool's layout, while -/// the remaining read-visible fields must agree before the pool index can -/// provide a deterministic tie-break. +/// Pool-specific erasure geometry, version history, and data-movement rewrites +/// are intentionally excluded. A copied version can be the target's latest +/// while the source still holds newer versions. The selected winner keeps its +/// pool's layout; the version's read-visible identity must still agree. pub(super) fn same_latest_object_info_identity(left: &ObjectInfo, right: &ObjectInfo) -> bool { let same_read_surface = left.bucket == right.bucket && left.name == right.name && left.is_dir == right.is_dir && left.restore_ongoing == right.restore_ongoing && left.restore_expires == right.restore_expires - && left.is_latest == right.is_latest && left.content_type == right.content_type && left.content_encoding == right.content_encoding - && left.num_versions == right.num_versions - && left.successor_mod_time == right.successor_mod_time && left.inlined == right.inlined && left.metadata_only == right.metadata_only && left.version_only == right.version_only @@ -250,7 +247,13 @@ pub(super) fn same_latest_object_info_identity(left: &ObjectInfo, right: &Object } pub(super) fn validate_prepared_pool_refetch_identity(expected: &ObjectInfo, refetched: &ObjectInfo) -> Result<()> { - if same_latest_object_info_identity(expected, refetched) { + // A refetch stays in the same pool, so its history must still match the + // snapshot even though cross-pool copies may have different inventories. + if expected.is_latest == refetched.is_latest + && expected.num_versions == refetched.num_versions + && expected.successor_mod_time == refetched.successor_mod_time + && same_latest_object_info_identity(expected, refetched) + { Ok(()) } else { Err(Error::ErasureReadQuorum) @@ -348,6 +351,34 @@ mod tests { assert_eq!(error, Error::ErasureReadQuorum); } + #[test] + fn prepared_pool_refetch_identity_fails_closed_on_version_history_change() { + let expected = ObjectInfo { + mod_time: Some(OffsetDateTime::from_unix_timestamp(10).expect("test timestamp should be valid")), + version_id: Some(uuid::Uuid::from_u128(1)), + etag: Some("etag-a".to_string()), + ..Default::default() + }; + for refetched in [ + ObjectInfo { + is_latest: !expected.is_latest, + ..expected.clone() + }, + ObjectInfo { + num_versions: expected.num_versions + 1, + ..expected.clone() + }, + ObjectInfo { + successor_mod_time: Some(OffsetDateTime::from_unix_timestamp(20).expect("successor timestamp should be valid")), + ..expected.clone() + }, + ] { + let error = validate_prepared_pool_refetch_identity(&expected, &refetched) + .expect_err("a changed same-pool history must invalidate the prepared snapshot"); + assert_eq!(error, Error::ErasureReadQuorum); + } + } + #[test] fn rebalance_delete_aggregation_preserves_typed_failures() { for expected in [ diff --git a/docs/testing/scanner-heal-topology-gates.md b/docs/testing/scanner-heal-topology-gates.md index 39f80f795..073253be2 100644 --- a/docs/testing/scanner-heal-topology-gates.md +++ b/docs/testing/scanner-heal-topology-gates.md @@ -58,6 +58,7 @@ The barrier directory is set for every node of a run; only the process whose ope |---|---|---| | movement active | 30 s | matches the existing decommission and rebalance cases | | movement complete | 240 s | the existing cases allow 180 s for 96 objects of 128-256 KiB; this inventory adds 1 MiB externals, multipart and version history (about 45 MiB) | +| resumed decommission complete | 360 s | recovery includes the fleet-proof retry before migration resumes; a healthy CI restart completed after 253 s while still making progress, below the distributed lane's 720 s process limit | | movement barrier reach | 240 s (the movement budget) | a window can sit deep in the object order; on Linux CI a busy target capacity gate (160 inline retries, entries deferred to round end) delayed the first fault object by about 2 minutes. The wait fails at once if the operation completes, fails or is canceled without entering the window. Each status probe is bounded to 5 s and a slow probe counts as "still running", because the admin status request can wait on the held mover until the barrier itself times out. The server releases a held barrier on its own after 120 s | | pre-ACK window reach | 60 s | the decommission start normally reaches its durable save in about a second (POST to response ~0.8 s locally); exceeding 60 s is reported as a stalled start path, not retried with a longer budget | | cancel, stop and terminal transitions | 90 s | stop and cancel are acknowledged before every in-flight entry finishes its cleanup, so terminal state and placement are polled within this budget |