fix(ecstore): allow data movement size sentinel

crates/ecstore/src/disk/local.rs

@@ -2028,14 +2028,17 @@ static RENAME_DATA_REMOVE_DST_BASE_BEFORE_COMMIT: std::sync::Mutex<Option<(Strin
 #[cfg(test)]
 type InlinePreparationHook = Box<dyn FnOnce() + Send>;
 #[cfg(test)]
+type RenameDataPublicationHookKey = (PathBuf, String, String);
+#[cfg(test)]
 static INLINE_PREPARATION_BEFORE_BACKUP: std::sync::LazyLock<std::sync::Mutex<HashMap<String, InlinePreparationHook>>> =
     std::sync::LazyLock::new(|| std::sync::Mutex::new(HashMap::new()));
 #[cfg(test)]
 static INLINE_BEFORE_FILE_SYNC_ADMISSION: std::sync::LazyLock<std::sync::Mutex<HashMap<String, InlinePreparationHook>>> =
     std::sync::LazyLock::new(|| std::sync::Mutex::new(HashMap::new()));
 #[cfg(test)]
-static RENAME_DATA_AFTER_FIRST_PUBLICATION: std::sync::LazyLock<std::sync::Mutex<HashMap<String, InlinePreparationHook>>> =
-    std::sync::LazyLock::new(|| std::sync::Mutex::new(HashMap::new()));
+static RENAME_DATA_AFTER_FIRST_PUBLICATION: std::sync::LazyLock<
+    std::sync::Mutex<HashMap<RenameDataPublicationHookKey, InlinePreparationHook>>,
+> = std::sync::LazyLock::new(|| std::sync::Mutex::new(HashMap::new()));
 #[cfg(test)]
 static OWNED_FILE_WRITE_BEFORE_OPEN: std::sync::LazyLock<std::sync::Mutex<HashMap<PathBuf, InlinePreparationHook>>> =
     std::sync::LazyLock::new(|| std::sync::Mutex::new(HashMap::new()));
@@ -2107,11 +2110,11 @@ fn set_inline_before_file_sync_admission(dst_path: &str, hook: impl FnOnce() + S
 }
 
 #[cfg(test)]
-fn set_rename_data_after_first_publication(dst_path: &str, hook: impl FnOnce() + Send + 'static) {
+fn set_rename_data_after_first_publication(root: &Path, dst_volume: &str, dst_path: &str, hook: impl FnOnce() + Send + 'static) {
     RENAME_DATA_AFTER_FIRST_PUBLICATION
         .lock()
         .expect("test publication hook lock should not be poisoned")
-        .insert(dst_path.to_string(), Box::new(hook));
+        .insert((root.to_path_buf(), dst_volume.to_string(), dst_path.to_string()), Box::new(hook));
 }
 
 #[cfg(test)]
@@ -2263,11 +2266,11 @@ fn run_inline_before_file_sync_admission(dst_path: &str) {
 }
 
 #[cfg(test)]
-fn run_rename_data_after_first_publication(dst_path: &str) {
+fn run_rename_data_after_first_publication(root: &Path, dst_volume: &str, dst_path: &str) {
     let hook = RENAME_DATA_AFTER_FIRST_PUBLICATION
         .lock()
         .expect("test publication hook lock should not be poisoned")
-        .remove(dst_path);
+        .remove(&(root.to_path_buf(), dst_volume.to_string(), dst_path.to_string()));
     if let Some(hook) = hook {
         hook();
     }
@@ -2365,9 +2368,6 @@ async fn remove_dst_base_before_commit(
 #[cfg(not(test))]
 fn run_inline_preparation_before_backup(_dst_path: &str) {}
 
-#[cfg(not(test))]
-fn run_rename_data_after_first_publication(_dst_path: &str) {}
-
 #[cfg(not(test))]
 fn should_fail_after_delete_data_staged(_path: &str) -> bool {
     false
@@ -9015,8 +9015,9 @@ impl DiskAPI for LocalDisk {
                 .await?;
                 return Err(err);
             }
+            #[cfg(test)]
             if has_data_dir_path.is_some() {
-                run_rename_data_after_first_publication(dst_path);
+                run_rename_data_after_first_publication(&self.root, dst_volume, dst_path);
             }
 
             // Crash-consistency injection: hard power loss after the data dir
@@ -9449,7 +9450,8 @@ impl DiskAPI for LocalDisk {
                     let _ = remove_file_if_exists(staged_backup);
                     return Err(err);
                 }
-                run_rename_data_after_first_publication(dst_path);
+                #[cfg(test)]
+                run_rename_data_after_first_publication(&self.root, dst_volume, dst_path);
                 if sync {
                     file_sync_admission = Some(
                         os::acquire_file_sync_admission(self.file_sync_permits.clone())
@@ -13133,7 +13135,7 @@ mod test {
         let replacement_staging_parent_for_hook = replacement_staging_parent.clone();
         let staged_metadata_for_hook = staged_metadata.clone();
         let replacement_staged_metadata_for_hook = replacement_staged_metadata.clone();
-        set_rename_data_after_first_publication(object, move || {
+        set_rename_data_after_first_publication(&disk.root, bucket, object, move || {
             std::fs::rename(&object_dir_for_hook, &replacement_dir_for_hook)
                 .expect_err("the destination object identity must remain pinned until xl.meta commits");
             std::fs::rename(&staging_parent_for_hook, &replacement_staging_parent_for_hook)
@@ -13400,7 +13402,7 @@ mod test {
         let replacement_dir_for_hook = replacement_dir.clone();
         let staged_metadata_for_hook = staged_metadata.clone();
         let replacement_staged_metadata_for_hook = replacement_staged_metadata.clone();
-        set_rename_data_after_first_publication(object, move || {
+        set_rename_data_after_first_publication(&disk.root, bucket, object, move || {
             std::fs::rename(&object_dir_for_hook, &replacement_dir_for_hook)
                 .expect_err("the destination object identity must remain pinned after publishing its rollback backup");
             std::fs::rename(&staged_metadata_for_hook, &replacement_staged_metadata_for_hook)
@@ -13766,7 +13768,7 @@ mod test {
 
         let (entered_tx, entered_rx) = mpsc::channel();
         let (release_tx, release_rx) = mpsc::channel();
-        set_rename_data_after_first_publication(object, move || {
+        set_rename_data_after_first_publication(&disk.root, bucket, object, move || {
             entered_tx.send(()).expect("signal first publication");
             release_rx.recv().expect("wait while delete_volume is blocked");
         });
@@ -14360,7 +14362,7 @@ mod test {
 
         let (published_tx, published_rx) = mpsc::channel();
         let (release_tx, release_rx) = mpsc::channel();
-        set_rename_data_after_first_publication(object, move || {
+        set_rename_data_after_first_publication(&disk.root, bucket, object, move || {
             published_tx.send(()).expect("signal backup publication");
             release_rx.recv().expect("wait for lock-order assertion");
         });

crates/ecstore/src/set_disk/ops/multipart.rs

@@ -2034,7 +2034,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
             }
 
             object_size = object_size.checked_add(ext_part.size).ok_or(Error::PartMissingOrCorrupt)?;
-            if ext_part.actual_size < 0 && (!opts.replication_request || quota_context.is_enforced()) {
+            if ext_part.actual_size < 0 && (quota_context.is_enforced() || (!opts.replication_request && !opts.data_movement)) {
                 return Err(Error::PartMissingOrCorrupt);
             }
             let normalized_actual_size = if ext_part.actual_size >= 0 && !transformed_object {

rustfs/src/storage/rpc/node_service.rs

@@ -2943,7 +2943,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn snapshot_lease_acquire_and_renew_handlers_fail_closed() {
+    async fn snapshot_lease_acquire_and_renew_missing_disk_fail_closed() {
         let service = make_server();
         let disk = "http://node-a:9000/data/rustfs0".to_string();
 
@@ -2960,7 +2960,7 @@ mod tests {
         let acquire = service
             .acquire_snapshot_lease(acquire)
             .await
-            .expect("disabled acquire should return a protocol response")
+            .expect("missing-disk acquire should return a protocol response")
             .into_inner();
 
         let mut renew = Request::new(SnapshotLeaseRenewRequest {
@@ -2977,14 +2977,14 @@ mod tests {
         let renew = service
             .renew_snapshot_lease(renew)
             .await
-            .expect("disabled renew should return a protocol response")
+            .expect("missing-disk renew should return a protocol response")
             .into_inner();
 
         for response in [acquire, renew] {
             assert!(!response.success);
             assert!(response.token.is_empty());
             assert_eq!(response.protocol_version, 1);
-            assert_eq!(response.error, Some(DiskError::UnsupportedDisk.into()));
+            assert_eq!(response.error, Some(DiskError::other("cannot find disk").into()));
         }
     }