fix(ecstore): isolate inline rollback cleanup (#5703)

crates/ecstore/src/cluster/rpc/remote_disk.rs

@@ -3381,6 +3381,8 @@ mod tests {
         crate::cluster::rpc::runtime_sources::reset_internode_metrics_for_test();
         let response = RenameDataResp {
             old_data_dir: Some(Uuid::new_v4()),
+            rollback_data_dir: Some(Uuid::new_v4()),
+            cleanup_data_dir: Some(Uuid::new_v4()),
             sign: Some(vec![0x14, 0x35]),
             old_current_size: Some(crate::disk::OldCurrentSize::Present(64 * 1024)),
         };
@@ -3394,6 +3396,8 @@ mod tests {
         let decode_errors_after = crate::cluster::rpc::runtime_sources::internode_msgpack_json_decode_error_total_for_test();
 
         assert_eq!(decoded.old_data_dir, response.old_data_dir);
+        assert_eq!(decoded.rollback_data_dir, response.rollback_data_dir);
+        assert_eq!(decoded.cleanup_data_dir, response.cleanup_data_dir);
         assert_eq!(decoded.sign, response.sign);
         assert_eq!(decoded.old_current_size, response.old_current_size);
         assert!(
@@ -3758,6 +3762,26 @@ mod tests {
         );
     }
 
+    #[test]
+    fn rename_data_resp_named_msgpack_is_smaller_than_json() {
+        let response = RenameDataResp {
+            old_data_dir: Some(Uuid::new_v4()),
+            rollback_data_dir: Some(Uuid::new_v4()),
+            cleanup_data_dir: Some(Uuid::new_v4()),
+            sign: Some(vec![1_u8; 32]),
+            old_current_size: Some(crate::disk::OldCurrentSize::Present(4096)),
+        };
+        let json = serde_json::to_vec(&response).expect("rename data response json should encode");
+        let named_msgpack = rmp_serde::encode::to_vec_named(&response).expect("rename data response named msgpack should encode");
+
+        assert!(
+            named_msgpack.len() < json.len(),
+            "expected named msgpack payload to be smaller than json (msgpack={}, json={})",
+            named_msgpack.len(),
+            json.len()
+        );
+    }
+
     #[derive(Debug, Default)]
     struct SinkTestWriter;
 

crates/ecstore/src/disk/local.rs

@@ -8156,7 +8156,9 @@ impl DiskAPI for LocalDisk {
             }
 
             Ok(RenameDataResp {
-                old_data_dir: rollback_data_dir,
+                old_data_dir: has_old_data_dir,
+                rollback_data_dir,
+                cleanup_data_dir: has_old_data_dir,
                 sign: version_signature,
                 old_current_size,
             })
@@ -8347,8 +8349,9 @@ impl DiskAPI for LocalDisk {
                     let _ = remove_file_if_exists(backup_path);
                 }
 
-                Ok::<(Option<Uuid>, Option<Vec<u8>>, Option<OldCurrentSize>), std::io::Error>((
+                Ok::<(Option<Uuid>, Option<Uuid>, Option<Vec<u8>>, Option<OldCurrentSize>), std::io::Error>((
                     rollback_data_dir,
+                    old_data_dir,
                     version_signature,
                     old_current_size,
                 ))
@@ -8363,7 +8366,7 @@ impl DiskAPI for LocalDisk {
             // invalidate itself, so it is done here. Inline objects carry their
             // data in xl.meta rather than separate part inodes, so this is mostly
             // defensive, but it keeps the inline and streaming branches consistent.
-            let (old_data_dir, version_signature, old_current_size) = match inline_commit {
+            let (old_data_dir, cleanup_data_dir, version_signature, old_current_size) = match inline_commit {
                 Ok(committed) => committed,
                 Err(err) => {
                     for part_path in &invalidate_part_paths {
@@ -8396,7 +8399,9 @@ impl DiskAPI for LocalDisk {
             }
 
             Ok(RenameDataResp {
-                old_data_dir,
+                old_data_dir: cleanup_data_dir,
+                rollback_data_dir: old_data_dir,
+                cleanup_data_dir,
                 sign: version_signature,
                 old_current_size,
             })
@@ -9418,6 +9423,53 @@ mod test {
         assert!(!rollback_dir.is_nil());
     }
 
+    #[tokio::test]
+    async fn inline_overwrite_does_not_report_rollback_dir_for_cleanup() {
+        let dir = tempfile::tempdir().expect("temp dir should be created");
+        let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse");
+        let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created");
+        let bucket = "bucket";
+        let object = "parent";
+        let tmp_object = "tmp-write";
+        let version_id = Uuid::nil();
+
+        ensure_test_volume(&disk, bucket).await;
+        ensure_test_volume(&disk, RUSTFS_META_TMP_BUCKET).await;
+        fs::create_dir_all(dir.path().join(bucket).join(object))
+            .await
+            .expect("destination object directory should be created");
+        fs::write(
+            dir.path().join(bucket).join(object).join(STORAGE_FORMAT_FILE),
+            test_meta(test_file_info(object, version_id, None, Some(Bytes::from_static(b"old")))),
+        )
+        .await
+        .expect("old inline metadata should be written");
+        fs::create_dir_all(dir.path().join(RUSTFS_META_TMP_BUCKET).join(tmp_object))
+            .await
+            .expect("staging object directory should be created");
+
+        let response = disk
+            .rename_data(
+                RUSTFS_META_TMP_BUCKET,
+                tmp_object,
+                test_file_info(object, version_id, None, Some(Bytes::from_static(b"new"))),
+                bucket,
+                object,
+            )
+            .await
+            .expect("inline overwrite should commit");
+
+        assert_eq!(response.old_data_dir, None);
+        assert_eq!(
+            response.rollback_data_dir,
+            Some(inline_metadata_rollback_dir(version_id, &FileMeta::new()))
+        );
+        assert_eq!(
+            response.cleanup_data_dir, None,
+            "synthetic rollback state must not be recursively reclaimed"
+        );
+    }
+
     #[test]
     fn local_inline_rollback_backup_falls_back_when_hardlink_fails() {
         let dir = tempfile::tempdir().expect("temp dir should be created");

crates/ecstore/src/disk/mod.rs

@@ -1080,7 +1080,16 @@ pub enum OldCurrentSize {
 
 #[derive(Debug, Default, Serialize, Deserialize)]
 pub struct RenameDataResp {
+    /// Dereferenced erasure data directory retained for older coordinators.
     pub old_data_dir: Option<Uuid>,
+    /// Directory containing rollback metadata. This may be synthetic and must
+    /// not be used for post-commit data cleanup.
+    #[serde(default)]
+    pub rollback_data_dir: Option<Uuid>,
+    /// Dereferenced erasure data directory that is safe to reclaim after the
+    /// rename commits. Older peers omit this field, so missing means no cleanup.
+    #[serde(default)]
+    pub cleanup_data_dir: Option<Uuid>,
     pub sign: Option<Vec<u8>>,
     /// `None` means unknown — the disk could not determine the previous
     /// current version (pre-#1009 peer on the wire, or an existing dst
@@ -1477,11 +1486,15 @@ mod tests {
 
         let resp = RenameDataResp {
             old_data_dir: Some(uuid),
+            rollback_data_dir: Some(uuid),
+            cleanup_data_dir: Some(uuid),
             sign: Some(signature.clone()),
             old_current_size: Some(OldCurrentSize::Present(42)),
         };
 
         assert_eq!(resp.old_data_dir, Some(uuid));
+        assert_eq!(resp.rollback_data_dir, Some(uuid));
+        assert_eq!(resp.cleanup_data_dir, Some(uuid));
         assert_eq!(resp.sign, Some(signature));
         assert_eq!(resp.old_current_size, Some(OldCurrentSize::Present(42)));
     }
@@ -1493,6 +1506,8 @@ mod tests {
         for old_current_size in [None, Some(OldCurrentSize::Absent), Some(OldCurrentSize::Present(1337))] {
             let resp = RenameDataResp {
                 old_data_dir: Some(Uuid::new_v4()),
+                rollback_data_dir: Some(Uuid::new_v4()),
+                cleanup_data_dir: Some(Uuid::new_v4()),
                 sign: Some(vec![0x01, 0x02, 0x03]),
                 old_current_size,
             };
@@ -1501,6 +1516,8 @@ mod tests {
             let decoded: RenameDataResp = rmp_serde::decode::from_slice(&encoded).expect("named msgpack should decode");
 
             assert_eq!(decoded.old_data_dir, resp.old_data_dir);
+            assert_eq!(decoded.rollback_data_dir, resp.rollback_data_dir);
+            assert_eq!(decoded.cleanup_data_dir, resp.cleanup_data_dir);
             assert_eq!(decoded.sign, resp.sign);
             assert_eq!(decoded.old_current_size, resp.old_current_size);
         }
@@ -1526,6 +1543,8 @@ mod tests {
         let decoded: RenameDataResp = rmp_serde::decode::from_slice(&encoded).expect("legacy payload should decode");
 
         assert_eq!(decoded.old_data_dir, legacy.old_data_dir);
+        assert_eq!(decoded.rollback_data_dir, None);
+        assert_eq!(decoded.cleanup_data_dir, None);
         assert_eq!(decoded.sign, legacy.sign);
         assert_eq!(decoded.old_current_size, None);
     }

crates/ecstore/src/set_disk/core/io_primitives.rs

@@ -2837,6 +2837,7 @@ impl SetDisks {
 
         let mut disk_versions = vec![None; disks.len()];
         let mut data_dirs = vec![None; disks.len()];
+        let mut cleanup_data_dirs = vec![None; disks.len()];
         let mut old_current_sizes = vec![None; disks.len()];
 
         let results = join_all(futures).await;
@@ -2844,7 +2845,8 @@ impl SetDisks {
         for (idx, result) in results.iter().enumerate() {
             match result.as_ref().map_err(|_| DiskError::Unexpected)? {
                 Ok(res) => {
-                    data_dirs[idx] = res.old_data_dir;
+                    data_dirs[idx] = res.rollback_data_dir.or(res.old_data_dir);
+                    cleanup_data_dirs[idx] = res.cleanup_data_dir;
                     disk_versions[idx].clone_from(&res.sign);
                     old_current_sizes[idx] = res.old_current_size;
                     errs.push(None);
@@ -2948,7 +2950,7 @@ impl SetDisks {
             return Err(ret_err);
         }
 
-        let data_dir = Self::reduce_common_data_dir(&data_dirs, write_quorum);
+        let data_dir = Self::reduce_common_data_dir(&cleanup_data_dirs, write_quorum);
         let convergence = Self::classify_rename_convergence(&disk_versions, &errs);
         let old_current_size = Self::reduce_common_old_current_size(&old_current_sizes, write_quorum);
         let online_disks = Self::eval_disks(disks, &errs);
@@ -2956,7 +2958,7 @@ impl SetDisks {
             disks
                 .iter()
                 .zip(errs.iter())
-                .zip(data_dirs.iter())
+                .zip(cleanup_data_dirs.iter())
                 .map(|((disk, err), old_data_dir)| {
                     if err.is_none() && *old_data_dir == Some(data_dir) {
                         disk.clone()