diff --git a/crates/ecstore/src/cluster/rpc/remote_disk.rs b/crates/ecstore/src/cluster/rpc/remote_disk.rs index b319b2083..36b660f65 100644 --- a/crates/ecstore/src/cluster/rpc/remote_disk.rs +++ b/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; diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index bd2e9d608..bcd662cf8 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/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, Option>, Option), std::io::Error>(( + Ok::<(Option, Option, Option>, Option), 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"); diff --git a/crates/ecstore/src/disk/mod.rs b/crates/ecstore/src/disk/mod.rs index 312722310..233f2909f 100644 --- a/crates/ecstore/src/disk/mod.rs +++ b/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, + /// 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, + /// 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, pub sign: Option>, /// `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); } diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index 8cbbb5b22..b04745d10 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/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()