From f5929a830594805314f2e4cd51a4603e2a85ec53 Mon Sep 17 00:00:00 2001 From: cxymds Date: Fri, 7 Aug 2026 20:30:09 +0800 Subject: [PATCH] fix(ecstore): preserve newer writes during data movement (#5798) --- crates/ecstore/src/data_movement/mod.rs | 245 +++++++++++++++++++++- crates/ecstore/src/set_disk/ops/object.rs | 90 +++++++- crates/ecstore/src/store/init.rs | 140 +++++++++++++ 3 files changed, 466 insertions(+), 9 deletions(-) diff --git a/crates/ecstore/src/data_movement/mod.rs b/crates/ecstore/src/data_movement/mod.rs index 2d026b330..503483bd6 100644 --- a/crates/ecstore/src/data_movement/mod.rs +++ b/crates/ecstore/src/data_movement/mod.rs @@ -25,7 +25,7 @@ use crate::set_disk::{SetDisks, get_lock_acquire_timeout}; use crate::storage_api_contracts::{ multipart::{CompletePart, MultipartOperations as _}, namespace::NamespaceLocking as _, - object::{ObjectIO as _, ObjectOperations as _}, + object::{HTTPPreconditions, ObjectIO as _, ObjectOperations as _}, }; use crate::store::ECStore; use bytes::Bytes; @@ -228,6 +228,7 @@ fn data_movement_complete_multipart_opts(object_info: &ObjectInfo, src_pool_idx: ObjectOptions { versioned: object_info.version_id.is_some(), version_id: object_info.version_id.as_ref().map(|v| v.to_string()), + http_preconditions: data_movement_unversioned_target_precondition(object_info), data_movement: true, mod_time: object_info.mod_time, preserve_etag: object_info.etag.clone(), @@ -242,6 +243,7 @@ fn data_movement_put_object_opts(object_info: &ObjectInfo, src_pool_idx: usize) src_pool_idx, data_movement: true, version_id: object_info.version_id.as_ref().map(|v| v.to_string()), + http_preconditions: data_movement_unversioned_target_precondition(object_info), mod_time: object_info.mod_time, user_defined: data_movement_user_defined(object_info), preserve_etag: object_info.etag.clone(), @@ -249,6 +251,17 @@ fn data_movement_put_object_opts(object_info: &ObjectInfo, src_pool_idx: usize) } } +fn is_unversioned_data_movement_object(object_info: &ObjectInfo) -> bool { + object_info.version_id.is_none_or(|version_id| version_id.is_nil()) +} + +fn data_movement_unversioned_target_precondition(object_info: &ObjectInfo) -> Option { + is_unversioned_data_movement_object(object_info).then(|| HTTPPreconditions { + if_none_match: Some("*".to_string()), + ..Default::default() + }) +} + fn data_movement_put_object_reader( bucket: &str, object_info: &ObjectInfo, @@ -337,7 +350,7 @@ fn schedule_data_movement_multipart_abort_cleanup( } fn should_check_data_movement_overwrite_resume(err: &Error) -> bool { - is_err_data_movement_overwrite(err) + is_err_data_movement_overwrite(err) || matches!(err, Error::PreconditionFailed) } fn effective_actual_size(info: &ObjectInfo) -> Option { @@ -403,6 +416,16 @@ fn is_equivalent_data_movement_object(source: &ObjectInfo, target: &ObjectInfo) && are_equivalent_data_movement_parts(&source.parts, &target.parts) } +fn is_superseding_unversioned_data_movement_object(source: &ObjectInfo, target: &ObjectInfo) -> bool { + is_unversioned_data_movement_object(source) + && is_unversioned_data_movement_object(target) + && !target.delete_marker + && source + .mod_time + .zip(target.mod_time) + .is_some_and(|(source_time, target_time)| target_time > source_time) +} + #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] struct SourceCleanupPartIdentity { number: usize, @@ -701,7 +724,11 @@ fn resolve_data_movement_overwrite_resume_result( return Ok(false); }; - Ok(is_equivalent_data_movement_object(source, &target)) + if is_equivalent_data_movement_object(source, &target) { + return Ok(true); + } + + Ok(matches!(err, Error::PreconditionFailed) && is_superseding_unversioned_data_movement_object(source, &target)) } async fn should_treat_data_movement_overwrite_as_complete( @@ -912,7 +939,6 @@ pub(crate) async fn migrate_object( bucket.as_str(), object_info.name.as_str() ); - mark_multipart_upload_completed(&abort_multipart_flag); return Ok(()); } @@ -931,6 +957,32 @@ pub(crate) async fn migrate_object( } .await; + if multipart_result.is_ok() && should_abort_multipart_upload(&abort_multipart_flag) { + let abort_result = match store.pools.get(target_pool_idx) { + Some(pool) => { + pool.abort_multipart_upload(&bucket, &object_info.name, &res.upload_id, &ObjectOptions::default()) + .await + } + None => Err(Error::other(format!( + "{op_label}: target pool {target_pool_idx} is out of range while aborting superseded multipart upload" + ))), + }; + if let Err(abort_err) = abort_result + && !is_err_invalid_upload_id(&abort_err) + { + error!("{op_label}: abort superseded multipart upload err {:?}", &abort_err); + schedule_data_movement_multipart_abort_cleanup( + store.clone(), + target_pool_idx, + bucket.clone(), + object_info.name.clone(), + res.upload_id.clone(), + op_label, + ); + } + return Ok(()); + } + if let Err(primary_err) = multipart_result { if should_abort_multipart_upload(&abort_multipart_flag) { return match store @@ -1245,12 +1297,13 @@ mod tests { } #[test] - fn test_should_check_data_movement_overwrite_resume_only_for_overwrite_error() { + fn test_should_check_data_movement_overwrite_resume_accepts_conflict_errors() { assert!(should_check_data_movement_overwrite_resume(&Error::DataMovementOverwriteErr( "bucket-a".to_string(), "object-a".to_string(), "version-a".to_string(), ))); + assert!(should_check_data_movement_overwrite_resume(&Error::PreconditionFailed)); assert!(!should_check_data_movement_overwrite_resume(&Error::SlowDown)); } @@ -1625,7 +1678,7 @@ mod tests { #[test] fn test_data_movement_complete_multipart_opts_preserves_mod_time_version_and_etag() { let mod_time = OffsetDateTime::now_utc(); - let version_id = Uuid::nil(); + let version_id = Uuid::from_u128(7); let object_info = ObjectInfo { version_id: Some(version_id), mod_time: Some(mod_time), @@ -1641,11 +1694,12 @@ mod tests { assert_eq!(opts.version_id.as_deref(), Some(version_id.to_string().as_str())); assert_eq!(opts.preserve_etag.as_deref(), Some("etag-value")); assert_eq!(opts.src_pool_idx, 7); + assert!(opts.http_preconditions.is_none()); } #[test] fn test_data_movement_put_object_opts_preserves_version_and_etag() { - let version_id = Uuid::nil(); + let version_id = Uuid::from_u128(9); let object_info = ObjectInfo { version_id: Some(version_id), mod_time: Some(OffsetDateTime::UNIX_EPOCH), @@ -1663,6 +1717,35 @@ mod tests { assert_eq!(opts.src_pool_idx, 9); assert!(opts.data_movement); assert_eq!(opts.mod_time, object_info.mod_time); + assert!(opts.http_preconditions.is_none()); + } + + #[test] + fn test_data_movement_unversioned_put_and_complete_require_absent_target() { + for version_id in [None, Some(Uuid::nil())] { + let object_info = ObjectInfo { + version_id, + ..Default::default() + }; + + let put_opts = data_movement_put_object_opts(&object_info, 9); + let complete_opts = data_movement_complete_multipart_opts(&object_info, 9); + + assert_eq!( + put_opts + .http_preconditions + .as_ref() + .and_then(HTTPPreconditions::if_none_match_value), + Some("*") + ); + assert_eq!( + complete_opts + .http_preconditions + .as_ref() + .and_then(HTTPPreconditions::if_none_match_value), + Some("*") + ); + } } #[test] @@ -1911,6 +1994,154 @@ mod tests { assert!(should_resume); } + #[test] + fn test_precondition_conflict_accepts_newer_unversioned_target() { + for version_id in [None, Some(Uuid::nil())] { + let source = ObjectInfo { + version_id, + size: 128, + etag: Some("etag-source".to_string()), + mod_time: Some(OffsetDateTime::UNIX_EPOCH), + ..Default::default() + }; + let target = ObjectInfo { + etag: Some("etag-client-write".to_string()), + mod_time: OffsetDateTime::UNIX_EPOCH.checked_add(time::Duration::SECOND), + ..source.clone() + }; + + let should_resume = + resolve_data_movement_overwrite_resume_result(&Error::PreconditionFailed, Ok(Some(target)), &source, 0, 1) + .expect("precondition conflict target should be evaluated"); + + assert!(should_resume); + } + } + + #[test] + fn test_precondition_conflict_accepts_equivalent_target() { + let source = ObjectInfo { + size: 128, + etag: Some("etag-source".to_string()), + mod_time: Some(OffsetDateTime::UNIX_EPOCH), + ..Default::default() + }; + + let should_resume = + resolve_data_movement_overwrite_resume_result(&Error::PreconditionFailed, Ok(Some(source.clone())), &source, 0, 1) + .expect("equivalent precondition target should be evaluated"); + + assert!(should_resume); + } + + #[test] + fn test_precondition_conflict_rejects_non_newer_unversioned_target() { + let source = ObjectInfo { + size: 128, + etag: Some("etag-source".to_string()), + mod_time: Some(OffsetDateTime::UNIX_EPOCH), + ..Default::default() + }; + let target = ObjectInfo { + etag: Some("etag-conflict".to_string()), + ..source.clone() + }; + + let should_resume = + resolve_data_movement_overwrite_resume_result(&Error::PreconditionFailed, Ok(Some(target)), &source, 0, 1) + .expect("precondition conflict target should be evaluated"); + + assert!(!should_resume); + } + + #[test] + fn test_precondition_conflict_rejects_newer_delete_marker() { + let source = ObjectInfo { + size: 128, + etag: Some("etag-source".to_string()), + mod_time: Some(OffsetDateTime::UNIX_EPOCH), + ..Default::default() + }; + let target = ObjectInfo { + delete_marker: true, + etag: None, + mod_time: OffsetDateTime::UNIX_EPOCH.checked_add(time::Duration::SECOND), + ..source.clone() + }; + + let should_resume = + resolve_data_movement_overwrite_resume_result(&Error::PreconditionFailed, Ok(Some(target)), &source, 0, 1) + .expect("delete marker conflict should be evaluated"); + + assert!(!should_resume); + } + + #[test] + fn test_overwrite_error_rejects_newer_unversioned_target() { + let source = ObjectInfo { + size: 128, + etag: Some("etag-source".to_string()), + mod_time: Some(OffsetDateTime::UNIX_EPOCH), + ..Default::default() + }; + let target = ObjectInfo { + etag: Some("etag-client-write".to_string()), + mod_time: OffsetDateTime::UNIX_EPOCH.checked_add(time::Duration::SECOND), + ..source.clone() + }; + let err = Error::DataMovementOverwriteErr("bucket".to_string(), "object".to_string(), "version".to_string()); + + let should_resume = resolve_data_movement_overwrite_resume_result(&err, Ok(Some(target)), &source, 0, 1) + .expect("pool-selection overwrite must require target equivalence"); + + assert!(!should_resume); + } + + #[test] + fn test_precondition_conflict_rejects_newer_versioned_target() { + let source = ObjectInfo { + size: 128, + etag: Some("etag-source".to_string()), + mod_time: Some(OffsetDateTime::UNIX_EPOCH), + ..Default::default() + }; + let target = ObjectInfo { + version_id: Some(Uuid::from_u128(2)), + etag: Some("etag-conflict".to_string()), + mod_time: OffsetDateTime::UNIX_EPOCH.checked_add(time::Duration::SECOND), + ..source.clone() + }; + + let should_resume = + resolve_data_movement_overwrite_resume_result(&Error::PreconditionFailed, Ok(Some(target)), &source, 0, 1) + .expect("versioned conflict target should be evaluated"); + + assert!(!should_resume); + } + + #[test] + fn test_precondition_conflict_rejects_versioned_source_with_unversioned_target() { + let source = ObjectInfo { + version_id: Some(Uuid::from_u128(1)), + size: 128, + etag: Some("etag-source".to_string()), + mod_time: Some(OffsetDateTime::UNIX_EPOCH), + ..Default::default() + }; + let target = ObjectInfo { + version_id: None, + etag: Some("etag-conflict".to_string()), + mod_time: OffsetDateTime::UNIX_EPOCH.checked_add(time::Duration::SECOND), + ..source.clone() + }; + + let should_resume = + resolve_data_movement_overwrite_resume_result(&Error::PreconditionFailed, Ok(Some(target)), &source, 0, 1) + .expect("versioned source conflict should be evaluated"); + + assert!(!should_resume); + } + #[test] fn test_rebalance_overwrite_resume_accepts_equivalent_target_version() { let source = ObjectInfo { diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 2f994e262..c58184062 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -972,8 +972,9 @@ impl SetDisks { let mut object_lock_guard = None; let mut bucket_lifecycle_guard = None; + let deferred_data_movement_precondition = opts.data_movement && opts.http_preconditions.is_some(); - if opts.http_preconditions.is_some() { + if opts.http_preconditions.is_some() && !deferred_data_movement_precondition { if !opts.no_lock { if let Some(expected_incarnation_id) = opts.expected_bucket_incarnation_id && opts.bucket_lifecycle_lock_fence.is_none() @@ -1330,6 +1331,10 @@ impl SetDisks { #[cfg(test)] pause_put_object_commit(bucket, object, PutObjectCommitPause::AfterNamespace).await; + if deferred_data_movement_precondition && let Some(err) = self.check_write_precondition(bucket, object, opts).await { + return Err(err); + } + // Generate ordinary PUT timestamps under the commit lock so version // ordering follows durable commit ordering when writers queued on // the same object. Internal callers with an explicit timestamp keep @@ -5780,7 +5785,7 @@ mod transition_commit_failure_tests { use http::HeaderMap; use rustfs_filemeta::{RestoreStatusOps as _, parse_restore_obj_status}; use s3s::dto::RestoreRequest; - use tokio::io::AsyncReadExt; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; fn restore_operation_id_metadata(operation_id: Uuid) -> HashMap { let mut metadata = HashMap::new(); @@ -8721,6 +8726,87 @@ mod put_object_tmp_cleanup_tests { ); } + #[tokio::test] + async fn data_movement_precondition_is_rechecked_at_commit() { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "data-movement-commit-precondition"; + let object = "object"; + for disk in &disk_stores { + disk.make_volume(bucket).await.expect("bucket volume should be created"); + } + + let migration_body = vec![b'm'; 64 * 1024]; + let split = migration_body.len() / 2; + let (mut source, stream) = tokio::io::duplex(64); + let hash_reader = HashReader::from_stream( + stream, + i64::try_from(migration_body.len()).expect("migration body length should fit i64"), + i64::try_from(migration_body.len()).expect("migration body length should fit i64"), + None, + None, + false, + ) + .expect("migration hash reader should be created"); + let migration_store = Arc::clone(&set_disks); + let migration = tokio::spawn(async move { + let mut reader = PutObjReader::new(hash_reader); + migration_store + .put_object( + bucket, + object, + &mut reader, + &ObjectOptions { + data_movement: true, + http_preconditions: Some(HTTPPreconditions { + if_none_match: Some("*".to_string()), + ..Default::default() + }), + ..Default::default() + }, + ) + .await + }); + + source + .write_all(&migration_body[..split]) + .await + .expect("migration should consume the first half before commit"); + let mut client_reader = PutObjReader::from_vec(b"new client body".to_vec()); + tokio::time::timeout( + Duration::from_secs(5), + set_disks.put_object(bucket, object, &mut client_reader, &ObjectOptions::default()), + ) + .await + .expect("client write must not wait for the migration body") + .expect("client write should commit while migration waits for the remaining source body"); + let barrier = PutObjectCommitBarrier::install(bucket, object, PutObjectCommitPause::AfterNamespace); + source + .write_all(&migration_body[split..]) + .await + .expect("migration should consume the remaining source body"); + drop(source); + barrier.wait_until_paused().await; + barrier.release(); + + let err = migration + .await + .expect("migration task should join") + .expect_err("migration must recheck the target after acquiring its commit lock"); + assert_eq!(err, StorageError::PreconditionFailed); + + let mut reader = set_disks + .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default()) + .await + .expect("client object should remain readable"); + let mut body = Vec::new(); + reader + .stream + .read_to_end(&mut body) + .await + .expect("client object should drain"); + assert_eq!(body, b"new client body"); + } + #[tokio::test] async fn metadata_copy_no_lock_aborts_after_outer_namespace_lock_loss() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index a5ce7baf4..3304a4eeb 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -611,6 +611,7 @@ mod tests { }; use http::HeaderMap; use rustfs_config::server_config::KVS; + use rustfs_filemeta::ObjectPartInfo; #[cfg(feature = "test-util")] use rustfs_protos::{TIER_MUTATION_RPC_PROTOCOL_VERSION, TierMutationRpcPhase}; use std::{ @@ -1153,6 +1154,145 @@ mod tests { (instance_ctx, store, shutdown) } + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn data_movement_conflicts_preserve_newer_target_and_abort_staging() { + let temp_dir = tempfile::tempdir().expect("create data movement store dir"); + let (_ctx, store, _shutdown) = + without_storage_class_env(build_isolated_test_store(temp_dir.path(), "data-movement-conflict-convergence", &[4, 4])) + .await; + crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await; + + let bucket = format!("data-movement-conflict-{}", uuid::Uuid::new_v4()); + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("create data movement bucket"); + let source_mod_time = OffsetDateTime::UNIX_EPOCH; + let target_mod_time = source_mod_time + time::Duration::SECOND; + + let object = "single-object"; + let target_body = b"newer client body".to_vec(); + let mut target_reader = PutObjReader::from_vec(target_body.clone()); + store.pools[1] + .put_object( + &bucket, + object, + &mut target_reader, + &ObjectOptions { + mod_time: Some(target_mod_time), + ..Default::default() + }, + ) + .await + .expect("write newer single-part target"); + + let source_body = b"stale migration body".to_vec(); + crate::data_movement::migrate_object( + store.clone(), + 0, + bucket.clone(), + GetObjectReader { + stream: Box::new(Cursor::new(source_body.clone())), + object_info: ObjectInfo { + bucket: bucket.clone(), + name: object.to_string(), + size: i64::try_from(source_body.len()).expect("single source size should fit i64"), + actual_size: i64::try_from(source_body.len()).expect("single source size should fit i64"), + etag: Some("0123456789abcdef0123456789abcdef".to_string()), + mod_time: Some(source_mod_time), + ..Default::default() + }, + buffered_body: None, + body_source: Default::default(), + }, + "test_data_movement", + ) + .await + .expect("newer single-part target should converge migration"); + + let mut reader = store + .get_object_reader(&bucket, object, None, HeaderMap::new(), &ObjectOptions::default()) + .await + .expect("read converged single-part target"); + let mut body = Vec::new(); + reader.stream.read_to_end(&mut body).await.expect("drain single-part target"); + assert_eq!(body, target_body); + + let multipart_object = "multipart-object"; + let multipart_target_body = b"newer multipart client body".to_vec(); + let mut multipart_target_reader = PutObjReader::from_vec(multipart_target_body.clone()); + store.pools[1] + .put_object( + &bucket, + multipart_object, + &mut multipart_target_reader, + &ObjectOptions { + mod_time: Some(target_mod_time), + ..Default::default() + }, + ) + .await + .expect("write newer multipart target"); + + let first_part_size = 5 * 1024 * 1024; + let mut multipart_source_body = vec![b'a'; first_part_size]; + multipart_source_body.push(b'b'); + let multipart_source_size = i64::try_from(multipart_source_body.len()).expect("multipart source size should fit i64"); + crate::data_movement::migrate_object( + store.clone(), + 0, + bucket.clone(), + GetObjectReader { + stream: Box::new(Cursor::new(multipart_source_body)), + object_info: ObjectInfo { + bucket: bucket.clone(), + name: multipart_object.to_string(), + size: multipart_source_size, + actual_size: multipart_source_size, + etag: Some("source-multipart-etag-2".to_string()), + mod_time: Some(source_mod_time), + parts: Arc::new(vec![ + ObjectPartInfo { + number: 1, + size: first_part_size, + actual_size: i64::try_from(first_part_size).expect("first part size should fit i64"), + etag: "source-part-1".to_string(), + ..Default::default() + }, + ObjectPartInfo { + number: 2, + size: 1, + actual_size: 1, + etag: "source-part-2".to_string(), + ..Default::default() + }, + ]), + ..Default::default() + }, + buffered_body: None, + body_source: Default::default(), + }, + "test_data_movement", + ) + .await + .expect("newer multipart target should converge migration"); + + let uploads = store.pools[1] + .list_multipart_uploads(&bucket, multipart_object, None, None, None, 100) + .await + .expect("list target pool multipart uploads"); + assert!(uploads.uploads.is_empty(), "superseded migration staging must be aborted"); + + let mut reader = store + .get_object_reader(&bucket, multipart_object, None, HeaderMap::new(), &ObjectOptions::default()) + .await + .expect("read converged multipart target"); + let mut body = Vec::new(); + reader.stream.read_to_end(&mut body).await.expect("drain multipart target"); + assert_eq!(body, multipart_target_body); + } + #[cfg(feature = "test-util")] async fn tier_delete_journal_count(store: Arc) -> usize { store