gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-19 11:06:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(rebalance): converge multipart data movement retries

Browse Source
This commit is contained in:
马登山
2026-08-12 14:32:46 +08:00
parent 5d05897ae0
commit 3234c72683
36 changed files with 5045 additions and 681 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/set_disk/core/io_primitives.rs
+39 -5
View File
@@ -2442,13 +2442,14 @@ impl SetDisks {
bucket: &str,
object: &str,
) -> Result<Option<rustfs_filemeta::FileInfoVersions>> {
let disk_object = rustfs_utils::path::encode_dir_object(object);
let disks = self.get_disks_internal().await;
if disks.is_empty() {
return Err(to_object_err(StorageError::ErasureReadQuorum, vec![bucket, object]));
}
let read_quorum = disks.len().div_ceil(2).max(1);
let (raw_fileinfos, errs) = Self::read_all_raw_file_info(&disks, bucket, object, false).await;
let (raw_fileinfos, errs) = Self::read_all_raw_file_info(&disks, bucket, disk_object.as_str(), false).await;
if let Some(err) = reduce_read_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, read_quorum) {
let object_err = to_object_err(err.into(), vec![bucket, object]);
@@ -2604,7 +2605,7 @@ impl SetDisks {
//
// `into_fileinfo` with an empty version_id selects the first non-free version
// (see FileMeta::into_fileinfo); replicate that selection from the header here.
let vid = match meta.into_fileinfo(bucket, object, "", true, incl_free_vers, true) {
let vid = match meta.into_fileinfo_without_part_checksums(bucket, object, "", true, incl_free_vers) {
Ok(finfo) if file_info_is_valid_for_metadata(&finfo) => finfo.version_id.unwrap_or(Uuid::nil()),
_ => match meta
.versions
@@ -2627,7 +2628,13 @@ impl SetDisks {
for (idx, meta_op) in metadata_array.iter().enumerate() {
if let Some(meta) = meta_op {
match meta.into_fileinfo(bucket, object, vid.to_string().as_str(), read_data, incl_free_vers, true) {
match meta.into_fileinfo_without_part_checksums(
bucket,
object,
vid.to_string().as_str(),
read_data,
incl_free_vers,
) {
Ok(res) => match res.validate_for_metadata_read() {
Ok(_) => meta_file_infos[idx] = res,
Err(err) => errs[idx] = Some(err.into()),
@@ -4530,9 +4537,10 @@ impl SetDisks {
match oi {
Ok(oi) => {
// If top level is a delete marker proceed to upload.
// Ordinary writes may proceed past a top-level delete marker;
// data movement must not replace an acknowledged deletion.
if oi.delete_marker {
return None;
return opts.data_movement.then_some(StorageError::PreconditionFailed);
}
let if_none_match = http_preconditions.if_none_match_value().map(str::to_owned);
let if_match = http_preconditions.if_match_value().map(str::to_owned);
@@ -6388,6 +6396,32 @@ mod tests {
assert_eq!(versions.versions[0].name, object);
}
#[tokio::test]
async fn load_file_info_versions_exact_encodes_directory_key_but_returns_logical_name() {
let bucket = "exact-directory-versions-bucket";
let object = "prefix/directory/";
let disk_object = rustfs_utils::path::encode_dir_object(object);
let (_dir, disk) = read_multiple_test_disk(bucket, &[]).await;
let mut fi = metadata_test_fileinfo(object);
fi.version_id = Some(Uuid::new_v4());
fi.mod_time = Some(OffsetDateTime::now_utc());
disk.write_metadata(bucket, bucket, disk_object.as_str(), fi.clone())
.await
.expect("directory metadata should be written under the encoded key");
let set = io_primitives_test_set(vec![Some(disk)], 0).await;
let versions = set
.load_file_info_versions_exact(bucket, object)
.await
.expect("exact directory version load should succeed")
.expect("exact directory version load should find metadata");
assert_eq!(versions.name, object);
assert_eq!(versions.versions.len(), 1);
assert_eq!(versions.versions[0].name, object);
assert_eq!(versions.versions[0].version_id, fi.version_id);
}
#[tokio::test]
async fn load_file_info_versions_exact_rejects_transitioned_duplicate_parts() {
let bucket = "exact-versions-bucket";
crates/ecstore/src/set_disk/metadata.rs
+22
View File
@@ -466,6 +466,28 @@ impl SetDisks {
Self::find_file_info_in_quorum(metas, &mod_time, &etag, quorum)
}
pub(crate) fn hydrate_selected_fileinfo_part_checksums(fi: &mut FileInfo) -> disk::error::Result<()> {
fi.hydrate_data_movement_part_checksums().map_err(DiskError::from)?;
for part in &fi.parts {
let Some(checksums) = part.checksums.as_ref() else {
continue;
};
let mut algorithms = HashSet::with_capacity(checksums.len());
for (name, value) in checksums {
let Some(checksum) = rustfs_rio::Checksum::new_from_string(name, value) else {
return Err(DiskError::FileCorrupt);
};
if checksum.checksum_type.is(rustfs_rio::ChecksumType::MULTIPART) {
return Err(DiskError::FileCorrupt);
}
if !algorithms.insert(checksum.checksum_type.base().0) {
return Err(DiskError::FileCorrupt);
}
}
}
Ok(())
}
fn update_hash_bytes(hasher: &mut Sha256, value: &[u8]) {
hasher.update(value.len().to_le_bytes());
hasher.update(value);
crates/ecstore/src/set_disk/mod.rs
+151 -10
View File
@@ -3738,8 +3738,10 @@ fn resolve_delete_version_state(opts: &ObjectOptions, goi: &ObjectInfo, version_
if opts.version_id.is_some() {
// Decommission/rebalance may recreate a delete marker on a new pool before that
// exact version exists there, so we must still treat it as a mark-delete write.
if opts.data_movement && opts.delete_marker && !version_found {
let data_movement_missing_delete_marker = opts.data_movement && opts.delete_marker && !version_found;
if data_movement_missing_delete_marker {
mark_delete = true;
delete_marker = true;
}
let delete_marker_version_purge = version_found && goi.delete_marker && !opts.version_purge_status().is_empty();
@@ -3748,7 +3750,10 @@ fn resolve_delete_version_state(opts: &ObjectOptions, goi: &ObjectInfo, version_
mark_delete = false;
}
if opts.version_purge_status().is_empty() && opts.delete_marker_replication_status().is_empty() {
if !data_movement_missing_delete_marker
&& opts.version_purge_status().is_empty()
&& opts.delete_marker_replication_status().is_empty()
{
mark_delete = false;
}
@@ -3790,6 +3795,19 @@ impl SetDisks {
opts: &ObjectOptions,
) -> Result<()> {
let storage_class_config = self.storage_class_config_snapshot();
let bucket_lifecycle_guard = if let Some(expected_incarnation_id) = opts.expected_bucket_incarnation_id
&& opts.bucket_lifecycle_lock_fence.is_none()
&& !crate::bucket::utils::is_meta_bucketname(bucket)
{
Some(
metadata_sys::object_store_in(&self.ctx)
.await?
.acquire_bucket_incarnation_fence(bucket, expected_incarnation_id)
.await?,
)
} else {
None
};
let _lock_guard = if !opts.no_lock {
Some(
self.new_ns_lock(bucket, object)
@@ -3802,6 +3820,12 @@ impl SetDisks {
None
};
if opts.http_preconditions.is_some()
&& let Some(err) = self.check_write_precondition(bucket, object, opts).await
{
return Err(err);
}
let disks = self.disks.read().await.clone();
let storage_class = opts.user_defined.get(AMZ_STORAGE_CLASS).map(String::as_str);
let layout = resolve_write_layout(
@@ -3814,6 +3838,20 @@ impl SetDisks {
)?;
let fi = build_tiered_decommission_file_info(bucket, object, fi, layout);
let write_quorum = layout.write_quorum;
if opts
.bucket_lifecycle_lock_fence
.as_ref()
.is_some_and(NamespaceLockFence::is_lock_lost)
|| bucket_lifecycle_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
{
return Err(StorageError::NamespaceLockQuorumUnavailable {
mode: "decommission_tiered_object_commit",
bucket: bucket.to_string(),
object: object.to_string(),
required: 1,
achieved: 0,
});
}
let parts_metadata = vec![fi.clone(); disks.len()];
let (shuffle_disks, parts_metadata) = Self::shuffle_disks_and_parts_metadata(&disks, &parts_metadata, &fi);
@@ -5219,6 +5257,22 @@ mod tests {
assert!(delete_marker);
}
#[test]
fn resolve_delete_version_state_creates_missing_suspended_data_movement_marker() {
let opts = ObjectOptions {
version_suspended: true,
version_id: Some(Uuid::nil().to_string()),
data_movement: true,
delete_marker: true,
..Default::default()
};
let (mark_delete, delete_marker) = resolve_delete_version_state(&opts, &ObjectInfo::default(), false);
assert!(mark_delete);
assert!(delete_marker);
}
#[test]
fn should_force_delete_marker_for_missing_version_rejects_data_movement_latest_delete() {
let opts = ObjectOptions {
@@ -6902,6 +6956,7 @@ mod tests {
rustfs_filemeta::FileInfoOpts {
data: false,
include_free_versions: false,
include_part_checksums: true,
},
)
.expect("test file metadata should decode as file info")
@@ -7030,20 +7085,92 @@ mod tests {
fn test_latest_fileinfo_selection_preserves_degraded_read_quorum_without_competing_latest() {
let mod_time = OffsetDateTime::now_utc();
let data_dir = Uuid::new_v4();
let metas = vec![
quorum_test_fileinfo(mod_time, data_dir, "part-etag-old", 1),
quorum_test_fileinfo(mod_time, data_dir, "part-etag-old", 2),
FileInfo::default(),
FileInfo::default(),
];
let mut first = quorum_test_fileinfo(mod_time, data_dir, "part-etag-old", 1);
rustfs_utils::http::insert_str(
&mut first.metadata,
rustfs_utils::http::SUFFIX_PART_CHECKSUMS,
r#"[[1,[["CRC32C","AAAAAA=="]]]]"#.to_string(),
);
let mut second = first.clone();
second.erasure.index = 2;
let metas = vec![first, second, FileInfo::default(), FileInfo::default()];
let errs = vec![None, None, Some(DiskError::DiskNotFound), Some(DiskError::DiskNotFound)];
let (_, selected, selected_quorum) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, "", 2, 3)
.expect("read quorum should remain enough when no competing latest is visible");
let (_, mut selected, selected_quorum) =
SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, "", 2, 3)
.expect("read quorum should remain enough when no competing latest is visible");
assert_eq!(selected_quorum, 2);
assert_eq!(selected.data_dir, Some(data_dir));
assert_eq!(selected.parts[0].etag, "part-etag-old");
assert!(selected.parts[0].checksums.is_none());
SetDisks::hydrate_selected_fileinfo_part_checksums(&mut selected)
.expect("requested part checksums should hydrate after winner selection");
assert_eq!(
selected.parts[0]
.checksums
.as_ref()
.and_then(|checksums| checksums.get("CRC32C"))
.map(String::as_str),
Some("AAAAAA==")
);
}
#[test]
fn test_degraded_fileinfo_selection_rejects_malformed_part_checksum_metadata() {
let mod_time = OffsetDateTime::now_utc();
let data_dir = Uuid::new_v4();
let mut first = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 1);
rustfs_utils::http::insert_str(&mut first.metadata, rustfs_utils::http::SUFFIX_PART_CHECKSUMS, "not-json".to_string());
let mut second = first.clone();
second.erasure.index = 2;
let metas = vec![first, second, FileInfo::default(), FileInfo::default()];
let errs = vec![None, None, Some(DiskError::DiskNotFound), Some(DiskError::DiskNotFound)];
let (_, mut selected, _) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, "", 2, 3)
.expect("winner selection should defer sidecar decoding");
let err = SetDisks::hydrate_selected_fileinfo_part_checksums(&mut selected)
.expect_err("a malformed degraded winner must fail closed when checksums are requested");
assert_eq!(err, DiskError::FileCorrupt);
}
#[test]
fn test_pick_valid_fileinfo_rejects_malformed_part_checksum_metadata() {
let mod_time = OffsetDateTime::now_utc();
let data_dir = Uuid::new_v4();
let mut meta = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 1);
rustfs_utils::http::insert_str(&mut meta.metadata, rustfs_utils::http::SUFFIX_PART_CHECKSUMS, "not-json".to_string());
let mut second = meta.clone();
second.erasure.index = 2;
let mut selected = SetDisks::pick_valid_fileinfo(&[meta, second], Some(mod_time), None, 2)
.expect("winner selection should defer sidecar decoding");
let err = SetDisks::hydrate_selected_fileinfo_part_checksums(&mut selected)
.expect_err("a malformed winning part-checksum sidecar must fail closed when checksums are requested");
assert_eq!(err, DiskError::FileCorrupt);
}
#[test]
fn test_part_checksum_hydration_rejects_invalid_algorithm_and_value() {
let mod_time = OffsetDateTime::now_utc();
let data_dir = Uuid::new_v4();
for encoded in [
r#"[[1,[["UNKNOWN","AAAAAA=="]]]]"#,
r#"[[1,[["CRC32C","not-base64"]]]]"#,
r#"[[1,[["CRC32C","AA=="]]]]"#,
r#"[[1,[["CRC32C","AAAAAA==-0"]]]]"#,
r#"[[1,[["CRC32C","AAAAAA==-1"]]]]"#,
r#"[[1,[["CRC32C","AAAAAA=="],["crc32c","BBBBBB=="]]]]"#,
] {
let mut meta = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 1);
rustfs_utils::http::insert_str(&mut meta.metadata, rustfs_utils::http::SUFFIX_PART_CHECKSUMS, encoded.to_string());
let err = SetDisks::hydrate_selected_fileinfo_part_checksums(&mut meta)
.expect_err("invalid persisted part checksum metadata must fail closed");
assert_eq!(err, DiskError::FileCorrupt);
}
}
#[test]
@@ -10578,6 +10705,20 @@ mod tests {
assert!(marker.delete_marker);
let marker_version = marker.version_id.expect("versioned delete marker should carry a version id");
let create_only = ObjectOptions {
versioned: true,
data_movement: true,
http_preconditions: Some(HTTPPreconditions {
if_none_match: Some("*".to_string()),
..Default::default()
}),
..Default::default()
};
assert_eq!(
set_disks.check_write_precondition(bucket, object, &create_only).await,
Some(StorageError::PreconditionFailed),
"data movement must not replace a target delete marker"
);
let err = match set_disks
.get_object_reader(bucket, object, None, HeaderMap::new(), &opts)
.await
crates/ecstore/src/set_disk/ops/heal.rs
+2 -1
View File
@@ -542,7 +542,8 @@ impl SetDisks {
let filter_by_etag = quorum_etag.is_some();
match Self::pick_valid_fileinfo(&parts_metadata, quorum_mod_time, quorum_etag.clone(), read_quorum as usize) {
Ok(latest_meta) => {
Ok(mut latest_meta) => {
Self::hydrate_selected_fileinfo_part_checksums(&mut latest_meta)?;
trace!(
event = EVENT_SET_DISK_HEAL,
component = LOG_COMPONENT_ECSTORE,
crates/ecstore/src/set_disk/ops/multipart.rs
+233 -6
View File
@@ -1360,6 +1360,10 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
let disks = disks.clone();
let mut user_defined = opts.user_defined.clone();
rustfs_utils::http::remove_str(&mut user_defined, rustfs_utils::http::SUFFIX_PART_CHECKSUMS);
if !opts.data_movement {
rustfs_utils::http::remove_str(&mut user_defined, rustfs_utils::http::SUFFIX_DATA_MOVEMENT_UPLOAD);
}
rustfs_utils::http::metadata_compat::remove_str(
&mut user_defined,
crate::object_api::ENCRYPTED_PART_LAYOUT_QUORUM_SUFFIX,
@@ -1579,6 +1583,11 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
let expected_restore_operation_id = restore_commit_operation_id_from_metadata(&opts.user_defined)?;
let (mut fi, files_metas) = self.check_upload_id_exists(bucket, object, upload_id, true).await?;
if rustfs_utils::http::contains_key_str(&fi.metadata, rustfs_utils::http::SUFFIX_DATA_MOVEMENT_UPLOAD)
&& !opts.data_movement
{
return Err(StorageError::InvalidUploadID(bucket.to_owned(), object.to_owned(), upload_id.to_owned()));
}
ensure_multipart_bucket_incarnation(&self.ctx, &fi, bucket, object, upload_id, opts.expected_bucket_incarnation_id)
.await?;
let has_layout_candidate = range_seek_rollout_enabled
@@ -1747,7 +1756,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
// Build a lookup map for O(1) part resolution instead of O(n) find() in the loop
// This optimizes from O(n^2) to O(n) when processing many parts
use std::collections::HashMap;
use std::collections::{HashMap, HashSet};
let part_lookup: HashMap<usize, &ObjectPartInfo> = curr_fi.parts.iter().map(|part| (part.number, part)).collect();
for (i, p) in uploaded_parts.iter().enumerate() {
@@ -1783,7 +1792,10 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
// TODO: crypto
if (i < uploaded_parts.len() - 1) && !is_min_allowed_part_size(ext_part.actual_size) {
if (i < uploaded_parts.len() - 1)
&& !(opts.data_movement && ext_part.actual_size < 0)
&& !is_min_allowed_part_size(ext_part.actual_size)
{
error!(
"complete_multipart_upload part size too small: part {} size {} is less than minimum {}",
p.part_num,
@@ -1954,7 +1966,27 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
);
}
if opts.replication_request {
let data_movement_actual_size = if opts.data_movement {
rustfs_utils::http::get_consistent_str(&opts.user_defined, SUFFIX_ACTUAL_SIZE)
.map(|value| {
value
.parse::<i64>()
.ok()
.filter(|value| *value >= 0)
.ok_or_else(|| Error::other("data movement actual size metadata is invalid"))
})
.transpose()?
} else {
None
};
if let Some(actual_size) = data_movement_actual_size {
insert_str(&mut fi.metadata, SUFFIX_ACTUAL_SIZE, actual_size.to_string());
if persist_encryption_original_size {
fi.metadata
.insert("x-rustfs-encryption-original-size".to_string(), actual_size.to_string());
}
} else if opts.replication_request {
if let Some(actual_size) = get_str(&opts.user_defined, SUFFIX_ACTUAL_OBJECT_SIZE_CAP) {
insert_str(&mut fi.metadata, SUFFIX_ACTUAL_SIZE, actual_size.clone());
if persist_encryption_original_size {
@@ -1974,7 +2006,15 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
insert_str(&mut fi.metadata, SUFFIX_COMPRESSION_SIZE, object_size.to_string());
}
rustfs_utils::http::remove_str(&mut fi.metadata, rustfs_utils::http::SUFFIX_PART_CHECKSUMS);
if let Some(part_checksums) =
rustfs_utils::http::get_consistent_str(&opts.user_defined, rustfs_utils::http::SUFFIX_PART_CHECKSUMS)
{
insert_str(&mut fi.metadata, rustfs_utils::http::SUFFIX_PART_CHECKSUMS, part_checksums.to_string());
}
if opts.data_movement {
rustfs_utils::http::remove_str(&mut fi.metadata, rustfs_utils::http::SUFFIX_DATA_MOVEMENT_UPLOAD);
fi.set_data_moved();
}
@@ -1990,19 +2030,21 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
}
}
let completed_part_numbers = fi.parts.iter().map(|part| part.number).collect::<HashSet<_>>();
let upload_data_dir = curr_fi.data_dir.unwrap_or(Uuid::nil()).to_string();
let mut parts = Vec::with_capacity(curr_fi.parts.len());
for p in curr_fi.parts.iter() {
parts.push(path_join_buf(&[
&upload_id_path,
curr_fi.data_dir.unwrap_or(Uuid::nil()).to_string().as_str(),
upload_data_dir.as_str(),
format!("part.{}.meta", p.number).as_str(),
]));
if !fi.parts.iter().any(|v| v.number == p.number) {
if !completed_part_numbers.contains(&p.number) {
parts.push(path_join_buf(&[
&upload_id_path,
curr_fi.data_dir.unwrap_or(Uuid::nil()).to_string().as_str(),
upload_data_dir.as_str(),
format!("part.{}", p.number).as_str(),
]));
}
@@ -2718,6 +2760,191 @@ mod tests {
assert!(matches!(overflow_err, StorageError::PartMissingOrCorrupt));
}
#[tokio::test]
#[serial]
async fn data_movement_upload_rejects_external_completion() {
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let bucket = "data-movement-upload-complete-bucket";
make_bucket_on_all(&disk_stores, bucket).await;
let mut durable_metadata = HashMap::new();
rustfs_utils::http::insert_str(&mut durable_metadata, rustfs_utils::http::SUFFIX_DATA_MOVED, "true".to_string());
rustfs_utils::http::insert_str(
&mut durable_metadata,
rustfs_utils::http::SUFFIX_PART_CHECKSUMS,
r#"[[1,[["CRC32C","AAAAAA=="]]]]"#.to_string(),
);
rustfs_utils::http::insert_str(
&mut durable_metadata,
rustfs_utils::http::SUFFIX_DATA_MOVEMENT_UPLOAD,
"residual-final-marker".to_string(),
);
let ordinary_opts = ObjectOptions {
user_defined: durable_metadata,
..Default::default()
};
let (upload_id, parts) =
stage_upload_with_create_opts(&set_disks, bucket, "ordinary-object", b"ordinary multipart body", &ordinary_opts)
.await;
set_disks
.clone()
.complete_multipart_upload(bucket, "ordinary-object", &upload_id, parts, &ordinary_opts)
.await
.expect("durable object metadata must not claim upload ownership");
let completed = set_disks
.get_object_info(
bucket,
"ordinary-object",
&ObjectOptions {
include_part_checksums: true,
..Default::default()
},
)
.await
.expect("completed staging checksum metadata should remain readable");
assert_eq!(
completed.parts[0]
.checksums
.as_ref()
.and_then(|checksums| checksums.get("CRC32C"))
.map(String::as_str),
Some("AAAAAA==")
);
let object = "owned-object";
let mut metadata = HashMap::new();
rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_DATA_MOVEMENT_UPLOAD, "true".to_string());
rustfs_utils::http::insert_str(
&mut metadata,
rustfs_utils::http::SUFFIX_PART_CHECKSUMS,
r#"[[1,[["CRC32C","AAAAAA=="]]]]"#.to_string(),
);
let create_opts = ObjectOptions {
data_movement: true,
user_defined: metadata,
..Default::default()
};
let (upload_id, parts) =
stage_upload_with_create_opts(&set_disks, bucket, object, b"data movement multipart body", &create_opts).await;
let external_err = set_disks
.clone()
.complete_multipart_upload(bucket, object, &upload_id, parts.clone(), &ObjectOptions::default())
.await
.expect_err("external completion must not finalize a data movement upload");
assert!(matches!(external_err, StorageError::InvalidUploadID(..)));
set_disks
.clone()
.complete_multipart_upload(bucket, object, &upload_id, parts, &create_opts)
.await
.expect("data movement completion should retain ownership of its upload");
let completed = set_disks
.get_object_info(
bucket,
object,
&ObjectOptions {
include_part_checksums: true,
..Default::default()
},
)
.await
.expect("completed data movement object should be readable");
assert!(!rustfs_utils::http::contains_key_str(
&completed.user_defined,
rustfs_utils::http::SUFFIX_DATA_MOVEMENT_UPLOAD
));
for suffix in [
rustfs_utils::http::SUFFIX_DATA_MOVED,
rustfs_utils::http::SUFFIX_PART_CHECKSUMS,
] {
assert!(
completed
.user_defined
.contains_key(&rustfs_utils::http::internal_key_rustfs(suffix))
);
assert!(
completed
.user_defined
.contains_key(&format!("{}{suffix}", rustfs_utils::http::MINIO_INTERNAL_PREFIX))
);
}
assert_eq!(
completed.parts[0]
.checksums
.as_ref()
.and_then(|checksums| checksums.get("CRC32C"))
.map(String::as_str),
Some("AAAAAA==")
);
}
#[tokio::test]
#[serial]
async fn data_movement_complete_accepts_unknown_compressed_part_actual_size() {
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let bucket = "data-movement-unknown-actual-size-bucket";
let object = "object";
make_bucket_on_all(&disk_stores, bucket).await;
let mut metadata = HashMap::new();
rustfs_utils::http::insert_str(
&mut metadata,
rustfs_utils::http::SUFFIX_COMPRESSION,
crate::io_support::rio::compression_metadata_value(rustfs_utils::CompressionAlgorithm::default()),
);
rustfs_utils::http::insert_str(
&mut metadata,
rustfs_utils::http::SUFFIX_DATA_MOVEMENT_UPLOAD,
"source-generation".to_string(),
);
let create_opts = ObjectOptions {
data_movement: true,
user_defined: metadata.clone(),
..Default::default()
};
let upload = set_disks
.new_multipart_upload(bucket, object, &create_opts)
.await
.expect("data movement upload should be created");
let mut completed_parts = Vec::new();
for (number, actual_size) in [(1, -1), (2, 1)] {
let mut reader = PutObjReader::new(
HashReader::from_stream(Cursor::new(vec![number as u8]), 1, actual_size, None, None, false)
.expect("part reader should be constructed"),
);
let part = set_disks
.put_object_part(bucket, object, &upload.upload_id, number, &mut reader, &ObjectOptions::default())
.await
.expect("data movement part should be written");
completed_parts.push(CompletePart {
part_num: number,
etag: part.etag,
..Default::default()
});
}
rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, "2".to_string());
let completed = set_disks
.clone()
.complete_multipart_upload(
bucket,
object,
&upload.upload_id,
completed_parts,
&ObjectOptions {
data_movement: true,
user_defined: metadata,
..Default::default()
},
)
.await
.expect("data movement completion should accept the persisted unknown-size sentinel");
assert_eq!(completed.parts[0].actual_size, -1);
assert_eq!(completed.get_actual_size().expect("completed object actual size"), 2);
}
async fn assert_complete_first_linearizes(bucket: &'static str, object: &'static str, create_opts: ObjectOptions) {
let manager = Arc::new(rustfs_lock::GlobalLockManager::new());
let signaling = Arc::new(SignalingLockClient::new(Arc::new(LocalClient::with_manager(manager))));
crates/ecstore/src/set_disk/ops/object.rs
+452 -16
View File
@@ -3167,7 +3167,11 @@ impl SetDisks {
// Force the full quorum fanout (allow_early_stop=false): `disks` is the
// write target below, and an early-stop subset would only carry read
// quorum, failing write quorum on update_object_meta (backlog#872).
let (mut fi, _, disks) = self.get_object_fileinfo_gated(bucket, object, opts, false, false).await?;
let mut read_opts = opts.clone();
read_opts.include_part_checksums = true;
let (mut fi, _, disks) = self
.get_object_fileinfo_gated(bucket, object, &read_opts, false, false)
.await?;
fi.metadata.insert(AMZ_OBJECT_TAGGING.to_owned(), tags.to_owned());
if let Some(eval_metadata) = &opts.eval_metadata {
@@ -3377,11 +3381,26 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
}
};
fi.metadata = (*src_info.user_defined).clone();
if let Some(etag) = &src_info.etag {
fi.metadata.insert("etag".to_owned(), etag.clone());
let preserved_part_checksums = if (src_info.metadata_only || src_info.version_only)
&& rustfs_utils::http::contains_key_str(&fi.metadata, rustfs_utils::http::SUFFIX_PART_CHECKSUMS)
{
Self::hydrate_selected_fileinfo_part_checksums(&mut fi).map_err(|_| Error::FileCorrupt)?;
Some(
rustfs_utils::http::get_consistent_str(&fi.metadata, rustfs_utils::http::SUFFIX_PART_CHECKSUMS)
.ok_or(Error::FileCorrupt)?
.to_string(),
)
} else {
None
};
let mut replacement_metadata = (*src_info.user_defined).clone();
if let Some(part_checksums) = preserved_part_checksums {
rustfs_utils::http::insert_str(&mut replacement_metadata, rustfs_utils::http::SUFFIX_PART_CHECKSUMS, part_checksums);
}
if let Some(etag) = &src_info.etag {
replacement_metadata.insert("etag".to_owned(), etag.clone());
}
fi.metadata = replacement_metadata.clone();
let mod_time = OffsetDateTime::now_utc();
fi.mod_time = Some(mod_time);
@@ -3412,10 +3431,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
for fi in metas.iter_mut() {
if fi.has_valid_erasure_geometry() {
fi.metadata = (*src_info.user_defined).clone();
if let Some(etag) = &src_info.etag {
fi.metadata.insert("etag".to_owned(), etag.clone());
}
fi.metadata.clone_from(&replacement_metadata);
fi.mod_time = Some(mod_time);
fi.version_id = version_id;
fi.versioned = src_opts.versioned || src_opts.version_suspended;
@@ -4356,7 +4372,10 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
}
fi.version_id = if let Some(vid) = opts.version_id.as_ref() {
Some(Uuid::parse_str(vid.as_str())?)
let vid = Uuid::parse_str(vid.as_str())?;
(!opts.version_suspended || !vid.is_nil()).then_some(vid)
} else if opts.version_suspended {
None
} else if opts.versioned {
Some(Uuid::new_v4())
} else {
@@ -4381,7 +4400,11 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
// Create a single object deletion request
let mut dfi = FileInfo {
name: object.to_string(),
version_id: opts.version_id.as_ref().and_then(|v| Uuid::parse_str(v).ok()),
version_id: opts
.version_id
.as_ref()
.and_then(|v| Uuid::parse_str(v).ok())
.filter(|vid| !opts.version_suspended || !vid.is_nil()),
mark_deleted: mark_delete,
deleted: delete_marker,
mod_time: Some(mod_time),
@@ -4594,7 +4617,11 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
// _lock_guard = guard_opt;
// }
let (mut fi, meta_arr, online_disks) = self.get_object_fileinfo(bucket, object, opts, true, false).await?;
let mut transition_read_opts = opts.clone();
transition_read_opts.include_part_checksums = true;
let (mut fi, meta_arr, online_disks) = self
.get_object_fileinfo(bucket, object, &transition_read_opts, true, false)
.await?;
/*if err != nil {
return Err(to_object_err(err, vec![bucket, object]));
}*/
@@ -4651,6 +4678,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
let transaction_id = transaction.transaction_id;
let dest_obj = transaction.remote_object.clone();
let mut transition_meta = (*oi.user_defined).clone();
rustfs_utils::http::remove_str(&mut transition_meta, rustfs_utils::http::SUFFIX_PART_CHECKSUMS);
transition_meta.insert("name".to_string(), object.to_string());
rustfs_utils::http::metadata_compat::insert_str(
&mut transition_meta,
@@ -4811,6 +4839,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
let mut commit_opts = opts.clone();
commit_opts.no_lock = true;
commit_opts.metadata_cache_safe = false;
commit_opts.include_part_checksums = true;
let transition_lock_guard = if opts.no_lock {
None
} else {
@@ -5063,7 +5092,12 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
achieved: 0,
});
}
let fi = self.clone().get_object_fileinfo(bucket, object, opts, true, false).await;
let mut restore_read_opts = opts.clone();
restore_read_opts.include_part_checksums = true;
let fi = self
.clone()
.get_object_fileinfo(bucket, object, &restore_read_opts, true, false)
.await;
drop(bucket_lifecycle_guard);
if let Err(err) = fi {
return set_restore_header_fn(&mut oi, Some(to_object_err(err, vec![bucket, object]))).await;
@@ -5083,7 +5117,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
expected_operation_id.to_string(),
);
}
let restore_commit_metadata = if let Some(expected_operation_id) = expected_operation_id {
let mut restore_commit_metadata = if let Some(expected_operation_id) = expected_operation_id {
let mut metadata = HashMap::new();
metadata.insert(X_AMZ_RESTORE.as_str().to_string(), "ongoing-request=\"false\"".to_string());
rustfs_utils::http::metadata_compat::insert_str(
@@ -5095,6 +5129,15 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
} else {
HashMap::new()
};
if let Some(part_checksums) =
rustfs_utils::http::get_consistent_str(&actual_fi.metadata, rustfs_utils::http::SUFFIX_PART_CHECKSUMS)
{
rustfs_utils::http::insert_str(
&mut restore_commit_metadata,
rustfs_utils::http::SUFFIX_PART_CHECKSUMS,
part_checksums.to_string(),
);
}
// The restore copy-back re-writes this same object via put_object /
// new_multipart_upload / complete_multipart_upload, each of which takes
// the object write lock in its commit phase. The caller
@@ -5694,7 +5737,7 @@ mod get_object_downstream_close_accounting_tests {
mod metadata_mutation_generation_tests {
use super::hermetic_set_disks_support::hermetic_set_disks_isolated as hermetic_set_disks;
use super::*;
use crate::disk::DiskAPI as _;
use crate::disk::{DiskAPI as _, ReadOptions};
use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _};
async fn put_and_prime(
@@ -5731,6 +5774,29 @@ mod metadata_mutation_generation_tests {
);
}
async fn persist_part_checksum_sidecar(set_disks: &Arc<SetDisks>, bucket: &str, object: &str, value: &str) {
let (mut fi, _, disks) = set_disks
.get_object_fileinfo(
bucket,
object,
&ObjectOptions {
no_lock: true,
metadata_cache_safe: false,
..Default::default()
},
true,
false,
)
.await
.expect("object metadata should be readable before adding the checksum sidecar");
rustfs_utils::http::insert_str(&mut fi.metadata, rustfs_utils::http::SUFFIX_PART_CHECKSUMS, value.to_string());
set_disks
.update_object_meta(bucket, object, fi, &disks)
.await
.expect("checksum sidecar should be persisted");
set_disks.invalidate_get_object_metadata_cache(bucket, object).await;
}
#[tokio::test]
#[serial_test::serial(metadata_cache_invalidation_probe)]
async fn metadata_semantic_mutation_generation_matrix_retires_cached_snapshot() {
@@ -5817,6 +5883,110 @@ mod metadata_mutation_generation_tests {
);
assert_retired(&set_disks, &metadata_key).await;
}
#[tokio::test]
async fn metadata_only_copy_preserves_valid_part_checksums_and_rejects_conflicting_aliases() {
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let bucket = "metadata-copy-part-checksums-bucket";
for disk in &disk_stores {
disk.make_volume(bucket).await.expect("bucket volume should be created");
}
let valid_object = "valid-sidecar";
let (mut valid_source, _) = put_and_prime(&set_disks, bucket, valid_object, b"valid sidecar body").await;
persist_part_checksum_sidecar(&set_disks, bucket, valid_object, r#"[[1,[["CRC32C","AAAAAA=="]]]]"#).await;
valid_source.metadata_only = true;
Arc::make_mut(&mut valid_source.user_defined).insert("x-amz-meta-copy".to_string(), "updated".to_string());
set_disks
.copy_object(
bucket,
valid_object,
bucket,
valid_object,
&mut valid_source,
&ObjectOptions::default(),
&ObjectOptions::default(),
)
.await
.expect("metadata-only copy should preserve a valid checksum sidecar");
let copied = set_disks
.get_object_info(
bucket,
valid_object,
&ObjectOptions {
include_part_checksums: true,
..Default::default()
},
)
.await
.expect("copied object should retain readable part checksums");
assert_eq!(
copied.parts[0]
.checksums
.as_ref()
.and_then(|checksums| checksums.get("CRC32C"))
.map(String::as_str),
Some("AAAAAA==")
);
let conflicting_object = "conflicting-sidecar";
let (mut conflicting_source, _) =
put_and_prime(&set_disks, bucket, conflicting_object, b"conflicting sidecar body").await;
let (mut fi, _, disks) = set_disks
.get_object_fileinfo(
bucket,
conflicting_object,
&ObjectOptions {
no_lock: true,
metadata_cache_safe: false,
..Default::default()
},
true,
false,
)
.await
.expect("conflicting object metadata should be readable before corruption is injected");
let rustfs_key = format!(
"{}{}",
rustfs_utils::http::RUSTFS_INTERNAL_PREFIX,
rustfs_utils::http::SUFFIX_PART_CHECKSUMS
);
let minio_key = format!(
"{}{}",
rustfs_utils::http::MINIO_INTERNAL_PREFIX,
rustfs_utils::http::SUFFIX_PART_CHECKSUMS
);
fi.metadata
.insert(rustfs_key.clone(), r#"[[1,[["CRC32C","AAAAAA=="]]]]"#.to_string());
fi.metadata
.insert(minio_key.clone(), r#"[[1,[["CRC32C","AQAAAA=="]]]]"#.to_string());
set_disks
.update_object_meta(bucket, conflicting_object, fi, &disks)
.await
.expect("conflicting aliases should be persisted for the fail-closed regression");
set_disks
.invalidate_get_object_metadata_cache(bucket, conflicting_object)
.await;
conflicting_source.metadata_only = true;
let err = set_disks
.copy_object(
bucket,
conflicting_object,
bucket,
conflicting_object,
&mut conflicting_source,
&ObjectOptions::default(),
&ObjectOptions::default(),
)
.await
.expect_err("metadata-only copy must reject conflicting checksum aliases");
assert!(matches!(err, Error::FileCorrupt));
let raw_err = disk_stores[0]
.read_version("", bucket, conflicting_object, "", &ReadOptions::default())
.await
.expect_err("the rejected copy must leave the conflicting persisted aliases fail-closed");
assert!(matches!(raw_err, crate::disk::error::DiskError::FileCorrupt));
}
}
#[cfg(all(test, feature = "test-util"))]
@@ -5878,6 +6048,7 @@ mod transition_commit_failure_tests {
}
}
#[tokio::test]
#[serial_test::serial(restore_multipart_failure_point)]
async fn multipart_restore_aborts_every_post_create_failure() {
let (temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
@@ -5924,6 +6095,30 @@ mod transition_commit_failure_tests {
)
.await
.expect("source multipart upload should complete");
let (mut source_fi, _, online_disks) = set_disks
.get_object_fileinfo(
bucket,
object,
&ObjectOptions {
no_lock: true,
metadata_cache_safe: false,
..Default::default()
},
true,
false,
)
.await
.expect("source metadata should be readable before adding the checksum sidecar");
rustfs_utils::http::insert_str(
&mut source_fi.metadata,
rustfs_utils::http::SUFFIX_PART_CHECKSUMS,
r#"[[1,[["CRC32C","AAAAAA=="]]],[2,[["CRC32C","AQAAAA=="]]]]"#.to_string(),
);
set_disks
.update_object_meta(bucket, object, source_fi, &online_disks)
.await
.expect("source checksum sidecar should be persisted before transition");
set_disks.invalidate_get_object_metadata_cache(bucket, object).await;
let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await;
set_disks
@@ -6007,9 +6202,32 @@ mod transition_commit_failure_tests {
"successful multipart completion must disarm cleanup without aborting"
);
let restored = set_disks
.get_object_info(bucket, object, &ObjectOptions::default())
.get_object_info(
bucket,
object,
&ObjectOptions {
include_part_checksums: true,
..Default::default()
},
)
.await
.expect("successful multipart restore must leave the committed object intact");
assert_eq!(
restored.parts[0]
.checksums
.as_ref()
.and_then(|checksums| checksums.get("CRC32C"))
.map(String::as_str),
Some("AAAAAA==")
);
assert_eq!(
restored.parts[1]
.checksums
.as_ref()
.and_then(|checksums| checksums.get("CRC32C"))
.map(String::as_str),
Some("AQAAAA==")
);
let restore_header = restored
.user_defined
.get(s3s::header::X_AMZ_RESTORE.as_str())
@@ -7491,6 +7709,88 @@ mod transition_upload_integrity_tests {
.expect("source object should be written")
}
#[tokio::test]
#[serial_test::serial]
async fn data_movement_tiered_metadata_is_create_only_under_object_lock() {
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let bucket = "tiered-data-movement-create-only";
let object = "object.bin";
let version_id = Uuid::new_v4();
for disk in &disk_stores {
disk.make_volume(bucket).await.expect("bucket volume should be created");
}
let mut reader = PutObjReader::from_vec(b"existing target".to_vec());
set_disks
.put_object(
bucket,
object,
&mut reader,
&ObjectOptions {
versioned: true,
version_id: Some(version_id.to_string()),
..Default::default()
},
)
.await
.expect("existing target should be written");
let conflicting = FileInfo {
volume: bucket.to_string(),
name: object.to_string(),
version_id: Some(version_id),
mod_time: Some(OffsetDateTime::UNIX_EPOCH + time::Duration::SECOND),
size: 12,
parts: vec![ObjectPartInfo {
number: 1,
size: 12,
actual_size: 12,
etag: "part-etag".to_string(),
..Default::default()
}],
transition_status: TRANSITION_COMPLETE.to_string(),
transition_tier: "WARM".to_string(),
transitioned_objname: "remote/object-a".to_string(),
transition_version: Some("remote-version-a".to_string()),
transition_version_state: rustfs_filemeta::TransitionVersionState::Exact,
fresh: true,
..Default::default()
};
let err = set_disks
.decommission_tiered_object(
bucket,
object,
&conflicting,
&ObjectOptions {
versioned: true,
version_id: Some(version_id.to_string()),
mod_time: conflicting.mod_time,
data_movement: true,
http_preconditions: Some(crate::data_movement::data_movement_target_precondition()),
..Default::default()
},
)
.await
.expect_err("data movement must not overwrite an existing tiered version");
assert!(matches!(err, StorageError::PreconditionFailed));
let (stored, _, _) = set_disks
.get_object_fileinfo(
bucket,
object,
&ObjectOptions {
version_id: Some(version_id.to_string()),
no_lock: true,
metadata_cache_safe: false,
..Default::default()
},
true,
false,
)
.await
.expect("the existing target should remain readable");
assert_ne!(stored.transition_status, TRANSITION_COMPLETE);
assert!(stored.transition_version.is_none());
}
fn transition_options(original: &ObjectInfo, tier_name: String) -> ObjectOptions {
ObjectOptions {
no_lock: true,
@@ -7545,6 +7845,29 @@ mod transition_upload_integrity_tests {
let object = "object.bin";
let payload = b"transition remote object must be bound to its transaction id".repeat(1024);
let original = write_source(&set_disks, &disk_stores, bucket, object, &payload).await;
let (mut source_fi, _, online_disks) = set_disks
.get_object_fileinfo(
bucket,
object,
&ObjectOptions {
no_lock: true,
metadata_cache_safe: false,
..Default::default()
},
true,
false,
)
.await
.expect("source metadata should be readable");
rustfs_utils::http::insert_str(
&mut source_fi.metadata,
rustfs_utils::http::SUFFIX_PART_CHECKSUMS,
r#"[[1,[["CRC32C","AAAAAA=="]]]]"#.to_string(),
);
set_disks
.update_object_meta(bucket, object, source_fi, &online_disks)
.await
.expect("source checksum sidecar should be persisted");
let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
let remote_version = Uuid::new_v4().to_string();
let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await;
@@ -7569,6 +7892,7 @@ mod transition_upload_integrity_tests {
&ObjectOptions {
no_lock: true,
metadata_cache_safe: false,
include_part_checksums: true,
..Default::default()
},
true,
@@ -7582,6 +7906,22 @@ mod transition_upload_integrity_tests {
fi.transition_version_id,
Some(Uuid::parse_str(&remote_version).expect("test version id should parse"))
);
assert_eq!(
fi.parts[0]
.checksums
.as_ref()
.and_then(|checksums| checksums.get("CRC32C"))
.map(String::as_str),
Some("AAAAAA==")
);
let remote_metadata = backend
.metadata(remote_object)
.await
.expect("remote metadata should be stored");
assert!(
!rustfs_utils::http::contains_key_str(&remote_metadata, rustfs_utils::http::SUFFIX_PART_CHECKSUMS),
"the internal checksum sidecar must not be uploaded as remote user metadata"
);
assert!(backend.contains(remote_object).await, "committed remote object should remain available");
}
@@ -7658,6 +7998,8 @@ mod transition_upload_integrity_tests {
object,
&expected,
&[],
None,
None,
"test_data_movement",
)
.await
@@ -7682,6 +8024,70 @@ mod transition_upload_integrity_tests {
assert_local_source_intact(&set_disks, bucket, object, &payload).await;
}
#[tokio::test(flavor = "current_thread", start_paused = true)]
#[serial_test::serial]
async fn data_movement_cleanup_aborts_after_bucket_fence_loss() {
let refresh_calls = Arc::new(AtomicUsize::new(0));
let lockers: Vec<Arc<dyn LockClient>> = (0..4)
.map(|_| Arc::new(LockLostRefreshClient::new(Arc::clone(&refresh_calls))) as Arc<dyn LockClient>)
.collect();
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks_with_lockers(4, 0, 2, lockers).await;
let bucket = "data-movement-cleanup-bucket-fence-lost";
let object = "object.bin";
let payload = b"lost bucket fence must preserve the source".repeat(1024);
write_source(&set_disks, &disk_stores, bucket, object, &payload).await;
let expected = set_disks
.load_file_info_versions_exact(bucket, object)
.await
.expect("source versions should be readable")
.expect("source versions should exist");
let distributed_setup = SetupTypeGuard::switch_to(SetupType::DistErasure).await;
let bucket_guard = set_disks
.new_ns_lock(bucket, crate::storage_api_contracts::bucket::BUCKET_LIFECYCLE_LOCK_OBJECT)
.await
.expect("create bucket lifecycle lock")
.get_read_lock(get_lock_acquire_timeout())
.await
.expect("acquire bucket lifecycle read lock");
let local_cleanup_setup = SetupTypeGuard::switch_to(SetupType::Erasure).await;
let barrier = crate::data_movement::SourceCleanupDeleteBarrier::install(bucket, object);
let cleanup_set = Arc::clone(&set_disks);
let cleanup = tokio::spawn(async move {
let result = crate::data_movement::cleanup_source_entry_if_unchanged(
cleanup_set,
bucket,
object,
&expected,
&[],
None,
Some(&bucket_guard),
"test_data_movement",
)
.await;
drop(bucket_guard);
result
});
barrier.wait_until_paused().await;
tokio::time::advance(Duration::from_secs(11)).await;
tokio::task::yield_now().await;
assert!(refresh_calls.load(Ordering::SeqCst) > 0, "bucket guard refresh must run before commit");
barrier.release();
let error = cleanup
.await
.expect("cleanup task should not panic")
.expect_err("cleanup must fail after its bucket lifecycle fence loses refresh quorum");
assert!(matches!(
error,
crate::data_movement::SourceCleanupError::Storage(StorageError::NamespaceLockQuorumUnavailable { .. })
));
assert_local_source_intact(&set_disks, bucket, object, &payload).await;
drop(local_cleanup_setup);
drop(distributed_setup);
}
#[tokio::test]
#[serial_test::serial]
async fn partial_remote_acceptance_cleans_exact_candidate_and_preserves_source() {
@@ -9352,6 +9758,31 @@ mod put_object_tags_early_stop_regression_tests {
.await
.expect("put_object should succeed");
let (mut fi, _, disks) = set_disks
.get_object_fileinfo(
bucket,
object,
&ObjectOptions {
no_lock: true,
metadata_cache_safe: false,
..Default::default()
},
true,
false,
)
.await
.expect("object metadata should be readable before adding the checksum sidecar");
rustfs_utils::http::insert_str(
&mut fi.metadata,
rustfs_utils::http::SUFFIX_PART_CHECKSUMS,
r#"[[1,[["CRC32C","AAAAAA=="]]]]"#.to_string(),
);
set_disks
.update_object_meta(bucket, object, fi, &disks)
.await
.expect("checksum sidecar should be persisted before tagging");
set_disks.invalidate_get_object_metadata_cache(bucket, object).await;
let tags = "unit=backlog881&stage=regression";
set_disks
.put_object_tags(bucket, object, tags, &ObjectOptions::default())
@@ -9371,6 +9802,11 @@ mod put_object_tags_early_stop_regression_tests {
Some(tags),
"disk {idx} must carry the tag written under early-stop (write set not shrunk to a read-quorum subset)"
);
assert_eq!(
rustfs_utils::http::get_consistent_str(&fi.metadata, rustfs_utils::http::SUFFIX_PART_CHECKSUMS),
Some(r#"[[1,[["CRC32C","AAAAAA=="]]]]"#),
"disk {idx} must retain the checksum sidecar across the tag metadata update"
);
}
},
)
crates/ecstore/src/set_disk/read.rs
+25 -3
View File
@@ -19,7 +19,7 @@ use crate::diagnostics::get::{
GET_METADATA_CACHE_REASON_DATA_MOVEMENT, GET_METADATA_CACHE_REASON_DELETE_MARKER, GET_METADATA_CACHE_REASON_DIST_ERASURE,
GET_METADATA_CACHE_REASON_INCL_FREE_VERSIONS, GET_METADATA_CACHE_REASON_INSUFFICIENT_CACHED_QUORUM,
GET_METADATA_CACHE_REASON_META_BUCKET, GET_METADATA_CACHE_REASON_NO_LOCK, GET_METADATA_CACHE_REASON_NOT_FOUND_OR_EXPIRED,
GET_METADATA_CACHE_REASON_NOT_READ_DATA, GET_METADATA_CACHE_REASON_PART_NUMBER,
GET_METADATA_CACHE_REASON_NOT_READ_DATA, GET_METADATA_CACHE_REASON_PART_CHECKSUMS, GET_METADATA_CACHE_REASON_PART_NUMBER,
GET_METADATA_CACHE_REASON_RAW_DATA_MOVEMENT_READ, GET_METADATA_CACHE_REASON_STALE_PUBLICATION,
GET_METADATA_CACHE_REASON_USABLE, GET_METADATA_CACHE_REASON_VERSION_ID, GET_METADATA_CACHE_REASON_VERSION_SUSPENDED,
GET_METADATA_CACHE_REASON_VERSIONED, GET_METADATA_EARLY_STOP_REASON_CONFLICTING_METADATA,
@@ -336,7 +336,7 @@ impl SetDisks {
// read_all_fileinfo_observed (see read_all_fileinfo_early_stop in
// core/io_primitives.rs); unsafe requests and callers that opt out
// (allow_early_stop=false) fall back to full-wait.
let (parts_metadata, errs, metadata_fanout_diagnostics) = Self::read_all_fileinfo_observed(
let (mut parts_metadata, errs, metadata_fanout_diagnostics) = Self::read_all_fileinfo_observed(
&disks,
"",
bucket,
@@ -385,8 +385,17 @@ impl SetDisks {
return Err(to_object_err(err.into(), vec![bucket, object]));
}
let (op_online_disks, fi, fileinfo_selection_quorum) =
let (op_online_disks, mut fi, fileinfo_selection_quorum) =
Self::select_valid_fileinfo(&disks, &parts_metadata, &errs, vid.as_str(), read_quorum, write_quorum)?;
let include_part_checksums =
opts.include_part_checksums || opts.part_number.is_some() || opts.data_movement || opts.raw_data_movement_read;
if include_part_checksums {
Self::hydrate_selected_fileinfo_part_checksums(&mut fi)?;
} else {
for metadata in std::iter::once(&mut fi).chain(parts_metadata.iter_mut()) {
rustfs_utils::http::remove_str(&mut metadata.metadata, rustfs_utils::http::SUFFIX_PART_CHECKSUMS);
}
}
metadata_fanout_diagnostics.record_quorum_candidate_latency(GET_OBJECT_PATH_LEGACY_DUPLEX, fileinfo_selection_quorum);
if errs.iter().any(|err| err.is_some()) {
let version_id = resolved_read_repair_version_id(&fi, opts.version_id.as_deref());
@@ -1813,6 +1822,9 @@ fn get_object_metadata_cache_request_bypass_reason(bucket: &str, opts: &ObjectOp
if opts.part_number.is_some() {
return Some(GET_METADATA_CACHE_REASON_PART_NUMBER);
}
if opts.include_part_checksums {
return Some(GET_METADATA_CACHE_REASON_PART_CHECKSUMS);
}
if opts.data_movement {
return Some(GET_METADATA_CACHE_REASON_DATA_MOVEMENT);
}
@@ -2484,6 +2496,16 @@ mod metadata_cache_tests {
Some(GET_METADATA_CACHE_REASON_PART_NUMBER)
);
opts = ObjectOptions {
include_part_checksums: true,
..Default::default()
};
assert!(!is_get_object_metadata_cache_request_eligible("bucket", &opts, true));
assert_eq!(
get_object_metadata_cache_request_bypass_reason("bucket", &opts, true),
Some(GET_METADATA_CACHE_REASON_PART_CHECKSUMS)
);
opts = ObjectOptions {
data_movement: true,
..Default::default()
crates/ecstore/src/set_disk/replication.rs
+2
View File
@@ -75,6 +75,7 @@ impl SetDisks {
version_id,
versioned: opts.versioned,
version_suspended: opts.version_suspended,
include_part_checksums: true,
..Default::default()
};
let (mut fi, _, disks) = self
@@ -141,6 +142,7 @@ impl SetDisks {
version_id,
versioned: opts.versioned,
version_suspended: opts.version_suspended,
include_part_checksums: true,
..Default::default()
};
let (mut fi, _, disks) = self