diff --git a/crates/ecstore/src/cluster/rpc/remote_disk.rs b/crates/ecstore/src/cluster/rpc/remote_disk.rs index efb1b4d8d..32bf3ae57 100644 --- a/crates/ecstore/src/cluster/rpc/remote_disk.rs +++ b/crates/ecstore/src/cluster/rpc/remote_disk.rs @@ -50,10 +50,10 @@ use rustfs_protos::evict_failed_connection; 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, PreparePartTransactionRequest, ReadAllRequest, ReadMetadataRequest, - ReadMultipleRequest, ReadMultipleResponse, ReadPartsRequest, ReadVersionRequest, ReadXlRequest, RenameDataRequest, - RenameFileRequest, SettlePartTransactionRequest, SnapshotLeaseReleaseRequest, SnapshotLeaseRenewRequest, + DeleteVersionRequest, DeleteVersionsRequest, DeleteVersionsResponse, DeleteVolumeRequest, DiskInfoRequest, ListDirRequest, + ListVolumesRequest, MakeVolumeRequest, MakeVolumesRequest, PreparePartTransactionRequest, ReadAllRequest, + ReadMetadataRequest, ReadMultipleRequest, ReadMultipleResponse, ReadPartsRequest, ReadVersionRequest, ReadXlRequest, + RenameDataRequest, RenameFileRequest, SettlePartTransactionRequest, SnapshotLeaseReleaseRequest, SnapshotLeaseRenewRequest, SnapshotLeaseRequest, SnapshotLeaseResponse, StatVolumeRequest, UpdateMetadataRequest, VerifyFileRequest, WriteAllRequest, WriteMetadataRequest, node_service_client::NodeServiceClient, }; @@ -112,6 +112,28 @@ const EVENT_REMOTE_DISK_RPC: &str = "remote_disk_rpc"; const SNAPSHOT_LEASE_PROTOCOL_VERSION: u32 = 1; pub const REMOTE_SNAPSHOT_LEASE_TTL: Duration = Duration::from_secs(60); +fn decode_delete_versions_errors(response: DeleteVersionsResponse, expected_len: usize) -> Vec> { + if !response.item_errors.is_empty() { + if response.item_errors.len() != expected_len { + return vec![Some(Error::other("malformed delete_versions item errors")); expected_len]; + } + return response + .item_errors + .into_iter() + .map(|error| (error.code != 0).then(|| error.into())) + .collect(); + } + + if response.errors.len() != expected_len { + return vec![Some(Error::other("malformed delete_versions errors")); expected_len]; + } + response + .errors + .into_iter() + .map(|error| (!error.is_empty()).then(|| Error::other(error))) + .collect() +} + fn snapshot_lease_token_from_response(response: SnapshotLeaseResponse) -> Result { if !response.success { return Err(response.error.unwrap_or_default().into()); @@ -2406,8 +2428,6 @@ impl DiskAPI for RemoteDisk { return errors; } - // TODO(backlog): replace string errors with typed `StorageError` variants - let result = self .execute_with_timeout( || async { @@ -2439,17 +2459,7 @@ impl DiskAPI for RemoteDisk { } return errors; } - response - .errors - .iter() - .map(|error| { - if error.is_empty() { - None - } else { - Some(Error::other(error.to_string())) - } - }) - .collect() + decode_delete_versions_errors(response, versions.len()) } #[tracing::instrument(level = "trace", skip_all)] @@ -3760,6 +3770,63 @@ mod tests { static INIT: Once = Once::new(); + #[test] + fn delete_versions_response_preserves_typed_item_errors() { + let errors = decode_delete_versions_errors( + DeleteVersionsResponse { + success: true, + errors: vec!["file not found".to_string(), String::new()], + error: None, + item_errors: vec![ + rustfs_protos::proto_gen::node_service::Error { + code: DiskError::FileNotFound.to_u32(), + error_info: "file not found".to_string(), + }, + rustfs_protos::proto_gen::node_service::Error::default(), + ], + }, + 2, + ); + + assert!(matches!(errors.as_slice(), [Some(DiskError::FileNotFound), None])); + } + + #[test] + fn delete_versions_response_accepts_legacy_string_errors() { + let errors = decode_delete_versions_errors( + DeleteVersionsResponse { + success: true, + errors: vec!["legacy error".to_string(), String::new()], + error: None, + item_errors: Vec::new(), + }, + 2, + ); + + assert_eq!(errors.len(), 2); + assert_eq!(errors[0].as_ref().map(ToString::to_string).as_deref(), Some("io error legacy error")); + assert!(errors[1].is_none()); + } + + #[test] + fn delete_versions_response_rejects_misaligned_item_errors() { + let errors = decode_delete_versions_errors( + DeleteVersionsResponse { + success: true, + errors: vec!["file not found".to_string()], + error: None, + item_errors: vec![rustfs_protos::proto_gen::node_service::Error { + code: DiskError::FileNotFound.to_u32(), + error_info: "file not found".to_string(), + }], + }, + 2, + ); + + assert_eq!(errors.len(), 2); + assert!(errors.iter().all(Option::is_some)); + } + #[test] fn disk_mutation_digest_marks_rolling_compatibility() { let mut request = Request::new(()); diff --git a/crates/protos/src/generated/proto_gen/node_service.rs b/crates/protos/src/generated/proto_gen/node_service.rs index 6fa01bf1e..3885be4a3 100644 --- a/crates/protos/src/generated/proto_gen/node_service.rs +++ b/crates/protos/src/generated/proto_gen/node_service.rs @@ -722,6 +722,10 @@ pub struct DeleteVersionsResponse { pub errors: ::prost::alloc::vec::Vec<::prost::alloc::string::String>, #[prost(message, optional, tag = "3")] pub error: ::core::option::Option, + /// Senders dual-write the legacy strings and typed entries. Receivers prefer typed entries + /// when present and fall back to strings for peers that predate this field. Code zero means success. + #[prost(message, repeated, tag = "4")] + pub item_errors: ::prost::alloc::vec::Vec, } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct ReadMultipleRequest { diff --git a/crates/protos/src/node.proto b/crates/protos/src/node.proto index 290b93b03..70ca93565 100644 --- a/crates/protos/src/node.proto +++ b/crates/protos/src/node.proto @@ -493,6 +493,9 @@ message DeleteVersionsResponse { bool success = 1; repeated string errors = 2; optional Error error = 3; + // Senders dual-write the legacy strings and typed entries. Receivers prefer typed entries + // when present and fall back to strings for peers that predate this field. Code zero means success. + repeated Error item_errors = 4; } message ReadMultipleRequest { diff --git a/rustfs/src/storage/rpc/node_service/disk.rs b/rustfs/src/storage/rpc/node_service/disk.rs index 5d9fbc969..de80a10f0 100644 --- a/rustfs/src/storage/rpc/node_service/disk.rs +++ b/rustfs/src/storage/rpc/node_service/disk.rs @@ -146,6 +146,29 @@ fn encode_file_info_msgpack(value: &FileInfo) -> std::result::Result, Di encode_msgpack_with_capacity(value, "FileInfo", FILE_INFO_MSGPACK_ENCODE_CAPACITY_HINT) } +fn encode_delete_versions_errors(disk_errors: Vec>) -> (Vec, Vec) { + let mut errors = Vec::with_capacity(disk_errors.len()); + let mut item_errors = Vec::with_capacity(disk_errors.len()); + for error in disk_errors { + match error { + Some(error) => { + let code = match &error { + DiskError::Io(source) if source.kind() == std::io::ErrorKind::NotFound => DiskError::FileNotFound.to_u32(), + _ => error.to_u32(), + }; + let error_info = error.to_string(); + errors.push(error_info.clone()); + item_errors.push(Error { code, error_info }); + } + None => { + errors.push(String::new()); + item_errors.push(Error::default()); + } + } + } + (errors, item_errors) +} + fn encode_msgpack_named(value: &T, value_name: &str) -> std::result::Result, DiskError> { let mut serializer = rmp_serde::Serializer::new(Vec::with_capacity(MSGPACK_ENCODE_CAPACITY_HINT)).with_struct_map(); value @@ -552,6 +575,7 @@ impl NodeService { success: false, errors: Vec::new(), error: Some(DiskError::other(format!("decode FileInfoVersions failed: {err}")).into()), + item_errors: Vec::new(), })); } }; @@ -563,30 +587,26 @@ impl NodeService { success: false, errors: Vec::new(), error: Some(DiskError::other(format!("decode DeleteOptions failed: {err}")).into()), + item_errors: Vec::new(), })); } }; - let errors = disk - .delete_versions(&request.volume, versions, opts) - .await - .into_iter() - .map(|error| match error { - Some(e) => e.to_string(), - None => "".to_string(), - }) - .collect(); + let (errors, item_errors) = + encode_delete_versions_errors(disk.delete_versions(&request.volume, versions, opts).await); Ok(Response::new(DeleteVersionsResponse { success: true, errors, error: None, + item_errors, })) } else { Ok(Response::new(DeleteVersionsResponse { success: false, errors: Vec::new(), error: Some(DiskError::other("cannot find disk".to_string()).into()), + item_errors: Vec::new(), })) } } @@ -1612,8 +1632,8 @@ impl NodeService { mod tests { use super::{ compat_response_json, decode_msgpack_or_json, decode_rename_data_request_file_info, - encode_batch_read_version_response_payloads, encode_file_info_msgpack, encode_msgpack, encode_msgpack_named, - encode_read_multiple_response_payloads, encode_rename_data_response_payloads, + encode_batch_read_version_response_payloads, encode_delete_versions_errors, encode_file_info_msgpack, encode_msgpack, + encode_msgpack_named, encode_read_multiple_response_payloads, encode_rename_data_response_payloads, }; use crate::storage::rpc::node_service::make_server; use crate::storage::storage_api::ReadMultipleResp; @@ -1632,6 +1652,18 @@ mod tests { count: u32, } + #[test] + fn delete_versions_response_dual_writes_typed_item_errors() { + let raw_not_found = super::DiskError::Io(std::io::Error::from(std::io::ErrorKind::NotFound)); + let (errors, item_errors) = encode_delete_versions_errors(vec![Some(raw_not_found), None]); + + assert!(errors[0].starts_with("io error ")); + assert!(errors[1].is_empty()); + assert_eq!(item_errors[0].code, super::DiskError::FileNotFound.to_u32()); + assert_eq!(item_errors[0].error_info, errors[0]); + assert_eq!(item_errors[1].code, 0); + } + #[tokio::test] #[serial] async fn handle_read_version_records_attribution_for_missing_disk() {