fix(ecstore): preserve newer writes during data movement (#5798)

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<HTTPPreconditions> {
+    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<i64> {
@@ -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 {

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<String, String> {
         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;

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<crate::store::ECStore>) -> usize {
         store