mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-22 12:26:37 +00:00
Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| a0dcbcda9b | |||
| 76e7d979d2 | |||
| 1a3be70d98 |
@@ -585,9 +585,12 @@ impl VersionsHistogram {
|
||||
}
|
||||
}
|
||||
|
||||
/// Replication statistics for a single target
|
||||
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
|
||||
pub struct ReplicationStats {
|
||||
/// Replication statistics for a single target.
|
||||
///
|
||||
/// Renamed from `ReplicationStats`; serde field names are preserved
|
||||
/// byte-identically to maintain wire compatibility with existing snapshots.
|
||||
#[derive(Debug, Default, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
||||
pub struct ReplicationTargetUsage {
|
||||
pub pending_size: u64,
|
||||
pub replicated_size: u64,
|
||||
pub failed_size: u64,
|
||||
@@ -600,7 +603,7 @@ pub struct ReplicationStats {
|
||||
pub replicated_count: u64,
|
||||
}
|
||||
|
||||
impl ReplicationStats {
|
||||
impl ReplicationTargetUsage {
|
||||
pub fn is_empty(&self) -> bool {
|
||||
let Self {
|
||||
pending_size,
|
||||
@@ -636,7 +639,7 @@ impl ReplicationStats {
|
||||
/// Replication statistics for all targets
|
||||
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
|
||||
pub struct ReplicationAllStats {
|
||||
pub targets: HashMap<String, ReplicationStats>,
|
||||
pub targets: HashMap<String, ReplicationTargetUsage>,
|
||||
pub replica_size: u64,
|
||||
pub replica_count: u64,
|
||||
}
|
||||
@@ -649,7 +652,7 @@ impl ReplicationAllStats {
|
||||
targets,
|
||||
} = self;
|
||||
|
||||
*replica_size == 0 && *replica_count == 0 && targets.values().all(ReplicationStats::is_empty)
|
||||
*replica_size == 0 && *replica_count == 0 && targets.values().all(ReplicationTargetUsage::is_empty)
|
||||
}
|
||||
|
||||
#[deprecated(note = "use is_empty instead")]
|
||||
@@ -2466,7 +2469,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn replication_stats_empty_checks_every_field() {
|
||||
type SetField = fn(&mut ReplicationStats);
|
||||
type SetField = fn(&mut ReplicationTargetUsage);
|
||||
|
||||
let cases: [(&str, SetField); 10] = [
|
||||
("pending_size", |stats| stats.pending_size = 1),
|
||||
@@ -2481,9 +2484,9 @@ mod tests {
|
||||
("replicated_count", |stats| stats.replicated_count = 1),
|
||||
];
|
||||
|
||||
assert!(ReplicationStats::default().is_empty());
|
||||
assert!(ReplicationTargetUsage::default().is_empty());
|
||||
for (field, set_nonzero) in cases {
|
||||
let mut stats = ReplicationStats::default();
|
||||
let mut stats = ReplicationTargetUsage::default();
|
||||
set_nonzero(&mut stats);
|
||||
assert!(!stats.is_empty(), "{field} must make replication stats non-empty");
|
||||
}
|
||||
@@ -2514,17 +2517,17 @@ mod tests {
|
||||
}
|
||||
|
||||
let empty_targets = ReplicationAllStats {
|
||||
targets: HashMap::from([("arn:test:empty".to_string(), ReplicationStats::default())]),
|
||||
targets: HashMap::from([("arn:test:empty".to_string(), ReplicationTargetUsage::default())]),
|
||||
..Default::default()
|
||||
};
|
||||
assert!(empty_targets.is_empty(), "all-empty targets must keep aggregate stats empty");
|
||||
|
||||
let stats = ReplicationAllStats {
|
||||
targets: HashMap::from([
|
||||
("arn:test:empty".to_string(), ReplicationStats::default()),
|
||||
("arn:test:empty".to_string(), ReplicationTargetUsage::default()),
|
||||
(
|
||||
"arn:test:non-empty".to_string(),
|
||||
ReplicationStats {
|
||||
ReplicationTargetUsage {
|
||||
pending_count: 1,
|
||||
..Default::default()
|
||||
},
|
||||
@@ -2565,7 +2568,7 @@ mod tests {
|
||||
replication_stats: Some(ReplicationAllStats {
|
||||
targets: HashMap::from([(
|
||||
"arn:test:pending".to_string(),
|
||||
ReplicationStats {
|
||||
ReplicationTargetUsage {
|
||||
pending_count: 1,
|
||||
..Default::default()
|
||||
},
|
||||
@@ -2714,7 +2717,7 @@ mod tests {
|
||||
targets: HashMap::from([
|
||||
(
|
||||
"arn:self-only".to_string(),
|
||||
ReplicationStats {
|
||||
ReplicationTargetUsage {
|
||||
pending_size: 7,
|
||||
pending_count: 1,
|
||||
..Default::default()
|
||||
@@ -2722,7 +2725,7 @@ mod tests {
|
||||
),
|
||||
(
|
||||
"arn:shared".to_string(),
|
||||
ReplicationStats {
|
||||
ReplicationTargetUsage {
|
||||
failed_size: 3,
|
||||
failed_count: 1,
|
||||
missed_threshold_size: 2,
|
||||
@@ -2741,7 +2744,7 @@ mod tests {
|
||||
targets: HashMap::from([
|
||||
(
|
||||
"arn:shared".to_string(),
|
||||
ReplicationStats {
|
||||
ReplicationTargetUsage {
|
||||
failed_size: 5,
|
||||
failed_count: 2,
|
||||
after_threshold_size: 4,
|
||||
@@ -2751,7 +2754,7 @@ mod tests {
|
||||
),
|
||||
(
|
||||
"arn:other-only".to_string(),
|
||||
ReplicationStats {
|
||||
ReplicationTargetUsage {
|
||||
replicated_size: 11,
|
||||
replicated_count: 3,
|
||||
..Default::default()
|
||||
@@ -2993,7 +2996,9 @@ mod tests {
|
||||
fn replication_target_deserialization_preserves_large_historical_maps() {
|
||||
let mut stats = ReplicationAllStats::default();
|
||||
for index in 0..=1024 {
|
||||
stats.targets.insert(format!("target-{index}"), ReplicationStats::default());
|
||||
stats
|
||||
.targets
|
||||
.insert(format!("target-{index}"), ReplicationTargetUsage::default());
|
||||
}
|
||||
let encoded = rmp_serde::to_vec_named(&stats).expect("large replication target fixture should encode");
|
||||
let decoded = rmp_serde::from_slice::<ReplicationAllStats>(&encoded)
|
||||
@@ -3002,6 +3007,47 @@ mod tests {
|
||||
assert_eq!(decoded.targets.len(), stats.targets.len());
|
||||
}
|
||||
|
||||
/// Round-trip test: encoding a [`ReplicationTargetUsage`] and decoding it back
|
||||
/// must produce the exact same value. This guards against accidental serde
|
||||
/// field-name drift during the `ReplicationStats` -> `ReplicationTargetUsage`
|
||||
/// rename. Wire-level field names are the serialized Rust field identifiers,
|
||||
/// which must remain byte-identical.
|
||||
#[test]
|
||||
fn replication_target_usage_rmp_round_trip() {
|
||||
let original = ReplicationTargetUsage {
|
||||
pending_size: 100,
|
||||
replicated_size: 2_000,
|
||||
failed_size: 50,
|
||||
failed_count: 3,
|
||||
pending_count: 7,
|
||||
missed_threshold_size: 11,
|
||||
after_threshold_size: 22,
|
||||
missed_threshold_count: 1,
|
||||
after_threshold_count: 2,
|
||||
replicated_count: 99,
|
||||
};
|
||||
|
||||
let buf = rmp_serde::to_vec_named(&original).expect("encode ReplicationTargetUsage to msgpack");
|
||||
let decoded: ReplicationTargetUsage = rmp_serde::from_slice(&buf).expect("decode ReplicationTargetUsage from msgpack");
|
||||
assert_eq!(original, decoded, "round-trip through rmp must preserve every field");
|
||||
|
||||
// Also verify that encoding as an unnamed sequence and then decoding
|
||||
// with named fields produces the correct mapping (this catches reordering).
|
||||
let named_buf = rmp_serde::to_vec_named(&original).expect("re-encode for field-name pinning");
|
||||
// Spot-check that known field names appear in the named encoding.
|
||||
let named_str = String::from_utf8_lossy(&named_buf);
|
||||
assert!(named_str.contains("pending_size"), "field 'pending_size' must survive the rename");
|
||||
assert!(named_str.contains("replicated_size"), "field 'replicated_size' must survive the rename");
|
||||
assert!(
|
||||
named_str.contains("missed_threshold_size"),
|
||||
"field 'missed_threshold_size' must survive the rename"
|
||||
);
|
||||
assert!(
|
||||
named_str.contains("after_threshold_count"),
|
||||
"field 'after_threshold_count' must survive the rename"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn checked_merge_rejects_noncanonical_histograms_without_mutation() {
|
||||
let mut entry = DataUsageEntry {
|
||||
|
||||
@@ -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<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> {
|
||||
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(());
|
||||
|
||||
@@ -784,24 +784,6 @@ pub(crate) fn create_deferred_bitrot_reader_with_stripe_handle(
|
||||
///
|
||||
/// # Returns
|
||||
/// A Result containing the BitrotWriterWrapper or an error
|
||||
/// Size hint handed to `DiskAPI::create_file` for a bitrot-wrapped shard.
|
||||
///
|
||||
/// A known length is grown by one checksum per shard so the on-disk file size
|
||||
/// matches what the bitrot writer emits. A negative length is the
|
||||
/// unknown-size sentinel (`HashReader::SIZE_PRESERVE_LAYER`, used by SSE and
|
||||
/// compression) and must be preserved: `RemoteDisk::create_file` forwards it
|
||||
/// in the `put_file_stream` query, and the receiver only treats `size > 0` as
|
||||
/// a fixed body length when locating the authenticated trailer. Clamping it
|
||||
/// to `0` would claim an empty body and misframe the stream. `0` stays `0`
|
||||
/// because a genuinely empty object still means an empty body.
|
||||
fn bitrot_create_file_size(length: i64, shard_size: usize, checksum_algo: &HashAlgorithm) -> i64 {
|
||||
if length <= 0 {
|
||||
return length;
|
||||
}
|
||||
let length = length as usize;
|
||||
(length.div_ceil(shard_size) * checksum_algo.size() + length) as i64
|
||||
}
|
||||
|
||||
pub async fn create_bitrot_writer(
|
||||
is_inline_buffer: bool,
|
||||
disk: Option<&DiskStore>,
|
||||
@@ -814,7 +796,12 @@ pub async fn create_bitrot_writer(
|
||||
let writer = if is_inline_buffer {
|
||||
CustomWriter::new_inline_buffer()
|
||||
} else if let Some(disk) = disk {
|
||||
let length = bitrot_create_file_size(length, shard_size, &checksum_algo);
|
||||
let length = if length > 0 {
|
||||
let length = length as usize;
|
||||
(length.div_ceil(shard_size) * checksum_algo.size() + length) as i64
|
||||
} else {
|
||||
0
|
||||
};
|
||||
|
||||
let file = disk.create_file("", volume, path, length).await?;
|
||||
#[cfg(feature = "hotpath")]
|
||||
@@ -833,25 +820,6 @@ mod tests {
|
||||
use rustfs_rio::ChunkReader;
|
||||
use std::collections::VecDeque;
|
||||
|
||||
#[test]
|
||||
fn bitrot_create_file_size_grows_known_length_by_checksums() {
|
||||
// 10 bytes over 4-byte shards = 3 shards, each followed by a 32-byte hash.
|
||||
assert_eq!(bitrot_create_file_size(10, 4, &HashAlgorithm::HighwayHash256), 10 + 3 * 32);
|
||||
assert_eq!(bitrot_create_file_size(10, 4, &HashAlgorithm::None), 10);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn bitrot_create_file_size_keeps_empty_and_unknown_distinct() {
|
||||
assert_eq!(bitrot_create_file_size(0, 4, &HashAlgorithm::HighwayHash256), 0);
|
||||
// SSE/compression streams advertise SIZE_PRESERVE_LAYER (-1); the remote
|
||||
// put_file_stream receiver relies on a non-positive size to parse the auth
|
||||
// trailer from the stream tail, so the sentinel must survive untouched.
|
||||
assert_eq!(
|
||||
bitrot_create_file_size(rustfs_rio::HashReader::SIZE_PRESERVE_LAYER, 4, &HashAlgorithm::HighwayHash256),
|
||||
rustfs_rio::HashReader::SIZE_PRESERVE_LAYER
|
||||
);
|
||||
}
|
||||
|
||||
struct TestChunkReader {
|
||||
chunks: VecDeque<Bytes>,
|
||||
}
|
||||
|
||||
@@ -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<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)]
|
||||
pub struct ReadMultipleRequest {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -16,7 +16,7 @@ use super::persistence::DataUsageCacheLoadAttempt;
|
||||
use super::*;
|
||||
use crate::storage_api::scanner_io::{HTTPRangeSpec, ObjectIO};
|
||||
use crate::{ScannerGetObjectReader, ScannerPutObjReader};
|
||||
use rustfs_data_usage::{ReplicationAllStats, ReplicationStats};
|
||||
use rustfs_data_usage::{ReplicationAllStats, ReplicationTargetUsage};
|
||||
use serde_json::Value;
|
||||
use std::io::Cursor;
|
||||
use std::pin::Pin;
|
||||
@@ -1636,7 +1636,7 @@ fn size_recursive_prunes_empty_and_preserves_threshold_replication_stats() {
|
||||
replication_stats: Some(ReplicationAllStats {
|
||||
targets: HashMap::from([(
|
||||
"arn:test:threshold".to_string(),
|
||||
ReplicationStats {
|
||||
ReplicationTargetUsage {
|
||||
after_threshold_count: 1,
|
||||
..Default::default()
|
||||
},
|
||||
|
||||
@@ -13,7 +13,7 @@
|
||||
// limitations under the License.
|
||||
|
||||
use super::*;
|
||||
use rustfs_data_usage::{ReplicationAllStats, ReplicationStats};
|
||||
use rustfs_data_usage::{ReplicationAllStats, ReplicationTargetUsage};
|
||||
|
||||
const TEST_PLAN_DIGEST: DataUsageScanPlanDigest = DataUsageScanPlanDigest([7; 32]);
|
||||
|
||||
@@ -271,7 +271,7 @@ fn completed_data_usage_info_flattens_nested_bucket_entries() {
|
||||
replication_stats: Some(ReplicationAllStats {
|
||||
targets: HashMap::from([(
|
||||
"arn:target".to_string(),
|
||||
ReplicationStats {
|
||||
ReplicationTargetUsage {
|
||||
replicated_size: 2048,
|
||||
replicated_count: 2,
|
||||
..Default::default()
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
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> {
|
||||
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() {
|
||||
|
||||
Reference in New Issue
Block a user