From b432f31c2cfb9ba9d79714948f04a2595970876f Mon Sep 17 00:00:00 2001 From: cxymds Date: Tue, 28 Jul 2026 17:04:46 +0800 Subject: [PATCH] fix(multipart): preserve retried parts on quorum failure (#5363) * fix(multipart): preserve retried parts on quorum failure * style(multipart): format transaction rollback * fix(proto): regenerate multipart transaction RPCs * fix(multipart): import rollback marker constant * fix(multipart): export transaction action * fix: import multipart transaction test requests --- .../e2e_test/src/reliant/grpc_lock_server.rs | 14 + crates/ecstore/src/api/mod.rs | 6 +- crates/ecstore/src/cluster/rpc/remote_disk.rs | 76 ++++- crates/ecstore/src/disk/disk_store.rs | 27 ++ crates/ecstore/src/disk/local.rs | 293 +++++++++++++++++- crates/ecstore/src/disk/mod.rs | 58 ++++ .../src/set_disk/core/io_primitives.rs | 228 +++++++++++++- crates/ecstore/src/set_disk/ops/multipart.rs | 237 ++++++++++++-- .../src/generated/proto_gen/node_service.rs | 134 ++++++++ crates/protos/src/lib.rs | 70 ++++- crates/protos/src/node.proto | 28 ++ rustfs/src/storage/rpc/node_service.rs | 81 ++++- rustfs/src/storage/rpc/node_service/disk.rs | 74 ++++- rustfs/src/storage/storage_api.rs | 31 +- 14 files changed, 1296 insertions(+), 61 deletions(-) diff --git a/crates/e2e_test/src/reliant/grpc_lock_server.rs b/crates/e2e_test/src/reliant/grpc_lock_server.rs index 2525be58b..301125188 100644 --- a/crates/e2e_test/src/reliant/grpc_lock_server.rs +++ b/crates/e2e_test/src/reliant/grpc_lock_server.rs @@ -400,6 +400,20 @@ impl NodeService for MinimalLockNodeService { Err(Status::unimplemented("lock-only test server")) } + async fn prepare_part_transaction( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("lock-only test server")) + } + + async fn settle_part_transaction( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("lock-only test server")) + } + async fn rename_file( &self, _request: Request, diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 61ef15198..fc042f97f 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -303,9 +303,9 @@ pub mod disk { pub use crate::disk::{ BATCH_READ_VERSION_MAX_ITEMS, BUCKET_META_PREFIX, BatchReadVersionItem, BatchReadVersionReq, BatchReadVersionResp, CheckPartsResp, DeleteOptions, Disk, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskOption, DiskStore, - FileInfoVersions, FileReader, FileWriter, HEALING_MARKER_PATH, NsScannerOpenRequest, OldCurrentSize, RUSTFS_META_BUCKET, - ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, STORAGE_FORMAT_FILE, UpdateMetadataOpts, VolumeInfo, - WalkDirOptions, new_disk, validate_batch_read_version_item_count, + FileInfoVersions, FileReader, FileWriter, HEALING_MARKER_PATH, NsScannerOpenRequest, OldCurrentSize, + PartTransactionAction, RUSTFS_META_BUCKET, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, + STORAGE_FORMAT_FILE, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, new_disk, validate_batch_read_version_item_count, }; pub use bytes::Bytes; pub use endpoint::Endpoint; diff --git a/crates/ecstore/src/cluster/rpc/remote_disk.rs b/crates/ecstore/src/cluster/rpc/remote_disk.rs index b1411a078..300f443d2 100644 --- a/crates/ecstore/src/cluster/rpc/remote_disk.rs +++ b/crates/ecstore/src/cluster/rpc/remote_disk.rs @@ -24,8 +24,8 @@ use crate::cluster::rpc::internode_data_transport::{ use crate::disk::error::{Error, Result}; use crate::disk::{ BatchReadVersionReq, BatchReadVersionResp, CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, - DiskOption, FileInfoVersions, FileReader, FileWriter, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, - UpdateMetadataOpts, VolumeInfo, WalkDirOptions, batch_read_version_one_by_one, + DiskOption, FileInfoVersions, FileReader, FileWriter, PartTransactionAction, ReadMultipleReq, ReadMultipleResp, ReadOptions, + RenameDataResp, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, batch_read_version_one_by_one, disk_store::{ DEFAULT_RUSTFS_DRIVE_ACTIVE_MONITORING, ENV_RUSTFS_DRIVE_ACTIVE_MONITORING, SKIP_IF_SUCCESS_BEFORE, get_drive_active_check_interval, get_drive_active_check_timeout, get_drive_disk_info_timeout, get_drive_list_dir_timeout, @@ -48,9 +48,10 @@ use rustfs_protos::proto_gen::node_service::RenamePartRequest; use rustfs_protos::proto_gen::node_service::{ BatchReadVersionRequest, BatchReadVersionResponse, CheckPartsRequest, DeletePathsRequest, DeleteRequest, DeleteVersionRequest, DeleteVersionsRequest, DeleteVolumeRequest, DiskInfoRequest, ListDirRequest, ListVolumesRequest, - MakeVolumeRequest, MakeVolumesRequest, ReadAllRequest, ReadMetadataRequest, ReadMultipleRequest, ReadMultipleResponse, - ReadPartsRequest, ReadVersionRequest, ReadXlRequest, RenameDataRequest, RenameFileRequest, StatVolumeRequest, - UpdateMetadataRequest, VerifyFileRequest, WriteAllRequest, WriteMetadataRequest, node_service_client::NodeServiceClient, + MakeVolumeRequest, MakeVolumesRequest, PreparePartTransactionRequest, ReadAllRequest, ReadMetadataRequest, + ReadMultipleRequest, ReadMultipleResponse, ReadPartsRequest, ReadVersionRequest, ReadXlRequest, RenameDataRequest, + RenameFileRequest, SettlePartTransactionRequest, StatVolumeRequest, UpdateMetadataRequest, VerifyFileRequest, + WriteAllRequest, WriteMetadataRequest, node_service_client::NodeServiceClient, }; use serde::{Serialize, de::DeserializeOwned}; use std::{ @@ -2481,6 +2482,71 @@ impl DiskAPI for RemoteDisk { .await } + #[tracing::instrument(level = "trace", skip_all)] + async fn prepare_part_transaction( + &self, + src_volume: &str, + src_path: &str, + dst_volume: &str, + dst_path: &str, + meta: Bytes, + ) -> Result<()> { + self.execute_with_timeout( + || async { + let mut client = self + .get_client() + .await + .map_err(|err| Error::other(format!("can not get client, err: {err}")))?; + let mut request = Request::new(PreparePartTransactionRequest { + disk: self.endpoint.to_string(), + src_volume: src_volume.to_string(), + src_path: src_path.to_string(), + dst_volume: dst_volume.to_string(), + dst_path: dst_path.to_string(), + meta, + }); + let canonical_body = rustfs_protos::canonical_prepare_part_transaction_request_body(request.get_ref()); + attach_mutation_body_digest(&mut request, canonical_body, "prepare_part_transaction")?; + + let response = client.prepare_part_transaction(request).await?.into_inner(); + if !response.success { + return Err(response.error.unwrap_or_default().into()); + } + Ok(()) + }, + get_max_timeout_duration(), + ) + .await + } + + #[tracing::instrument(level = "trace", skip_all)] + async fn settle_part_transaction(&self, volume: &str, path: &str, action: PartTransactionAction) -> Result<()> { + self.execute_with_timeout( + || async { + let mut client = self + .get_client() + .await + .map_err(|err| Error::other(format!("can not get client, err: {err}")))?; + let mut request = Request::new(SettlePartTransactionRequest { + disk: self.endpoint.to_string(), + volume: volume.to_string(), + path: path.to_string(), + rollback: action == PartTransactionAction::Rollback, + }); + let canonical_body = rustfs_protos::canonical_settle_part_transaction_request_body(request.get_ref()); + attach_mutation_body_digest(&mut request, canonical_body, "settle_part_transaction")?; + + let response = client.settle_part_transaction(request).await?.into_inner(); + if !response.success { + return Err(response.error.unwrap_or_default().into()); + } + Ok(()) + }, + get_max_timeout_duration(), + ) + .await + } + #[tracing::instrument(level = "trace", skip_all)] async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()> { trace!( diff --git a/crates/ecstore/src/disk/disk_store.rs b/crates/ecstore/src/disk/disk_store.rs index 2ae0ed8d1..c117672c8 100644 --- a/crates/ecstore/src/disk/disk_store.rs +++ b/crates/ecstore/src/disk/disk_store.rs @@ -1473,6 +1473,33 @@ impl DiskAPI for LocalDiskWrapper { .await } + async fn prepare_part_transaction( + &self, + src_volume: &str, + src_path: &str, + dst_volume: &str, + dst_path: &str, + meta: Bytes, + ) -> Result<()> { + self.track_disk_health( + || async { + self.disk + .prepare_part_transaction(src_volume, src_path, dst_volume, dst_path, meta) + .await + }, + get_max_timeout_duration(), + ) + .await + } + + async fn settle_part_transaction(&self, volume: &str, path: &str, action: crate::disk::PartTransactionAction) -> Result<()> { + self.track_disk_health( + || async { self.disk.settle_part_transaction(volume, path, action).await }, + get_max_timeout_duration(), + ) + .await + } + async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()> { self.track_disk_health(|| async { self.disk.delete(volume, path, opt).await }, get_max_timeout_duration()) .await diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index d6afb4c32..37ae245b2 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -19,7 +19,8 @@ use crate::disk::disk_store::{get_drive_walkdir_stall_timeout, get_object_disk_r use crate::disk::{ BUCKET_META_PREFIX, CHECK_PART_FILE_CORRUPT, CHECK_PART_FILE_NOT_FOUND, CHECK_PART_SUCCESS, CHECK_PART_UNKNOWN, CHECK_PART_VOLUME_NOT_FOUND, CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskMetrics, - FileInfoVersions, FileReader, FileWriter, MmapCopyStageMetrics, OldCurrentSize, RUSTFS_META_BUCKET, RUSTFS_META_TMP_BUCKET, + FileInfoVersions, FileReader, FileWriter, MmapCopyStageMetrics, OldCurrentSize, PART_TRANSACTION_NEW_META, + PART_TRANSACTION_OLD_META, PART_TRANSACTION_ROLLBACK, PartTransactionAction, RUSTFS_META_BUCKET, RUSTFS_META_TMP_BUCKET, RUSTFS_META_TMP_DELETED_BUCKET, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, STORAGE_FORMAT_FILE, STORAGE_FORMAT_FILE_BACKUP, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, conv_part_err_to_int, endpoint::Endpoint, @@ -80,6 +81,10 @@ const ENV_BITROT_SIZE_MISMATCH_RETRY_COUNT: &str = "RUSTFS_BITROT_SIZE_MISMATCH_ const ENV_BITROT_SIZE_MISMATCH_RETRY_DELAY_MS: &str = "RUSTFS_BITROT_SIZE_MISMATCH_RETRY_DELAY_MS"; const DEFAULT_BITROT_SIZE_MISMATCH_RETRY_COUNT: u64 = 2; const DEFAULT_BITROT_SIZE_MISMATCH_RETRY_DELAY_MS: u64 = 100; +const PART_TRANSACTION_OLD_DATA: &str = "old.data"; +const PART_TRANSACTION_OLD_DATA_ABSENT: &str = "old.data.absent"; +const PART_TRANSACTION_OLD_META_ABSENT: &str = "old.meta.absent"; +const PART_TRANSACTION_PUBLISH_META: &str = "publish.meta"; enum ReadAllError { Open(std::io::Error), Disk(DiskError), @@ -141,6 +146,28 @@ fn remove_dir_all_if_exists(path: &Path) -> std::io::Result<()> { } } +fn snapshot_part_transaction_file(src: &Path, backup: &Path, absent: &Path) -> std::io::Result<()> { + match std::fs::symlink_metadata(src) { + Ok(metadata) if metadata.is_file() => std::fs::hard_link(src, backup), + Ok(_) => Err(std::io::Error::new(ErrorKind::InvalidData, "multipart transaction source is not a file")), + Err(err) if err.kind() == ErrorKind::NotFound => std::fs::write(absent, []), + Err(err) => Err(err), + } +} + +fn restore_part_transaction_file(current: &Path, backup: &Path, absent: &Path, restore: &Path) -> std::io::Result<()> { + match std::fs::symlink_metadata(backup) { + Ok(metadata) if metadata.is_file() => { + remove_file_if_exists(restore)?; + std::fs::hard_link(backup, restore)?; + std::fs::rename(restore, current) + } + Ok(_) => Err(std::io::Error::new(ErrorKind::InvalidData, "multipart transaction backup is not a file")), + Err(err) if err.kind() == ErrorKind::NotFound && absent.is_file() => remove_file_if_exists(current), + Err(err) => Err(err), + } +} + fn rollback_committed_rename_std( dst_file_path: &Path, new_data_path: Option<&Path>, @@ -6447,6 +6474,151 @@ impl DiskAPI for LocalDisk { Ok(resp) } + #[tracing::instrument(level = "trace", skip_all)] + async fn prepare_part_transaction( + &self, + src_volume: &str, + src_path: &str, + dst_volume: &str, + dst_path: &str, + meta: Bytes, + ) -> Result<()> { + let src_volume_dir = self.get_bucket_path(src_volume)?; + let dst_volume_dir = self.get_bucket_path(dst_volume)?; + if !skip_access_checks(src_volume) { + super::fs::access_std(&src_volume_dir).map_err(|err| to_access_error(err, DiskError::VolumeAccessDenied))?; + } + if !skip_access_checks(dst_volume) { + super::fs::access_std(&dst_volume_dir).map_err(|err| to_access_error(err, DiskError::VolumeAccessDenied))?; + } + + let src_file_path = self.get_object_path(src_volume, src_path)?; + let dst_file_path = self.get_object_path(dst_volume, dst_path)?; + let dst_meta_path = self.get_object_path(dst_volume, &format!("{dst_path}.meta"))?; + let transaction_path = self.get_object_path(dst_volume, &crate::disk::part_transaction_path(dst_path))?; + for path in [&src_file_path, &dst_file_path, &dst_meta_path, &transaction_path] { + check_path_length(path.to_string_lossy().as_ref())?; + } + + let durability = effective_durability(dst_volume); + tokio::task::spawn_blocking(move || { + let source = std::fs::symlink_metadata(&src_file_path).map_err(to_file_error)?; + if !source.is_file() { + return Err(DiskError::FileAccessDenied); + } + if transaction_path.exists() { + return Err(DiskError::FileAccessDenied); + } + + let Some(transaction_parent) = transaction_path.parent() else { + return Err(DiskError::InvalidPath); + }; + std::fs::create_dir_all(transaction_parent).map_err(to_file_error)?; + let staging_path = transaction_parent.join(format!(".part-txn-{}", Uuid::new_v4())); + std::fs::create_dir(&staging_path).map_err(to_file_error)?; + + let prepare_result = (|| -> std::io::Result<()> { + snapshot_part_transaction_file( + &dst_file_path, + &staging_path.join(PART_TRANSACTION_OLD_DATA), + &staging_path.join(PART_TRANSACTION_OLD_DATA_ABSENT), + )?; + snapshot_part_transaction_file( + &dst_meta_path, + &staging_path.join(PART_TRANSACTION_OLD_META), + &staging_path.join(PART_TRANSACTION_OLD_META_ABSENT), + )?; + + let mut new_meta = std::fs::OpenOptions::new() + .create_new(true) + .write(true) + .open(staging_path.join(PART_TRANSACTION_NEW_META))?; + std::io::Write::write_all(&mut new_meta, &meta)?; + if durability.syncs_commit_metadata() { + new_meta.sync_data()?; + os::fsync_dir_std(&staging_path)?; + } + std::fs::rename(&staging_path, &transaction_path)?; + if durability.syncs_commit_metadata() { + os::fsync_dir_std(transaction_parent)?; + } + Ok(()) + })(); + + if let Err(err) = prepare_result { + let _ = remove_dir_all_if_exists(&staging_path); + return Err(to_file_error(err).into()); + } + + Ok(()) + }) + .await + .map_err(DiskError::from)? + } + + #[tracing::instrument(level = "trace", skip_all)] + async fn settle_part_transaction(&self, volume: &str, path: &str, action: PartTransactionAction) -> Result<()> { + self.get_bucket_path(volume)?; + let current_data_path = self.get_object_path(volume, path)?; + let current_meta_path = self.get_object_path(volume, &format!("{path}.meta"))?; + let transaction_path = self.get_object_path(volume, &crate::disk::part_transaction_path(path))?; + for candidate in [¤t_data_path, ¤t_meta_path, &transaction_path] { + check_path_length(candidate.to_string_lossy().as_ref())?; + } + let durability = effective_durability(volume); + + tokio::task::spawn_blocking(move || { + match std::fs::symlink_metadata(&transaction_path) { + Ok(metadata) if metadata.is_dir() => {} + Ok(_) => return Err(DiskError::FileCorrupt), + Err(err) if err.kind() == ErrorKind::NotFound => return Ok(()), + Err(err) => return Err(to_file_error(err).into()), + } + + if action == PartTransactionAction::Rollback { + std::fs::write(transaction_path.join(PART_TRANSACTION_ROLLBACK), []).map_err(to_file_error)?; + if durability.syncs_commit_metadata() { + os::fsync_dir_std(&transaction_path).map_err(to_file_error)?; + } + restore_part_transaction_file( + ¤t_data_path, + &transaction_path.join(PART_TRANSACTION_OLD_DATA), + &transaction_path.join(PART_TRANSACTION_OLD_DATA_ABSENT), + &transaction_path.join("restore.data"), + ) + .map_err(to_file_error)?; + restore_part_transaction_file( + ¤t_meta_path, + &transaction_path.join(PART_TRANSACTION_OLD_META), + &transaction_path.join(PART_TRANSACTION_OLD_META_ABSENT), + &transaction_path.join("restore.meta"), + ) + .map_err(to_file_error)?; + if durability.syncs_commit_metadata() + && let Some(parent) = current_data_path.parent() + { + os::fsync_dir_std(parent).map_err(to_file_error)?; + } + } + + let Some(parent) = transaction_path.parent() else { + return Err(DiskError::InvalidPath); + }; + let cleanup_path = parent.join(format!(".part-txn-settled-{}", Uuid::new_v4())); + std::fs::rename(&transaction_path, &cleanup_path).map_err(to_file_error)?; + if durability.syncs_commit_metadata() { + os::fsync_dir_std(parent).map_err(to_file_error)?; + } + remove_dir_all_if_exists(&cleanup_path).map_err(to_file_error)?; + Ok(()) + }) + .await + .map_err(DiskError::from)??; + + self.io_backend.invalidate_cached_fd(volume, path).await; + Ok(()) + } + #[tracing::instrument(level = "trace", skip_all)] async fn rename_part(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str, meta: Bytes) -> Result<()> { let src_volume_dir = self.get_bucket_path(src_volume)?; @@ -6528,6 +6700,42 @@ impl DiskAPI for LocalDisk { } } + let transaction_publish_meta = if src_is_dir { + None + } else { + let transaction_path = self.get_object_path(dst_volume, &crate::disk::part_transaction_path(dst_path))?; + let transaction_meta_path = transaction_path.join(PART_TRANSACTION_NEW_META); + match fs::read(&transaction_meta_path).await { + Ok(expected_meta) => { + if expected_meta.as_slice() != meta.as_ref() { + return Err(DiskError::FileCorrupt); + } + + let publish_meta_path = transaction_path.join(PART_TRANSACTION_PUBLISH_META); + let source_meta_path = transaction_meta_path.clone(); + let publish_path = publish_meta_path.clone(); + tokio::task::spawn_blocking(move || { + remove_file_if_exists(&publish_path)?; + std::fs::hard_link(source_meta_path, &publish_path) + }) + .await + .map_err(DiskError::from)? + .map_err(to_file_error)?; + Some(publish_meta_path) + } + // Old peers know only RenamePart. The new coordinator never + // reaches rename unless prepare succeeded, so an absent + // transaction directory identifies the rolling-upgrade legacy + // path. A present directory without new.meta is corruption. + Err(err) if err.kind() == ErrorKind::NotFound => match fs::metadata(&transaction_path).await { + Ok(_) => return Err(DiskError::FileCorrupt), + Err(meta_err) if meta_err.kind() == ErrorKind::NotFound => None, + Err(meta_err) => return Err(to_file_error(meta_err).into()), + }, + Err(err) => return Err(to_file_error(err).into()), + } + }; + // UploadPart is acknowledged once this rename lands, so the part data and // its directory entry must be durable before we return. Relaxed keeps the // part payload fdatasync but leaves the directory entry to the page cache. @@ -6561,7 +6769,17 @@ impl DiskAPI for LocalDisk { return Err(DiskError::FileAccessDenied); } - self.write_all(dst_volume, format!("{dst_path}.meta").as_str(), meta).await?; + if let Some(transaction_publish_meta) = transaction_publish_meta { + let dst_meta_path = self.get_object_path(dst_volume, &format!("{dst_path}.meta"))?; + rename_all(&transaction_publish_meta, &dst_meta_path, &dst_volume_dir).await?; + if durability.syncs_commit_metadata() + && let Some(parent) = dst_meta_path.parent() + { + os::fsync_dir(parent).await.map_err(to_file_error)?; + } + } else { + self.write_all(dst_volume, format!("{dst_path}.meta").as_str(), meta).await?; + } if let Some(parent) = src_file_path.parent() { self.delete_file(&src_volume_dir, &parent.to_path_buf(), false, false).await?; @@ -9370,9 +9588,15 @@ mod test { .await .expect("source part should be written"); + disk.prepare_part_transaction(tmp_volume, "upload/part.1", bucket, "object/part.1", meta.clone()) + .await + .expect("part transaction should be prepared"); disk.rename_part(tmp_volume, "upload/part.1", bucket, "object/part.1", meta.clone()) .await .expect("rename_part should commit part"); + disk.settle_part_transaction(bucket, "object/part.1", PartTransactionAction::Commit) + .await + .expect("part transaction should be committed"); assert_eq!( disk.read_all(bucket, "object/part.1") @@ -9390,6 +9614,71 @@ mod test { matches!(disk.read_all(tmp_volume, "upload/part.1").await, Err(DiskError::FileNotFound)), "source part must be removed after a successful commit" ); + + let legacy_payload = Bytes::from_static(b"legacy peer payload"); + let legacy_meta = Bytes::from_static(b"legacy peer metadata"); + disk.write_all(tmp_volume, "legacy/part.1", legacy_payload.clone()) + .await + .expect("legacy source part should be written"); + disk.rename_part(tmp_volume, "legacy/part.1", bucket, "object/part.1", legacy_meta.clone()) + .await + .expect("pre-transaction peer RenamePart should remain supported"); + assert_eq!( + disk.read_all(bucket, "object/part.1") + .await + .expect("legacy destination part should be readable"), + legacy_payload + ); + assert_eq!( + disk.read_all(bucket, "object/part.1.meta") + .await + .expect("legacy destination metadata should be readable"), + legacy_meta + ); + } + + #[tokio::test] + async fn test_part_transaction_rolls_back_data_published_before_metadata() { + use tempfile::tempdir; + + let dir = 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"); + ensure_test_volume(&disk, "tmp").await; + ensure_test_volume(&disk, "bucket").await; + + disk.write_all("tmp", "upload/part.1", Bytes::from_static(b"new data")) + .await + .expect("new part should be staged"); + disk.write_all("bucket", "object/part.1", Bytes::from_static(b"old data")) + .await + .expect("old part data should be staged"); + disk.write_all("bucket", "object/part.1.meta", Bytes::from_static(b"old metadata")) + .await + .expect("old part metadata should be staged"); + + disk.prepare_part_transaction("tmp", "upload/part.1", "bucket", "object/part.1", Bytes::from_static(b"new metadata")) + .await + .expect("part transaction should be prepared"); + disk.rename_file("tmp", "upload/part.1", "bucket", "object/part.1") + .await + .expect("data publication should succeed"); + disk.settle_part_transaction("bucket", "object/part.1", PartTransactionAction::Rollback) + .await + .expect("part transaction should roll back"); + + assert_eq!( + disk.read_all("bucket", "object/part.1") + .await + .expect("old part data should be restored"), + Bytes::from_static(b"old data") + ); + assert_eq!( + disk.read_all("bucket", "object/part.1.meta") + .await + .expect("old part metadata should be restored"), + Bytes::from_static(b"old metadata") + ); } struct BlockingScanWriter { diff --git a/crates/ecstore/src/disk/mod.rs b/crates/ecstore/src/disk/mod.rs index d683e860f..fc17998e8 100644 --- a/crates/ecstore/src/disk/mod.rs +++ b/crates/ecstore/src/disk/mod.rs @@ -38,6 +38,16 @@ pub const FORMAT_CONFIG_FILE: &str = "format.json"; pub const HEALING_MARKER_PATH: &str = "healing.bin"; pub const STORAGE_FORMAT_FILE: &str = "xl.meta"; pub const STORAGE_FORMAT_FILE_BACKUP: &str = "xl.meta.bkp"; +pub const PART_TRANSACTION_NEW_META: &str = "new.meta"; +pub const PART_TRANSACTION_OLD_META: &str = "old.meta"; +pub const PART_TRANSACTION_ROLLBACK: &str = "rollback"; + +pub fn part_transaction_path(part_path: &str) -> String { + match part_path.rsplit_once('/') { + Some((parent, name)) => format!("{parent}/.{name}.rustfs-txn"), + None => format!(".{part_path}.rustfs-txn"), + } +} use crate::cluster::rpc::RemoteDisk; use crate::cluster::rpc::build_internode_data_transport_from_env; @@ -62,6 +72,12 @@ pub type DiskStore = Arc; pub type FileReader = Box; pub type FileWriter = Box; +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum PartTransactionAction { + Commit, + Rollback, +} + #[derive(Clone, Copy, Debug)] pub struct MmapCopyStageMetrics { pub(crate) path: &'static str, @@ -381,6 +397,35 @@ impl DiskAPI for Disk { } } + async fn prepare_part_transaction( + &self, + src_volume: &str, + src_path: &str, + dst_volume: &str, + dst_path: &str, + meta: Bytes, + ) -> Result<()> { + match self { + Disk::Local(local_disk) => { + local_disk + .prepare_part_transaction(src_volume, src_path, dst_volume, dst_path, meta) + .await + } + Disk::Remote(remote_disk) => { + remote_disk + .prepare_part_transaction(src_volume, src_path, dst_volume, dst_path, meta) + .await + } + } + } + + async fn settle_part_transaction(&self, volume: &str, path: &str, action: PartTransactionAction) -> Result<()> { + match self { + Disk::Local(local_disk) => local_disk.settle_part_transaction(volume, path, action).await, + Disk::Remote(remote_disk) => remote_disk.settle_part_transaction(volume, path, action).await, + } + } + #[tracing::instrument(level = "trace", skip_all)] async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()> { match self { @@ -659,6 +704,19 @@ pub trait DiskAPI: Debug + Send + Sync + 'static { // ReadFileStream async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()>; async fn rename_part(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str, meta: Bytes) -> Result<()>; + async fn prepare_part_transaction( + &self, + _src_volume: &str, + _src_path: &str, + _dst_volume: &str, + _dst_path: &str, + _meta: Bytes, + ) -> Result<()> { + Err(DiskError::MethodNotAllowed) + } + async fn settle_part_transaction(&self, _volume: &str, _path: &str, _action: PartTransactionAction) -> Result<()> { + Err(DiskError::MethodNotAllowed) + } async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()>; // VerifyFile async fn verify_file(&self, volume: &str, path: &str, fi: &FileInfo) -> Result; diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index 9d07d8310..dd5a9b5b5 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -46,7 +46,10 @@ use crate::diagnostics::get::{ GetObjectFailureReason, classify_disk_error, get_stage_timer_if_enabled, record_get_object_pipeline_failure, record_get_object_pipeline_failure_for_path, record_get_stage_duration_if_enabled, }; -use crate::disk::OldCurrentSize; +use crate::disk::{ + OldCurrentSize, PART_TRANSACTION_NEW_META, PART_TRANSACTION_OLD_META, PART_TRANSACTION_ROLLBACK, PartTransactionAction, + part_transaction_path, +}; use crate::erasure::coding::BitrotReader; use crate::io_support::bitrot::ShardReader; use crate::io_support::bitrot::{ @@ -3174,6 +3177,155 @@ impl SetDisks { } } + async fn recover_part_transaction(&self, dst_object: &str, write_quorum: usize) -> disk::error::Result { + let disks = self.get_disks_internal().await; + let transaction_path = part_transaction_path(dst_object); + let transaction_meta_path = format!("{transaction_path}/{PART_TRANSACTION_NEW_META}"); + let rollback_path = format!("{transaction_path}/{PART_TRANSACTION_ROLLBACK}"); + let current_meta_path = format!("{dst_object}.meta"); + + let reads = disks.iter().map(|disk| { + let disk = disk.clone(); + let transaction_meta_path = transaction_meta_path.clone(); + let rollback_path = rollback_path.clone(); + let current_meta_path = current_meta_path.clone(); + async move { + let Some(disk) = disk else { + return Ok((None, None, false)); + }; + let transaction_meta = match disk.read_all(RUSTFS_META_MULTIPART_BUCKET, &transaction_meta_path).await { + Ok(meta) => Some(meta), + Err(DiskError::FileNotFound) => None, + Err(err) => return Err(err), + }; + let rollback = match disk.read_all(RUSTFS_META_MULTIPART_BUCKET, &rollback_path).await { + Ok(_) => true, + Err(DiskError::FileNotFound) => false, + Err(err) => return Err(err), + }; + let current_meta = match disk.read_all(RUSTFS_META_MULTIPART_BUCKET, ¤t_meta_path).await { + Ok(meta) => Some(meta), + Err(DiskError::FileNotFound | DiskError::DiskNotFound) => None, + Err(_) => None, + }; + Ok((transaction_meta, current_meta, rollback)) + } + }); + let observations = join_all(reads).await.into_iter().collect::>>()?; + if observations.iter().all(|(transaction, _, _)| transaction.is_none()) { + return Ok(false); + } + + let mut current_counts: HashMap = HashMap::new(); + for (_, current, _) in &observations { + if let Some(current) = current { + *current_counts.entry(current.clone()).or_default() += 1; + } + } + let current_quorum = current_counts + .into_iter() + .find_map(|(meta, count)| (count >= write_quorum).then_some(meta)); + + let old_meta_path = format!("{transaction_path}/{PART_TRANSACTION_OLD_META}"); + let old_meta_absent_path = format!("{transaction_path}/old.meta.absent"); + let decisions = observations + .iter() + .enumerate() + .filter_map(|(index, (transaction_meta, _, rollback))| { + transaction_meta.as_ref().map(|meta| (index, meta.clone(), *rollback)) + }) + .map(|(index, transaction_meta, rollback)| { + let disk = disks[index].clone(); + let old_meta_path = old_meta_path.clone(); + let old_meta_absent_path = old_meta_absent_path.clone(); + let current_quorum = current_quorum.clone(); + async move { + let Some(disk) = disk else { + return Err(DiskError::DiskNotFound); + }; + let action = if rollback { + PartTransactionAction::Rollback + } else if current_quorum.as_ref() == Some(&transaction_meta) { + PartTransactionAction::Commit + } else if let Some(current_quorum) = current_quorum { + match disk.read_all(RUSTFS_META_MULTIPART_BUCKET, &old_meta_path).await { + Ok(old_meta) if old_meta == current_quorum => PartTransactionAction::Rollback, + Ok(_) => PartTransactionAction::Commit, + Err(DiskError::FileNotFound) => { + match disk.read_all(RUSTFS_META_MULTIPART_BUCKET, &old_meta_absent_path).await { + Ok(_) => PartTransactionAction::Commit, + Err(_) => return Err(DiskError::FileCorrupt), + } + } + Err(err) => return Err(err), + } + } else { + PartTransactionAction::Rollback + }; + disk.settle_part_transaction(RUSTFS_META_MULTIPART_BUCKET, dst_object, action) + .await?; + Ok(action == PartTransactionAction::Commit) + } + }); + + let results = join_all(decisions).await; + if let Some(err) = results.iter().find_map(|result| result.as_ref().err()) { + return Err(err.clone()); + } + Ok(results.iter().any(|result| matches!(result, Ok(true)))) + } + + pub(in crate::set_disk) async fn recover_part_transactions( + &self, + part_path: &str, + read_quorum: usize, + write_quorum: usize, + ) -> disk::error::Result<()> { + let disks = self.get_disks_internal().await; + let listings = join_all(disks.iter().map(|disk| { + let disk = disk.clone(); + async move { + let Some(disk) = disk else { + return Err(DiskError::DiskNotFound); + }; + disk.list_dir(RUSTFS_META_MULTIPART_BUCKET, RUSTFS_META_MULTIPART_BUCKET, part_path, -1) + .await + } + })) + .await; + + let mut transaction_parts = HashSet::new(); + let mut errs = Vec::with_capacity(listings.len()); + for listing in listings { + match listing { + Ok(entries) => { + errs.push(None); + for entry in entries { + let name = entry.trim_end_matches('/'); + let Some(part_number) = name + .strip_prefix(".part.") + .and_then(|name| name.strip_suffix(".rustfs-txn")) + .and_then(|number| number.parse::().ok()) + else { + continue; + }; + transaction_parts.insert(part_number); + } + } + Err(err) => errs.push(Some(err)), + } + } + if let Some(err) = reduce_read_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, read_quorum) { + return Err(err); + } + + for part_number in transaction_parts { + self.recover_part_transaction(&format!("{part_path}part.{part_number}"), write_quorum) + .await?; + } + Ok(()) + } + #[tracing::instrument(skip(disks, meta))] #[allow(clippy::too_many_arguments)] pub(in crate::set_disk) async fn rename_part( @@ -3187,23 +3339,46 @@ impl SetDisks { write_quorum: usize, quorum_context: Option>, ) -> disk::error::Result>> { + self.recover_part_transaction(dst_object, write_quorum).await?; + let src_bucket = Arc::new(src_bucket.to_string()); let src_object = Arc::new(src_object.to_string()); let dst_bucket = Arc::new(dst_bucket.to_string()); let dst_object = Arc::new(dst_object.to_string()); - // Do NOT pre-delete the destination part before renaming: the per-disk - // `rename_part` replaces `part.N` atomically (std::fs::rename) and rewrites - // `part.N.meta`, so the pre-delete is redundant — and destructive. It - // opened a window where an already-committed (ACKed) part was removed on - // every disk before the new rename landed, so a re-upload that then failed - // quorum destroyed the committed part outright (backlog#853 / #799 B4). - // The atomic rename overwrites in place; on quorum failure below we roll - // the destination back. + let prepare_tasks = disks.iter().map(|disk| { + let disk = disk.clone(); + let src_bucket = src_bucket.clone(); + let src_object = src_object.clone(); + let dst_bucket = dst_bucket.clone(); + let dst_object = dst_object.clone(); + let meta = meta.clone(); + async move { + let disk = disk?; + Some( + disk.prepare_part_transaction(&src_bucket, &src_object, &dst_bucket, &dst_object, meta) + .await, + ) + } + }); + let prepare_results = join_all(prepare_tasks).await; + let prepare_errs = prepare_results + .into_iter() + .map(|result| match result { + Some(Ok(())) => None, + Some(Err(err)) => Some(err), + None => Some(DiskError::DiskNotFound), + }) + .collect::>(); + let prepared_disks = Self::eval_disks(disks, &prepare_errs); + if reduce_write_quorum_errs(&prepare_errs, OBJECT_OP_IGNORED_ERRS, write_quorum).is_some() { + self.recover_part_transaction(&dst_object, write_quorum).await?; + return Err(DiskError::ErasureWriteQuorum); + } let mut errs = Vec::with_capacity(disks.len()); - let futures = disks.iter().map(|disk| { + let futures = prepared_disks.iter().map(|disk| { let disk = disk.clone(); let meta = meta.clone(); let src_bucket = src_bucket.clone(); @@ -3253,19 +3428,42 @@ impl SetDisks { ); } - if let Some(err) = reduce_write_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, write_quorum) { + let reduced_err = reduce_write_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, write_quorum); + if let Some(err) = reduced_err { + let rollbacks = prepared_disks.iter().filter_map(|disk| { + disk.clone().map(|disk| { + let dst_object = dst_object.clone(); + async move { + disk.settle_part_transaction(RUSTFS_META_MULTIPART_BUCKET, &dst_object, PartTransactionAction::Rollback) + .await + } + }) + }); + let rollback_results = join_all(rollbacks).await; + self.recover_part_transaction(&dst_object, write_quorum).await?; + if let Some(rollback_err) = rollback_results.iter().find_map(|result| result.as_ref().err()) { + warn!(error = %rollback_err, "rename_part rollback did not settle on every prepared disk"); + } if let Some(context) = quorum_context { log_multipart_write_quorum_failure(context, &errs, write_quorum, &err); } else { warn!("rename_part errs {:?}", &errs); } - self.cleanup_multipart_path(&[dst_object.to_string(), format!("{dst_object}.meta")]) - .await; return Err(err); } - let disks = Self::eval_disks(disks, &errs); - Ok(disks) + let committed = self.recover_part_transaction(&dst_object, write_quorum).await?; + if !committed { + let err = DiskError::ErasureWriteQuorum; + if let Some(context) = quorum_context { + log_multipart_write_quorum_failure(context, &errs, write_quorum, &err); + } else { + warn!("rename_part errs {:?}", &errs); + } + return Err(err); + } + + Ok(Self::eval_disks(&prepared_disks, &errs)) } pub(in crate::set_disk) fn eval_disks(disks: &[Option], errs: &[Option]) -> Vec> { diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs index a3228e079..fd3416fa8 100644 --- a/crates/ecstore/src/set_disk/ops/multipart.rs +++ b/crates/ecstore/src/set_disk/ops/multipart.rs @@ -1289,8 +1289,18 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { } } + if !opts.no_lock && object_lock_guard.is_none() { + object_lock_guard = Some( + self.acquire_write_lock_diag("complete_multipart_upload_commit", bucket, object) + .await?, + ); + } + let upload_guard = self + .acquire_multipart_upload_write_lock("complete_multipart_upload_commit", bucket, object, upload_id, opts) + .await?; + let expected_restore_operation_id = restore_commit_operation_id_from_metadata(&opts.user_defined)?; - let (mut fi, mut files_metas) = self.check_upload_id_exists(bucket, object, upload_id, true).await?; + let (mut fi, files_metas) = self.check_upload_id_exists(bucket, object, upload_id, true).await?; let has_layout_candidate = range_seek_rollout_enabled && fi .data_dir @@ -1303,22 +1313,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { &token, ) }); - let upload_guard = if has_layout_candidate { - if object_lock_guard.is_none() { - object_lock_guard = Some( - self.acquire_write_lock_diag("complete_multipart_upload_commit", bucket, object) - .await?, - ); - } - let guard = self - .acquire_multipart_upload_write_lock("complete_multipart_upload_commit", bucket, object, upload_id, opts) - .await?; - (fi, files_metas) = self.check_upload_id_exists(bucket, object, upload_id, true).await?; - guard - } else { - None - }; - let quorum_validated_layout_token = upload_guard.as_ref().and_then(|_| { + let quorum_validated_layout_token = if has_layout_candidate { fi.data_dir .filter(|data_dir| !data_dir.is_nil()) .map(|data_dir| data_dir.to_string()) @@ -1329,7 +1324,9 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { token, ) }) - }); + } else { + None + }; rustfs_utils::http::metadata_compat::remove_str( &mut fi.metadata, crate::object_api::ENCRYPTED_PART_LAYOUT_CANDIDATE_SUFFIX, @@ -1356,6 +1353,9 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { // let disks = Self::shuffle_disks(&disks, &fi.erasure.distribution); let part_path = format!("{}/{}/", upload_id_path, fi.data_dir.unwrap_or(Uuid::nil())); + self.recover_part_transactions(&part_path, read_quorum, write_quorum) + .await + .map_err(|err| to_object_err(err.into(), vec![bucket, object]))?; let part_meta_paths = uploaded_parts .iter() @@ -1713,12 +1713,6 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { } } - if !opts.no_lock && object_lock_guard.is_none() { - object_lock_guard = Some( - self.acquire_write_lock_diag("complete_multipart_upload_commit", bucket, object) - .await?, - ); - } // Phase 2 (backlog#899): fence the commit on lock loss before any destructive // step. If the refresh heartbeat has observed a refresh-quorum loss, another // writer may have re-acquired this object's lock; proceeding would race a @@ -2183,6 +2177,201 @@ mod tests { ) } + async fn assert_quorum_minus_one_retry_preserves_completable_part( + disk_count: usize, + parity: usize, + success_indices: &[usize], + ) { + use tokio::io::AsyncReadExt as _; + + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks_for_pool_with_default_parity(disk_count, 0, parity).await; + let bucket = format!("multipart-retry-{disk_count}-{}", success_indices[0]); + let object = "object"; + make_bucket_on_all(&disk_stores, &bucket).await; + + let payload = vec![0x41; 1 << 20]; + let (upload_id, parts) = + stage_upload_with_create_opts(&set_disks, &bucket, object, &payload, &ObjectOptions::default()).await; + let (upload_meta, _) = set_disks + .check_upload_id_exists(&bucket, object, &upload_id, false) + .await + .expect("staged upload metadata should be readable"); + let upload_path = SetDisks::get_upload_id_dir(&bucket, object, &upload_id); + let write_quorum = upload_meta.write_quorum(set_disks.default_write_quorum()); + assert_eq!(success_indices.len() + 1, write_quorum); + let part_path = format!( + "{}/{}/part.1", + upload_path, + upload_meta.data_dir.expect("multipart upload should have a data directory") + ); + let acknowledged_meta = disk_stores[0] + .read_all(RUSTFS_META_MULTIPART_BUCKET, &format!("{part_path}.meta")) + .await + .expect("acknowledged part metadata should be readable"); + let retry_path = format!("{}/part.1", Uuid::new_v4()); + + let mut retry_disks = vec![None; disk_stores.len()]; + for &index in success_indices { + disk_stores[index] + .write_all(RUSTFS_META_TMP_BUCKET, &retry_path, Bytes::from_static(b"retry shard")) + .await + .expect("retry shard should be staged"); + retry_disks[index] = Some(disk_stores[index].clone()); + } + + let err = set_disks + .rename_part( + &retry_disks, + RUSTFS_META_TMP_BUCKET, + &retry_path, + RUSTFS_META_MULTIPART_BUCKET, + &part_path, + acknowledged_meta, + write_quorum, + None, + ) + .await + .expect_err("quorum-minus-one renamed shards must remain below write quorum"); + assert_eq!(err, DiskError::ErasureWriteQuorum); + + for (index, disk) in disk_stores.iter().enumerate() { + assert!( + disk.read_all(RUSTFS_META_MULTIPART_BUCKET, &part_path).await.is_ok(), + "old acknowledged shard on disk {index} must survive" + ); + } + + set_disks + .clone() + .complete_multipart_upload(&bucket, object, &upload_id, parts, &ObjectOptions::default()) + .await + .expect("the old acknowledged part should remain completable"); + let mut reader = set_disks + .get_object_reader(&bucket, object, None, HeaderMap::new(), &ObjectOptions::default()) + .await + .expect("completed object should be readable"); + let mut restored = Vec::new(); + reader + .stream + .read_to_end(&mut restored) + .await + .expect("completed object should stream fully"); + assert_eq!(restored, payload); + } + + #[tokio::test] + async fn upload_part_retry_quorum_failure_preserves_old_part_across_ec_geometries() { + assert_quorum_minus_one_retry_preserves_completable_part(4, 2, &[0, 1]).await; + assert_quorum_minus_one_retry_preserves_completable_part(4, 2, &[2, 3]).await; + assert_quorum_minus_one_retry_preserves_completable_part(6, 2, &[0, 1, 2]).await; + assert_quorum_minus_one_retry_preserves_completable_part(6, 2, &[3, 4, 5]).await; + } + + #[tokio::test] + async fn complete_multipart_upload_recovers_interrupted_part_retry() { + use tokio::io::AsyncReadExt as _; + + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks_for_pool_with_default_parity(6, 0, 2).await; + let bucket = "multipart-retry-recovery"; + let object = "object"; + make_bucket_on_all(&disk_stores, bucket).await; + + let payload = vec![0x42; 1 << 20]; + let (upload_id, parts) = + stage_upload_with_create_opts(&set_disks, bucket, object, &payload, &ObjectOptions::default()).await; + let (upload_meta, _) = set_disks + .check_upload_id_exists(bucket, object, &upload_id, false) + .await + .expect("staged upload metadata should be readable"); + let upload_path = SetDisks::get_upload_id_dir(bucket, object, &upload_id); + let part_path = format!( + "{}/{}/part.1", + upload_path, + upload_meta.data_dir.expect("multipart upload should have a data directory") + ); + let retry_meta = Bytes::from_static(b"interrupted retry metadata"); + + for (index, disk) in disk_stores.iter().enumerate().take(3) { + let retry_path = format!("{}/part.1", Uuid::new_v4()); + disk.write_all(RUSTFS_META_TMP_BUCKET, &retry_path, Bytes::from_static(b"interrupted retry shard")) + .await + .expect("retry shard should be staged"); + disk.prepare_part_transaction( + RUSTFS_META_TMP_BUCKET, + &retry_path, + RUSTFS_META_MULTIPART_BUCKET, + &part_path, + retry_meta.clone(), + ) + .await + .expect("part transaction should be prepared"); + disk.rename_part( + RUSTFS_META_TMP_BUCKET, + &retry_path, + RUSTFS_META_MULTIPART_BUCKET, + &part_path, + retry_meta.clone(), + ) + .await + .unwrap_or_else(|err| panic!("retry shard {index} should be published: {err}")); + } + + set_disks + .clone() + .complete_multipart_upload(bucket, object, &upload_id, parts, &ObjectOptions::default()) + .await + .expect("completion should roll back the interrupted quorum-minus-one retry"); + + let mut reader = set_disks + .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default()) + .await + .expect("completed object should be readable"); + let mut restored = Vec::new(); + reader + .stream + .read_to_end(&mut restored) + .await + .expect("completed object should stream fully"); + assert_eq!(restored, payload); + } + + #[tokio::test] + async fn rename_part_quorum_failure_without_old_part_removes_new_shards() { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks_for_pool_with_default_parity(4, 0, 2).await; + let src_path = format!("{}/part.1", Uuid::new_v4()); + let dst_path = format!("{}/part.1", Uuid::new_v4()); + + let mut retry_disks = vec![None; disk_stores.len()]; + for index in [0, 1] { + disk_stores[index] + .write_all(RUSTFS_META_TMP_BUCKET, &src_path, Bytes::from_static(b"new shard")) + .await + .expect("new shard should be staged"); + retry_disks[index] = Some(disk_stores[index].clone()); + } + + let err = set_disks + .rename_part( + &retry_disks, + RUSTFS_META_TMP_BUCKET, + &src_path, + RUSTFS_META_MULTIPART_BUCKET, + &dst_path, + Bytes::from_static(b"retry metadata"), + 3, + None, + ) + .await + .expect_err("two renamed shards must remain below write quorum"); + assert_eq!(err, DiskError::ErasureWriteQuorum); + for (index, disk) in disk_stores.iter().enumerate() { + assert!( + matches!(disk.read_all(RUSTFS_META_MULTIPART_BUCKET, &dst_path).await, Err(DiskError::FileNotFound)), + "failed first upload must not leave a destination on disk {index}" + ); + } + } + async fn make_multipart_lock_test_set_disks() -> Arc { let endpoints = vec![ Endpoint::try_from("http://127.0.0.1:9000/data").expect("first endpoint should parse"), diff --git a/crates/protos/src/generated/proto_gen/node_service.rs b/crates/protos/src/generated/proto_gen/node_service.rs index 6c7fc7fd5..8bc8a386f 100644 --- a/crates/protos/src/generated/proto_gen/node_service.rs +++ b/crates/protos/src/generated/proto_gen/node_service.rs @@ -233,6 +233,46 @@ pub struct RenamePartResponse { pub error: ::core::option::Option, } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct PreparePartTransactionRequest { + #[prost(string, tag = "1")] + pub disk: ::prost::alloc::string::String, + #[prost(string, tag = "2")] + pub src_volume: ::prost::alloc::string::String, + #[prost(string, tag = "3")] + pub src_path: ::prost::alloc::string::String, + #[prost(string, tag = "4")] + pub dst_volume: ::prost::alloc::string::String, + #[prost(string, tag = "5")] + pub dst_path: ::prost::alloc::string::String, + #[prost(bytes = "bytes", tag = "6")] + pub meta: ::prost::bytes::Bytes, +} +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct PreparePartTransactionResponse { + #[prost(bool, tag = "1")] + pub success: bool, + #[prost(message, optional, tag = "2")] + pub error: ::core::option::Option, +} +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct SettlePartTransactionRequest { + #[prost(string, tag = "1")] + pub disk: ::prost::alloc::string::String, + #[prost(string, tag = "2")] + pub volume: ::prost::alloc::string::String, + #[prost(string, tag = "3")] + pub path: ::prost::alloc::string::String, + #[prost(bool, tag = "4")] + pub rollback: bool, +} +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct SettlePartTransactionResponse { + #[prost(bool, tag = "1")] + pub success: bool, + #[prost(message, optional, tag = "2")] + pub error: ::core::option::Option, +} +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct RenameFileRequest { #[prost(string, tag = "1")] pub disk: ::prost::alloc::string::String, @@ -1600,6 +1640,21 @@ pub mod node_service_client { .insert(GrpcMethod::new("node_service.NodeService", "CheckParts")); self.inner.unary(req, path, codec).await } + pub async fn prepare_part_transaction( + &mut self, + request: impl tonic::IntoRequest, + ) -> std::result::Result, tonic::Status> { + self.inner + .ready() + .await + .map_err(|e| tonic::Status::unknown(format!("Service was not ready: {}", e.into())))?; + let codec = tonic_prost::ProstCodec::default(); + let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/PreparePartTransaction"); + let mut req = request.into_request(); + req.extensions_mut() + .insert(GrpcMethod::new("node_service.NodeService", "PreparePartTransaction")); + self.inner.unary(req, path, codec).await + } pub async fn rename_part( &mut self, request: impl tonic::IntoRequest, @@ -1615,6 +1670,21 @@ pub mod node_service_client { .insert(GrpcMethod::new("node_service.NodeService", "RenamePart")); self.inner.unary(req, path, codec).await } + pub async fn settle_part_transaction( + &mut self, + request: impl tonic::IntoRequest, + ) -> std::result::Result, tonic::Status> { + self.inner + .ready() + .await + .map_err(|e| tonic::Status::unknown(format!("Service was not ready: {}", e.into())))?; + let codec = tonic_prost::ProstCodec::default(); + let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/SettlePartTransaction"); + let mut req = request.into_request(); + req.extensions_mut() + .insert(GrpcMethod::new("node_service.NodeService", "SettlePartTransaction")); + self.inner.unary(req, path, codec).await + } pub async fn rename_file( &mut self, request: impl tonic::IntoRequest, @@ -2726,10 +2796,18 @@ pub mod node_service_server { &self, request: tonic::Request, ) -> std::result::Result, tonic::Status>; + async fn prepare_part_transaction( + &self, + request: tonic::Request, + ) -> std::result::Result, tonic::Status>; async fn rename_part( &self, request: tonic::Request, ) -> std::result::Result, tonic::Status>; + async fn settle_part_transaction( + &self, + request: tonic::Request, + ) -> std::result::Result, tonic::Status>; async fn rename_file( &self, request: tonic::Request, @@ -3432,6 +3510,34 @@ pub mod node_service_server { }; Box::pin(fut) } + "/node_service.NodeService/PreparePartTransaction" => { + #[allow(non_camel_case_types)] + struct PreparePartTransactionSvc(pub Arc); + impl tonic::server::UnaryService for PreparePartTransactionSvc { + type Response = super::PreparePartTransactionResponse; + type Future = BoxFuture, tonic::Status>; + fn call(&mut self, request: tonic::Request) -> Self::Future { + let inner = Arc::clone(&self.0); + let fut = async move { ::prepare_part_transaction(&inner, request).await }; + Box::pin(fut) + } + } + let accept_compression_encodings = self.accept_compression_encodings; + let send_compression_encodings = self.send_compression_encodings; + let max_decoding_message_size = self.max_decoding_message_size; + let max_encoding_message_size = self.max_encoding_message_size; + let inner = self.inner.clone(); + let fut = async move { + let method = PreparePartTransactionSvc(inner); + let codec = tonic_prost::ProstCodec::default(); + let mut grpc = tonic::server::Grpc::new(codec) + .apply_compression_config(accept_compression_encodings, send_compression_encodings) + .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + let res = grpc.unary(method, req).await; + Ok(res) + }; + Box::pin(fut) + } "/node_service.NodeService/RenamePart" => { #[allow(non_camel_case_types)] struct RenamePartSvc(pub Arc); @@ -3460,6 +3566,34 @@ pub mod node_service_server { }; Box::pin(fut) } + "/node_service.NodeService/SettlePartTransaction" => { + #[allow(non_camel_case_types)] + struct SettlePartTransactionSvc(pub Arc); + impl tonic::server::UnaryService for SettlePartTransactionSvc { + type Response = super::SettlePartTransactionResponse; + type Future = BoxFuture, tonic::Status>; + fn call(&mut self, request: tonic::Request) -> Self::Future { + let inner = Arc::clone(&self.0); + let fut = async move { ::settle_part_transaction(&inner, request).await }; + Box::pin(fut) + } + } + let accept_compression_encodings = self.accept_compression_encodings; + let send_compression_encodings = self.send_compression_encodings; + let max_decoding_message_size = self.max_decoding_message_size; + let max_encoding_message_size = self.max_encoding_message_size; + let inner = self.inner.clone(); + let fut = async move { + let method = SettlePartTransactionSvc(inner); + let codec = tonic_prost::ProstCodec::default(); + let mut grpc = tonic::server::Grpc::new(codec) + .apply_compression_config(accept_compression_encodings, send_compression_encodings) + .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + let res = grpc.unary(method, req).await; + Ok(res) + }; + Box::pin(fut) + } "/node_service.NodeService/RenameFile" => { #[allow(non_camel_case_types)] struct RenameFileSvc(pub Arc); diff --git a/crates/protos/src/lib.rs b/crates/protos/src/lib.rs index 6a1c96d69..165ac3c52 100644 --- a/crates/protos/src/lib.rs +++ b/crates/protos/src/lib.rs @@ -600,6 +600,30 @@ pub fn canonical_rename_part_request_body( Ok(body.finish()) } +pub fn canonical_prepare_part_transaction_request_body( + request: &proto_gen::node_service::PreparePartTransactionRequest, +) -> Result, std::num::TryFromIntError> { + let mut body = CanonicalBodyBuilder::new(b"rustfs-prepare-part-transaction-request-v1\0"); + body.push_str(&request.disk)?; + body.push_str(&request.src_volume)?; + body.push_str(&request.src_path)?; + body.push_str(&request.dst_volume)?; + body.push_str(&request.dst_path)?; + body.push_bytes(&request.meta)?; + Ok(body.finish()) +} + +pub fn canonical_settle_part_transaction_request_body( + request: &proto_gen::node_service::SettlePartTransactionRequest, +) -> Result, std::num::TryFromIntError> { + let mut body = CanonicalBodyBuilder::new(b"rustfs-settle-part-transaction-request-v1\0"); + body.push_str(&request.disk)?; + body.push_str(&request.volume)?; + body.push_str(&request.path)?; + body.push_bool(request.rollback); + Ok(body.finish()) +} + pub fn canonical_delete_volume_request_body( request: &proto_gen::node_service::DeleteVolumeRequest, ) -> Result, std::num::TryFromIntError> { @@ -637,8 +661,8 @@ pub fn canonical_make_volumes_request_body( mod disk_mutation_canonical_tests { use super::proto_gen::node_service::{ DeletePathsRequest, DeleteRequest, DeleteVersionRequest, DeleteVersionsRequest, DeleteVolumeRequest, MakeVolumeRequest, - MakeVolumesRequest, RenameDataRequest, RenameFileRequest, RenamePartRequest, UpdateMetadataRequest, WriteAllRequest, - WriteMetadataRequest, + MakeVolumesRequest, PreparePartTransactionRequest, RenameDataRequest, RenameFileRequest, RenamePartRequest, + SettlePartTransactionRequest, UpdateMetadataRequest, WriteAllRequest, WriteMetadataRequest, }; use super::*; @@ -886,6 +910,48 @@ mod disk_mutation_canonical_tests { } assert_all_distinct(&bodies); + let prepare_part = PreparePartTransactionRequest { + disk: "d".into(), + src_volume: "sv".into(), + src_path: "sp".into(), + dst_volume: "dv".into(), + dst_path: "dp".into(), + meta: vec![0x01].into(), + }; + let mut bodies = vec![canonical_prepare_part_transaction_request_body(&prepare_part).unwrap()]; + for mutate in [ + |r: &mut PreparePartTransactionRequest| r.disk = "d2".into(), + |r: &mut PreparePartTransactionRequest| r.src_volume = "sv2".into(), + |r: &mut PreparePartTransactionRequest| r.src_path = "sp2".into(), + |r: &mut PreparePartTransactionRequest| r.dst_volume = "dv2".into(), + |r: &mut PreparePartTransactionRequest| r.dst_path = "dp2".into(), + |r: &mut PreparePartTransactionRequest| r.meta = vec![0x02].into(), + ] { + let mut request = prepare_part.clone(); + mutate(&mut request); + bodies.push(canonical_prepare_part_transaction_request_body(&request).unwrap()); + } + assert_all_distinct(&bodies); + + let settle_part = SettlePartTransactionRequest { + disk: "d".into(), + volume: "v".into(), + path: "p".into(), + rollback: false, + }; + let mut bodies = vec![canonical_settle_part_transaction_request_body(&settle_part).unwrap()]; + for mutate in [ + |r: &mut SettlePartTransactionRequest| r.disk = "d2".into(), + |r: &mut SettlePartTransactionRequest| r.volume = "v2".into(), + |r: &mut SettlePartTransactionRequest| r.path = "p2".into(), + |r: &mut SettlePartTransactionRequest| r.rollback = true, + ] { + let mut request = settle_part.clone(); + mutate(&mut request); + bodies.push(canonical_settle_part_transaction_request_body(&request).unwrap()); + } + assert_all_distinct(&bodies); + let rename_part = RenamePartRequest { disk: "d".into(), src_volume: "sv".into(), diff --git a/crates/protos/src/node.proto b/crates/protos/src/node.proto index 3a0fc66ed..36df6b5cb 100644 --- a/crates/protos/src/node.proto +++ b/crates/protos/src/node.proto @@ -170,6 +170,32 @@ message RenamePartResponse { optional Error error = 2; } +message PreparePartTransactionRequest { + string disk = 1; + string src_volume = 2; + string src_path = 3; + string dst_volume = 4; + string dst_path = 5; + bytes meta = 6; +} + +message PreparePartTransactionResponse { + bool success = 1; + optional Error error = 2; +} + +message SettlePartTransactionRequest { + string disk = 1; + string volume = 2; + string path = 3; + bool rollback = 4; +} + +message SettlePartTransactionResponse { + bool success = 1; + optional Error error = 2; +} + message RenameFileRequest { string disk = 1; string src_volume = 2; @@ -943,7 +969,9 @@ service NodeService { rpc VerifyFile(VerifyFileRequest) returns (VerifyFileResponse) {}; rpc ReadParts(ReadPartsRequest) returns (ReadPartsResponse) {}; rpc CheckParts(CheckPartsRequest) returns (CheckPartsResponse) {}; + rpc PreparePartTransaction(PreparePartTransactionRequest) returns (PreparePartTransactionResponse) {}; rpc RenamePart(RenamePartRequest) returns (RenamePartResponse) {}; + rpc SettlePartTransaction(SettlePartTransactionRequest) returns (SettlePartTransactionResponse) {}; rpc RenameFile(RenameFileRequest) returns (RenameFileResponse) {}; rpc Write(WriteRequest) returns (WriteResponse) {}; rpc WriteStream(stream WriteRequest) returns (stream WriteResponse) {}; diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index fe50ccbde..a7d2b272c 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -926,10 +926,24 @@ impl Node for NodeService { self.handle_check_parts(request).await } + async fn prepare_part_transaction( + &self, + request: Request, + ) -> Result, Status> { + self.handle_prepare_part_transaction(request).await + } + async fn rename_part(&self, request: Request) -> Result, Status> { self.handle_rename_part(request).await } + async fn settle_part_transaction( + &self, + request: Request, + ) -> Result, Status> { + self.handle_settle_part_transaction(request).await + } + async fn rename_file(&self, request: Request) -> Result, Status> { self.handle_rename_file(request).await } @@ -2046,12 +2060,12 @@ mod tests { HealBucketRequest, HealControlRequest, ListBucketRequest, ListDirRequest, ListVolumesRequest, LoadBucketMetadataRequest, LoadGroupRequest, LoadPolicyMappingRequest, LoadPolicyRequest, LoadRebalanceMetaRequest, LoadServiceAccountRequest, LoadTransitionTierConfigRequest, LoadUserRequest, LocalStorageInfoRequest, MakeBucketRequest, MakeVolumeRequest, - MakeVolumesRequest, Mss, PingRequest, ReadAllRequest, ReadAtRequest, ReadMultipleRequest, ReadVersionRequest, - ReadXlRequest, ReloadPoolMetaRequest, ReloadSiteReplicationConfigRequest, RenameDataRequest, RenameFileRequest, - RenamePartRequest, ScannerActivityRequest, ServerInfoRequest, SignalServiceRequest, StartProfilingRequest, - StatVolumeRequest, StopRebalanceRequest, TierMutationPeerState, TierMutationPrepareRequest, - UpdateMetacacheListingRequest, UpdateMetadataRequest, VerifyFileRequest, WriteAllRequest, WriteMetadataRequest, - WriteRequest, + MakeVolumesRequest, Mss, PingRequest, PreparePartTransactionRequest, ReadAllRequest, ReadAtRequest, ReadMultipleRequest, + ReadVersionRequest, ReadXlRequest, ReloadPoolMetaRequest, ReloadSiteReplicationConfigRequest, RenameDataRequest, + RenameFileRequest, RenamePartRequest, ScannerActivityRequest, ServerInfoRequest, SettlePartTransactionRequest, + SignalServiceRequest, StartProfilingRequest, StatVolumeRequest, StopRebalanceRequest, TierMutationPeerState, + TierMutationPrepareRequest, UpdateMetacacheListingRequest, UpdateMetadataRequest, VerifyFileRequest, WriteAllRequest, + WriteMetadataRequest, WriteRequest, heal_control_service_client::HealControlServiceClient, heal_control_service_server::{HealControlService as _, HealControlServiceServer}, node_service_client::NodeServiceClient, @@ -2598,6 +2612,28 @@ mod tests { }, rustfs_protos::canonical_rename_part_request_body ); + assert_gated!( + prepare_part_transaction, + PreparePartTransactionRequest { + disk: disk.clone(), + src_volume: "src".into(), + src_path: "sp".into(), + dst_volume: "dst".into(), + dst_path: "dp".into(), + meta: vec![0x04].into(), + }, + rustfs_protos::canonical_prepare_part_transaction_request_body + ); + assert_gated!( + settle_part_transaction, + SettlePartTransactionRequest { + disk: disk.clone(), + volume: "dst".into(), + path: "dp".into(), + rollback: true, + }, + rustfs_protos::canonical_settle_part_transaction_request_body + ); assert_gated!( delete_volume, DeleteVolumeRequest { @@ -3316,6 +3352,39 @@ mod tests { assert!(rename_response.error.is_some()); } + #[tokio::test] + async fn test_part_transaction_invalid_disk() { + let service = create_test_node_service(); + + let prepare = service + .prepare_part_transaction(Request::new(PreparePartTransactionRequest { + disk: "invalid-disk-path".to_string(), + src_volume: "src-volume".to_string(), + src_path: "src-path".to_string(), + dst_volume: "dst-volume".to_string(), + dst_path: "dst-path".to_string(), + meta: Bytes::new(), + })) + .await + .expect("prepare RPC should return a structured disk error") + .into_inner(); + assert!(!prepare.success); + assert!(prepare.error.is_some()); + + let settle = service + .settle_part_transaction(Request::new(SettlePartTransactionRequest { + disk: "invalid-disk-path".to_string(), + volume: "dst-volume".to_string(), + path: "dst-path".to_string(), + rollback: true, + })) + .await + .expect("settle RPC should return a structured disk error") + .into_inner(); + assert!(!settle.success); + assert!(settle.error.is_some()); + } + #[tokio::test] async fn test_rename_file_invalid_disk() { let service = create_test_node_service(); diff --git a/rustfs/src/storage/rpc/node_service/disk.rs b/rustfs/src/storage/rpc/node_service/disk.rs index fef403ed0..e6c3fb60d 100644 --- a/rustfs/src/storage/rpc/node_service/disk.rs +++ b/rustfs/src/storage/rpc/node_service/disk.rs @@ -18,7 +18,7 @@ use crate::storage::storage_api::rpc_consumer::node_service::{ ReadMultipleResp, ReadOptions, StorageDiskRpcExt as _, UpdateMetadataOpts, validate_batch_read_version_item_count, }; use crate::storage::storage_api::runtime_sources_consumer::runtime_sources; -use crate::storage::storage_api::verify_tonic_mutation_body_digest; +use crate::storage::storage_api::{PartTransactionAction, verify_tonic_mutation_body_digest}; use bytes::Bytes; use rustfs_filemeta::FileInfo; use rustfs_io_metrics::internode_metrics::{ @@ -1038,6 +1038,78 @@ impl NodeService { } } + pub(super) async fn handle_prepare_part_transaction( + &self, + request: Request, + ) -> Result, Status> { + verify_disk_mutation_digest( + &request, + rustfs_protos::canonical_prepare_part_transaction_request_body(request.get_ref()), + "prepare_part_transaction", + )?; + let request = request.into_inner(); + if let Some(disk) = self.find_disk(&request.disk).await { + match disk + .prepare_part_transaction( + &request.src_volume, + &request.src_path, + &request.dst_volume, + &request.dst_path, + request.meta, + ) + .await + { + Ok(()) => Ok(Response::new(PreparePartTransactionResponse { + success: true, + error: None, + })), + Err(err) => Ok(Response::new(PreparePartTransactionResponse { + success: false, + error: Some(err.into()), + })), + } + } else { + Ok(Response::new(PreparePartTransactionResponse { + success: false, + error: Some(DiskError::other("cannot find disk".to_string()).into()), + })) + } + } + + pub(super) async fn handle_settle_part_transaction( + &self, + request: Request, + ) -> Result, Status> { + verify_disk_mutation_digest( + &request, + rustfs_protos::canonical_settle_part_transaction_request_body(request.get_ref()), + "settle_part_transaction", + )?; + let request = request.into_inner(); + if let Some(disk) = self.find_disk(&request.disk).await { + let action = if request.rollback { + PartTransactionAction::Rollback + } else { + PartTransactionAction::Commit + }; + match disk.settle_part_transaction(&request.volume, &request.path, action).await { + Ok(()) => Ok(Response::new(SettlePartTransactionResponse { + success: true, + error: None, + })), + Err(err) => Ok(Response::new(SettlePartTransactionResponse { + success: false, + error: Some(err.into()), + })), + } + } else { + Ok(Response::new(SettlePartTransactionResponse { + success: false, + error: Some(DiskError::other("cannot find disk".to_string()).into()), + })) + } + } + pub(super) async fn handle_check_parts( &self, request: Request, diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 1892d29bf..1735ce2b4 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -430,9 +430,9 @@ pub(crate) mod ecstore_data_usage { pub(crate) mod ecstore_disk { pub(crate) use rustfs_ecstore::api::disk::{ BatchReadVersionReq, BatchReadVersionResp, CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskStore, - FileInfoVersions, FileReader, FileWriter, OldCurrentSize, RUSTFS_META_BUCKET, ReadMultipleReq, ReadMultipleResp, - ReadOptions, RenameDataResp, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, get_object_disk_read_timeout, - validate_batch_read_version_item_count, + FileInfoVersions, FileReader, FileWriter, OldCurrentSize, PartTransactionAction, RUSTFS_META_BUCKET, ReadMultipleReq, + ReadMultipleResp, ReadOptions, RenameDataResp, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, + get_object_disk_read_timeout, validate_batch_read_version_item_count, }; pub(crate) use rustfs_ecstore::api::disk::{endpoint, error, error_reduce}; } @@ -629,6 +629,7 @@ pub(crate) type ReadMultipleReq = ecstore_disk::ReadMultipleReq; pub(crate) type ReadMultipleResp = ecstore_disk::ReadMultipleResp; pub(crate) type ReadOptions = ecstore_disk::ReadOptions; pub(crate) type OldCurrentSize = ecstore_disk::OldCurrentSize; +pub(crate) type PartTransactionAction = ecstore_disk::PartTransactionAction; pub(crate) type RenameDataResp = ecstore_disk::RenameDataResp; pub(crate) type ReplicationStatusType = ecstore_bucket::replication::ReplicationStatusType; pub(crate) type ReplicationStats = StorageReplicationStatsHandle; @@ -1097,6 +1098,15 @@ pub(crate) trait StorageDiskRpcExt { dst_path: &str, meta: bytes::Bytes, ) -> DiskResult<()>; + async fn prepare_part_transaction( + &self, + src_volume: &str, + src_path: &str, + dst_volume: &str, + dst_path: &str, + meta: bytes::Bytes, + ) -> DiskResult<()>; + async fn settle_part_transaction(&self, volume: &str, path: &str, action: PartTransactionAction) -> DiskResult<()>; async fn delete(&self, volume: &str, path: &str, options: DeleteOptions) -> DiskResult<()>; async fn verify_file(&self, volume: &str, path: &str, file_info: &rustfs_filemeta::FileInfo) -> DiskResult; async fn check_parts(&self, volume: &str, path: &str, file_info: &rustfs_filemeta::FileInfo) -> DiskResult; @@ -1241,6 +1251,21 @@ where ecstore_disk::DiskAPI::rename_part(self, src_volume, src_path, dst_volume, dst_path, meta).await } + async fn prepare_part_transaction( + &self, + src_volume: &str, + src_path: &str, + dst_volume: &str, + dst_path: &str, + meta: bytes::Bytes, + ) -> DiskResult<()> { + ecstore_disk::DiskAPI::prepare_part_transaction(self, src_volume, src_path, dst_volume, dst_path, meta).await + } + + async fn settle_part_transaction(&self, volume: &str, path: &str, action: PartTransactionAction) -> DiskResult<()> { + ecstore_disk::DiskAPI::settle_part_transaction(self, volume, path, action).await + } + async fn delete(&self, volume: &str, path: &str, options: DeleteOptions) -> DiskResult<()> { ecstore_disk::DiskAPI::delete(self, volume, path, options).await }