mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-23 12:49:04 +00:00
fix(ecstore): preserve remote delete error types (#6371)
This commit is contained in:
@@ -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::RenamePartRequest;
|
||||||
use rustfs_protos::proto_gen::node_service::{
|
use rustfs_protos::proto_gen::node_service::{
|
||||||
BatchReadVersionRequest, BatchReadVersionResponse, CheckPartsRequest, DeletePathsRequest, DeleteRequest,
|
BatchReadVersionRequest, BatchReadVersionResponse, CheckPartsRequest, DeletePathsRequest, DeleteRequest,
|
||||||
DeleteVersionRequest, DeleteVersionsRequest, DeleteVolumeRequest, DiskInfoRequest, ListDirRequest, ListVolumesRequest,
|
DeleteVersionRequest, DeleteVersionsRequest, DeleteVersionsResponse, DeleteVolumeRequest, DiskInfoRequest, ListDirRequest,
|
||||||
MakeVolumeRequest, MakeVolumesRequest, PreparePartTransactionRequest, ReadAllRequest, ReadMetadataRequest,
|
ListVolumesRequest, MakeVolumeRequest, MakeVolumesRequest, PreparePartTransactionRequest, ReadAllRequest,
|
||||||
ReadMultipleRequest, ReadMultipleResponse, ReadPartsRequest, ReadVersionRequest, ReadXlRequest, RenameDataRequest,
|
ReadMetadataRequest, ReadMultipleRequest, ReadMultipleResponse, ReadPartsRequest, ReadVersionRequest, ReadXlRequest,
|
||||||
RenameFileRequest, SettlePartTransactionRequest, SnapshotLeaseReleaseRequest, SnapshotLeaseRenewRequest,
|
RenameDataRequest, RenameFileRequest, SettlePartTransactionRequest, SnapshotLeaseReleaseRequest, SnapshotLeaseRenewRequest,
|
||||||
SnapshotLeaseRequest, SnapshotLeaseResponse, StatVolumeRequest, UpdateMetadataRequest, VerifyFileRequest, WriteAllRequest,
|
SnapshotLeaseRequest, SnapshotLeaseResponse, StatVolumeRequest, UpdateMetadataRequest, VerifyFileRequest, WriteAllRequest,
|
||||||
WriteMetadataRequest, node_service_client::NodeServiceClient,
|
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;
|
const SNAPSHOT_LEASE_PROTOCOL_VERSION: u32 = 1;
|
||||||
pub const REMOTE_SNAPSHOT_LEASE_TTL: Duration = Duration::from_secs(60);
|
pub const REMOTE_SNAPSHOT_LEASE_TTL: Duration = Duration::from_secs(60);
|
||||||
|
|
||||||
|
fn decode_delete_versions_errors(response: DeleteVersionsResponse, expected_len: usize) -> Vec<Option<Error>> {
|
||||||
|
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<SnapshotLeaseToken> {
|
fn snapshot_lease_token_from_response(response: SnapshotLeaseResponse) -> Result<SnapshotLeaseToken> {
|
||||||
if !response.success {
|
if !response.success {
|
||||||
return Err(response.error.unwrap_or_default().into());
|
return Err(response.error.unwrap_or_default().into());
|
||||||
@@ -2406,8 +2428,6 @@ impl DiskAPI for RemoteDisk {
|
|||||||
return errors;
|
return errors;
|
||||||
}
|
}
|
||||||
|
|
||||||
// TODO(backlog): replace string errors with typed `StorageError` variants
|
|
||||||
|
|
||||||
let result = self
|
let result = self
|
||||||
.execute_with_timeout(
|
.execute_with_timeout(
|
||||||
|| async {
|
|| async {
|
||||||
@@ -2439,17 +2459,7 @@ impl DiskAPI for RemoteDisk {
|
|||||||
}
|
}
|
||||||
return errors;
|
return errors;
|
||||||
}
|
}
|
||||||
response
|
decode_delete_versions_errors(response, versions.len())
|
||||||
.errors
|
|
||||||
.iter()
|
|
||||||
.map(|error| {
|
|
||||||
if error.is_empty() {
|
|
||||||
None
|
|
||||||
} else {
|
|
||||||
Some(Error::other(error.to_string()))
|
|
||||||
}
|
|
||||||
})
|
|
||||||
.collect()
|
|
||||||
}
|
}
|
||||||
|
|
||||||
#[tracing::instrument(level = "trace", skip_all)]
|
#[tracing::instrument(level = "trace", skip_all)]
|
||||||
@@ -3760,6 +3770,63 @@ mod tests {
|
|||||||
|
|
||||||
static INIT: Once = Once::new();
|
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]
|
#[test]
|
||||||
fn disk_mutation_digest_marks_rolling_compatibility() {
|
fn disk_mutation_digest_marks_rolling_compatibility() {
|
||||||
let mut request = Request::new(());
|
let mut request = Request::new(());
|
||||||
|
|||||||
@@ -722,6 +722,10 @@ pub struct DeleteVersionsResponse {
|
|||||||
pub errors: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
|
pub errors: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
|
||||||
#[prost(message, optional, tag = "3")]
|
#[prost(message, optional, tag = "3")]
|
||||||
pub error: ::core::option::Option<Error>,
|
pub error: ::core::option::Option<Error>,
|
||||||
|
/// 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<Error>,
|
||||||
}
|
}
|
||||||
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
|
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
|
||||||
pub struct ReadMultipleRequest {
|
pub struct ReadMultipleRequest {
|
||||||
|
|||||||
@@ -493,6 +493,9 @@ message DeleteVersionsResponse {
|
|||||||
bool success = 1;
|
bool success = 1;
|
||||||
repeated string errors = 2;
|
repeated string errors = 2;
|
||||||
optional Error error = 3;
|
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 {
|
message ReadMultipleRequest {
|
||||||
|
|||||||
@@ -146,6 +146,29 @@ fn encode_file_info_msgpack(value: &FileInfo) -> std::result::Result<Vec<u8>, Di
|
|||||||
encode_msgpack_with_capacity(value, "FileInfo", FILE_INFO_MSGPACK_ENCODE_CAPACITY_HINT)
|
encode_msgpack_with_capacity(value, "FileInfo", FILE_INFO_MSGPACK_ENCODE_CAPACITY_HINT)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn encode_delete_versions_errors(disk_errors: Vec<Option<DiskError>>) -> (Vec<String>, Vec<Error>) {
|
||||||
|
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<T: serde::Serialize>(value: &T, value_name: &str) -> std::result::Result<Vec<u8>, DiskError> {
|
fn encode_msgpack_named<T: serde::Serialize>(value: &T, value_name: &str) -> std::result::Result<Vec<u8>, DiskError> {
|
||||||
let mut serializer = rmp_serde::Serializer::new(Vec::with_capacity(MSGPACK_ENCODE_CAPACITY_HINT)).with_struct_map();
|
let mut serializer = rmp_serde::Serializer::new(Vec::with_capacity(MSGPACK_ENCODE_CAPACITY_HINT)).with_struct_map();
|
||||||
value
|
value
|
||||||
@@ -552,6 +575,7 @@ impl NodeService {
|
|||||||
success: false,
|
success: false,
|
||||||
errors: Vec::new(),
|
errors: Vec::new(),
|
||||||
error: Some(DiskError::other(format!("decode FileInfoVersions failed: {err}")).into()),
|
error: Some(DiskError::other(format!("decode FileInfoVersions failed: {err}")).into()),
|
||||||
|
item_errors: Vec::new(),
|
||||||
}));
|
}));
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
@@ -563,30 +587,26 @@ impl NodeService {
|
|||||||
success: false,
|
success: false,
|
||||||
errors: Vec::new(),
|
errors: Vec::new(),
|
||||||
error: Some(DiskError::other(format!("decode DeleteOptions failed: {err}")).into()),
|
error: Some(DiskError::other(format!("decode DeleteOptions failed: {err}")).into()),
|
||||||
|
item_errors: Vec::new(),
|
||||||
}));
|
}));
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
let errors = disk
|
let (errors, item_errors) =
|
||||||
.delete_versions(&request.volume, versions, opts)
|
encode_delete_versions_errors(disk.delete_versions(&request.volume, versions, opts).await);
|
||||||
.await
|
|
||||||
.into_iter()
|
|
||||||
.map(|error| match error {
|
|
||||||
Some(e) => e.to_string(),
|
|
||||||
None => "".to_string(),
|
|
||||||
})
|
|
||||||
.collect();
|
|
||||||
|
|
||||||
Ok(Response::new(DeleteVersionsResponse {
|
Ok(Response::new(DeleteVersionsResponse {
|
||||||
success: true,
|
success: true,
|
||||||
errors,
|
errors,
|
||||||
error: None,
|
error: None,
|
||||||
|
item_errors,
|
||||||
}))
|
}))
|
||||||
} else {
|
} else {
|
||||||
Ok(Response::new(DeleteVersionsResponse {
|
Ok(Response::new(DeleteVersionsResponse {
|
||||||
success: false,
|
success: false,
|
||||||
errors: Vec::new(),
|
errors: Vec::new(),
|
||||||
error: Some(DiskError::other("cannot find disk".to_string()).into()),
|
error: Some(DiskError::other("cannot find disk".to_string()).into()),
|
||||||
|
item_errors: Vec::new(),
|
||||||
}))
|
}))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -1612,8 +1632,8 @@ impl NodeService {
|
|||||||
mod tests {
|
mod tests {
|
||||||
use super::{
|
use super::{
|
||||||
compat_response_json, decode_msgpack_or_json, decode_rename_data_request_file_info,
|
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_batch_read_version_response_payloads, encode_delete_versions_errors, encode_file_info_msgpack, encode_msgpack,
|
||||||
encode_read_multiple_response_payloads, encode_rename_data_response_payloads,
|
encode_msgpack_named, encode_read_multiple_response_payloads, encode_rename_data_response_payloads,
|
||||||
};
|
};
|
||||||
use crate::storage::rpc::node_service::make_server;
|
use crate::storage::rpc::node_service::make_server;
|
||||||
use crate::storage::storage_api::ReadMultipleResp;
|
use crate::storage::storage_api::ReadMultipleResp;
|
||||||
@@ -1632,6 +1652,18 @@ mod tests {
|
|||||||
count: u32,
|
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]
|
#[tokio::test]
|
||||||
#[serial]
|
#[serial]
|
||||||
async fn handle_read_version_records_attribution_for_missing_disk() {
|
async fn handle_read_version_records_attribution_for_missing_disk() {
|
||||||
|
|||||||
Reference in New Issue
Block a user