gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-31 09:18:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(tiering): persist opaque remote versions

Browse Source
This commit is contained in:
马登山
2026-07-28 13:37:43 +08:00
parent 1e10d752b9
commit 837286f959
7 changed files with 158 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/object_api/types.rs
+5 -5
View File
@@ -464,11 +464,11 @@ impl ObjectInfo {
let transitioned_object = TransitionedObject {
name: fi.transitioned_objname.clone(),
version_id: if let Some(transition_version_id) = fi.transition_version_id {
transition_version_id.to_string()
} else {
"".to_string()
},
version_id: fi
.transition_version
.clone()
.or_else(|| fi.transition_version_id.map(|version_id| version_id.to_string()))
.unwrap_or_default(),
status: fi.transition_status.clone(),
free_version: fi.tier_free_version(),
tier: fi.transition_tier.clone(),
crates/ecstore/src/services/tier/test_util.rs
+4 -1
View File
@@ -931,7 +931,10 @@ pub async fn read_transition_meta(disk_path: &Path, bucket: &str, object: &str)
status: fi.transition_status.clone(),
tier: fi.transition_tier.clone(),
remote_object: fi.transitioned_objname.clone(),
remote_version_id: fi.transition_version_id.map(|id| id.to_string()),
remote_version_id: fi
.transition_version
.clone()
.or_else(|| fi.transition_version_id.map(|id| id.to_string())),
free_version_count,
})
}
crates/ecstore/src/set_disk/core/io_primitives.rs
+1
View File
@@ -544,6 +544,7 @@ pub(in crate::set_disk) fn metadata_early_stop_candidate_matches(left: &FileInfo
&& left.transitioned_objname == right.transitioned_objname
&& left.transition_tier == right.transition_tier
&& left.transition_version_id == right.transition_version_id
&& left.transition_version == right.transition_version
&& left.expire_restored == right.expire_restored
&& left.size == right.size
&& left.mod_time == right.mod_time
crates/ecstore/src/set_disk/metadata.rs
+1
View File
@@ -578,6 +578,7 @@ impl SetDisks {
Self::update_hash_str(hasher, &meta.transition_tier);
Self::update_hash_str(hasher, &meta.transitioned_objname);
Self::update_hash_optional_uuid(hasher, meta.transition_version_id);
Self::update_hash_optional_str(hasher, meta.transition_version.as_deref());
Self::update_hash_optional_u32(hasher, meta.mode);
Self::update_hash_optional_u64(hasher, meta.written_by_version);
crates/ecstore/src/set_disk/ops/object.rs
+36 -31
View File
@@ -2107,11 +2107,12 @@ async fn pause_transition_commit(bucket: &str, object: &str, pause: TransitionCo
}
}
fn parse_transition_version_id(remote_version: &str) -> std::result::Result<Option<Uuid>, uuid::Error> {
if remote_version.is_empty() {
return Ok(None);
fn parse_transition_version_id(remote_version: &str) -> Option<String> {
if remote_version.is_empty() || Uuid::parse_str(remote_version).is_ok_and(|version_id| version_id.is_nil()) {
None
} else {
Some(remote_version.to_string())
}
Uuid::parse_str(remote_version).map(|version_id| (!version_id.is_nil()).then_some(version_id))
}
#[cfg(test)]
@@ -2278,11 +2279,8 @@ mod transition_version_id_tests {
#[test]
fn normalizes_persisted_unversioned_ids_and_preserves_put_constraints() {
assert_eq!(parse_transition_version_id("").expect("empty remote version should be valid"), None);
assert_eq!(
parse_transition_version_id(&Uuid::nil().to_string()).expect("nil remote version should be valid"),
None
);
assert_eq!(parse_transition_version_id(""), None);
assert_eq!(parse_transition_version_id(&Uuid::nil().to_string()), None);
let nil_put_response = Uuid::nil().to_string();
let nil_candidate = TransitionUploadCandidate::from_put_response(nil_put_response.clone());
assert_eq!(nil_candidate.cleanup_version(), nil_put_response);
@@ -2294,11 +2292,12 @@ mod transition_version_id_tests {
}
#[test]
fn preserves_valid_remote_id_and_rejects_invalid_text() {
fn preserves_uuid_and_opaque_remote_ids() {
let version_id = Uuid::new_v4();
assert_eq!(parse_transition_version_id(&version_id.to_string()), Some(version_id.to_string()));
assert_eq!(
parse_transition_version_id(&version_id.to_string()).expect("UUID remote version should be valid"),
Some(version_id)
parse_transition_version_id("opaque-version-token"),
Some("opaque-version-token".to_string())
);
assert_eq!(
TransitionUploadCandidate::from_put_response(version_id.to_string()).cleanup_version(),
@@ -2308,7 +2307,6 @@ mod transition_version_id_tests {
TransitionUploadCandidate::from_put_response("opaque-version-token".to_string()).cleanup_version(),
"opaque-version-token"
);
assert!(parse_transition_version_id("not-a-uuid").is_err());
}
}
@@ -3554,16 +3552,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object).await;
return Err(err);
}
let transition_version_id = match parse_transition_version_id(candidate.remote_version()) {
Ok(version_id) => version_id,
Err(err) => {
if upload_cleanup.cleanup().await.is_ok() {
delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object)
.await;
}
return Err(err.into());
}
};
let transition_version_id = parse_transition_version_id(candidate.remote_version());
let mut commit_opts = opts.clone();
commit_opts.no_lock = true;
@@ -3621,7 +3610,10 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
current_fi.transition_status = TRANSITION_COMPLETE.to_string();
current_fi.transitioned_objname = dest_obj;
current_fi.transition_tier = opts.transition.tier.clone();
current_fi.transition_version_id = transition_version_id;
current_fi.transition_version_id = transition_version_id
.as_deref()
.and_then(|version_id| Uuid::parse_str(version_id).ok());
current_fi.transition_version = transition_version_id;
rustfs_utils::http::metadata_compat::insert_str(
&mut current_fi.metadata,
rustfs_utils::http::metadata_compat::SUFFIX_TRANSITION_TIER_DESTINATION_ID,
@@ -6242,7 +6234,7 @@ mod transition_upload_integrity_tests {
#[tokio::test]
#[serial_test::serial]
async fn opaque_remote_version_is_cleaned_before_parse_failure() {
async fn opaque_remote_version_is_persisted_exactly() {
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let bucket = "transition-unknown-version-bucket";
let object = "object.bin";
@@ -6255,12 +6247,25 @@ mod transition_upload_integrity_tests {
set_disks
.transition_object(bucket, object, &transition_options(&original, tier_name))
.await
.expect_err("an unparseable remote version must fail closed");
let removed_versions = backend.remove_versions().await;
assert_eq!(removed_versions.len(), 1);
assert_eq!(removed_versions[0].1, "opaque-version-token");
assert_eq!(backend.object_count().await, 0);
assert_local_source_intact(&set_disks, bucket, object, &payload).await;
.expect("an accepted opaque remote version must commit");
let (fi, _, _) = set_disks
.get_object_fileinfo(
bucket,
object,
&ObjectOptions {
no_lock: true,
metadata_cache_safe: false,
..Default::default()
},
true,
false,
)
.await
.expect("committed opaque transition metadata should be readable");
assert_eq!(fi.transition_version_id, None);
assert_eq!(fi.transition_version.as_deref(), Some("opaque-version-token"));
assert_eq!(backend.remove_count().await, 0);
assert_eq!(backend.object_count().await, 1);
}
#[tokio::test]