gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-23 20:59:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(ecstore): handle free-version migration races

Browse Source
This commit is contained in:
overtrue
2026-08-23 10:52:13 +08:00
parent dfbe165882
commit 3be5efab67
4 changed files with 108 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/core/pools.rs
+94 -25
View File
@@ -1350,6 +1350,7 @@ fn should_cleanup_decommission_source_entry(decommissioned: usize, total_version
}
const DECOMMISSION_FREE_VERSION_MIGRATED_REASON: &str = "tier_free_version_migrated";
const DECOMMISSION_FREE_VERSION_CONSUMED_REASON: &str = "tier_free_version_already_consumed";
const DECOMMISSION_FREE_VERSION_RETAINED_REASON: &str = "tier_free_version_migration_failed";
const DECOMMISSION_FREE_VERSION_SWEEP_REASON: &str = "tier_free_version_unresolved_after_decommission";
const DECOMMISSION_FREE_VERSION_DISPOSITION_REASON: &str = "tier_free_version_disposition_recorded";
@@ -1357,6 +1358,7 @@ const DECOMMISSION_FREE_VERSION_DISPOSITION_REASON: &str = "tier_free_version_di
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
struct DecommissionFreeVersionDisposition {
migrated: usize,
consumed: usize,
retained: usize,
}
@@ -1365,12 +1367,32 @@ impl DecommissionFreeVersionDisposition {
self.migrated += 1;
}
fn record_consumed(&mut self) {
self.consumed += 1;
}
fn record_retained(&mut self) {
self.retained += 1;
}
fn total(self) -> usize {
self.migrated.saturating_add(self.retained)
self.migrated.saturating_add(self.consumed).saturating_add(self.retained)
}
}
enum DecommissionFreeVersionAttempt {
Migrated,
Consumed,
CapacityFailure(Error),
Retry(Error),
}
fn classify_decommission_free_version_attempt(result: Result<()>) -> DecommissionFreeVersionAttempt {
match result {
Ok(()) => DecommissionFreeVersionAttempt::Migrated,
Err(err) if is_decommission_copy_cleanup_safe_error(&err) => DecommissionFreeVersionAttempt::Consumed,
Err(err) if is_decommission_target_capacity_error(&err) => DecommissionFreeVersionAttempt::CapacityFailure(err),
Err(err) => DecommissionFreeVersionAttempt::Retry(err),
}
}
@@ -3867,39 +3889,49 @@ impl ECStore {
let version_id = version.version_id.map(|v| v.to_string());
let mut migration_error = None;
let mut migrated = false;
let mut consumed = false;
let mut capacity_failure = false;
for _ in 0..3 {
match run_decommission_side_effect(&rx, &operation_gate, || async {
self.decommission_tiered_object(
bucket.as_str(),
&version.name,
version,
&decommission_remote_tiered_opts(version, version_id.clone(), idx, expected_bucket_incarnation_id),
)
.await
})
.await
{
Ok(()) => {
match classify_decommission_free_version_attempt(
run_decommission_side_effect(&rx, &operation_gate, || async {
self.decommission_tiered_object(
bucket.as_str(),
&version.name,
version,
&decommission_remote_tiered_opts(
version,
version_id.clone(),
idx,
expected_bucket_incarnation_id,
),
)
.await
})
.await,
) {
DecommissionFreeVersionAttempt::Migrated => {
migrated = true;
migration_error = None;
break;
}
Err(err) if is_decommission_target_capacity_error(&err) => {
return Err(with_decommission_entry_context(
"decommission_tier_free_version",
bucket.as_str(),
version.name.as_str(),
err,
));
DecommissionFreeVersionAttempt::Consumed => {
consumed = true;
migration_error = None;
break;
}
Err(err) => migration_error = Some(err),
DecommissionFreeVersionAttempt::CapacityFailure(err) => {
capacity_failure = true;
migration_error = Some(err);
break;
}
DecommissionFreeVersionAttempt::Retry(err) => migration_error = Some(err),
}
}
{
let mut pool_meta = self.pool_meta.write().await;
ensure_decommission_generation(&pool_meta, idx, generation)?;
if let Err(err) = count_decommission_item(&mut pool_meta, idx, 0, !migrated) {
if let Err(err) = count_decommission_item(&mut pool_meta, idx, 0, !migrated && !consumed) {
return Err(with_decommission_entry_context(
"count_decommission_item",
bucket.as_str(),
@@ -3909,10 +3941,14 @@ impl ECStore {
}
}
if migrated {
if migrated || consumed {
decommissioned += 1;
cleanup_preflight_allowed_missing.push(data_movement::source_cleanup_version_identity(version));
}
if migrated {
free_version_disposition.record_migrated();
} else if consumed {
free_version_disposition.record_consumed();
} else {
free_version_disposition.record_retained();
}
@@ -3928,14 +3964,31 @@ impl ECStore {
result = ?migration_error,
reason = if migrated {
DECOMMISSION_FREE_VERSION_MIGRATED_REASON
} else if consumed {
DECOMMISSION_FREE_VERSION_CONSUMED_REASON
} else {
DECOMMISSION_FREE_VERSION_RETAINED_REASON
},
state = if migrated { "free_version_migrated" } else { "free_version_retained" },
state = if migrated {
"free_version_migrated"
} else if consumed {
"free_version_consumed"
} else {
"free_version_retained"
},
"Decommission free-version disposition recorded"
);
if !migrated {
if capacity_failure {
return Err(with_decommission_entry_context(
"decommission_tier_free_version",
bucket.as_str(),
version.name.as_str(),
migration_error.expect("capacity failure must retain its error"),
));
}
if !migrated && !consumed {
break;
}
continue;
@@ -4249,6 +4302,7 @@ impl ECStore {
bucket = %bucket,
object = %entry.name,
free_versions_migrated = free_version_disposition.migrated,
free_versions_consumed = free_version_disposition.consumed,
free_versions_retained = free_version_disposition.retained,
free_versions_total = free_version_disposition.total(),
reason = DECOMMISSION_FREE_VERSION_DISPOSITION_REASON,
@@ -5555,6 +5609,21 @@ mod tests {
)));
}
#[test]
fn decommission_free_version_attempt_treats_missing_source_as_consumed() {
let attempt =
classify_decommission_free_version_attempt(Err(Error::ObjectNotFound("bucket".to_string(), "object".to_string())));
assert!(matches!(attempt, DecommissionFreeVersionAttempt::Consumed));
}
#[test]
fn decommission_free_version_attempt_preserves_capacity_failure() {
let attempt = classify_decommission_free_version_attempt(Err(Error::DiskFull));
assert!(matches!(attempt, DecommissionFreeVersionAttempt::CapacityFailure(Error::DiskFull)));
}
#[test]
fn decommission_delete_marker_copy_error_rejects_data_movement_overwrite() {
let err = Error::DataMovementOverwriteErr("bucket".to_string(), "object".to_string(), "version".to_string());
crates/ecstore/src/data_movement/mod.rs
+1 -2
View File
@@ -1956,8 +1956,7 @@ mod tests {
let mut free_version = cleanup_test_file_info("object.txt", Uuid::from_u128(2), "tier-cleanup");
free_version.deleted = true;
free_version.set_tier_free_version();
let mut expected = cleanup_test_versions(vec![migrated.clone()]);
expected.free_versions = vec![free_version.clone()];
let expected = cleanup_test_versions(vec![migrated.clone(), free_version.clone()]);
let current = cleanup_test_versions(vec![migrated]);
let allowed_missing = vec![source_cleanup_version_identity(&free_version)];
crates/ecstore/src/set_disk/mod.rs
+1
View File
@@ -249,6 +249,7 @@ async fn inspect_decommission_tier_free_version_target(
{
matching_count += 1;
let existing = existing.into_fileinfo(bucket, object, true)?;
existing.validate_for_metadata_read()?;
if !existing.tier_free_version() || !crate::store::tiered_data_movement_source_matches(source, &existing)? {
all_matching_versions_equivalent = false;
}