Compare commits

..

8 Commits

Author SHA1 Message Date
overtrue c2ce237d5a fix(ecstore): restore decommission test imports 2026-08-22 19:19:08 +08:00
overtrue 241271c256 fix(ecstore): satisfy decommission clippy checks 2026-08-22 18:55:44 +08:00
overtrue 3be1511214 style(ecstore): format decommission test imports 2026-08-22 18:44:46 +08:00
overtrue bc1caf0dc7 fix(ecstore): drop stale decommission test import 2026-08-22 18:44:46 +08:00
overtrue ce4a72869d style(ecstore): format decommission supervision tests 2026-08-22 18:44:14 +08:00
overtrue 255943fa43 test(ecstore): cover terminal save retry fencing 2026-08-22 18:43:56 +08:00
overtrue 5583e8373c fix(ecstore): supervise decommission worker exits 2026-08-22 18:43:56 +08:00
overtrue e1c657652f fix(ecstore): fence decommission canceler cleanup 2026-08-22 18:43:47 +08:00
5 changed files with 864 additions and 322 deletions
+18 -64
View File
@@ -585,12 +585,9 @@ impl VersionsHistogram {
}
}
/// 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 {
/// Replication statistics for a single target
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
pub struct ReplicationStats {
pub pending_size: u64,
pub replicated_size: u64,
pub failed_size: u64,
@@ -603,7 +600,7 @@ pub struct ReplicationTargetUsage {
pub replicated_count: u64,
}
impl ReplicationTargetUsage {
impl ReplicationStats {
pub fn is_empty(&self) -> bool {
let Self {
pending_size,
@@ -639,7 +636,7 @@ impl ReplicationTargetUsage {
/// Replication statistics for all targets
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
pub struct ReplicationAllStats {
pub targets: HashMap<String, ReplicationTargetUsage>,
pub targets: HashMap<String, ReplicationStats>,
pub replica_size: u64,
pub replica_count: u64,
}
@@ -652,7 +649,7 @@ impl ReplicationAllStats {
targets,
} = self;
*replica_size == 0 && *replica_count == 0 && targets.values().all(ReplicationTargetUsage::is_empty)
*replica_size == 0 && *replica_count == 0 && targets.values().all(ReplicationStats::is_empty)
}
#[deprecated(note = "use is_empty instead")]
@@ -2469,7 +2466,7 @@ mod tests {
#[test]
fn replication_stats_empty_checks_every_field() {
type SetField = fn(&mut ReplicationTargetUsage);
type SetField = fn(&mut ReplicationStats);
let cases: [(&str, SetField); 10] = [
("pending_size", |stats| stats.pending_size = 1),
@@ -2484,9 +2481,9 @@ mod tests {
("replicated_count", |stats| stats.replicated_count = 1),
];
assert!(ReplicationTargetUsage::default().is_empty());
assert!(ReplicationStats::default().is_empty());
for (field, set_nonzero) in cases {
let mut stats = ReplicationTargetUsage::default();
let mut stats = ReplicationStats::default();
set_nonzero(&mut stats);
assert!(!stats.is_empty(), "{field} must make replication stats non-empty");
}
@@ -2517,17 +2514,17 @@ mod tests {
}
let empty_targets = ReplicationAllStats {
targets: HashMap::from([("arn:test:empty".to_string(), ReplicationTargetUsage::default())]),
targets: HashMap::from([("arn:test:empty".to_string(), ReplicationStats::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(), ReplicationTargetUsage::default()),
("arn:test:empty".to_string(), ReplicationStats::default()),
(
"arn:test:non-empty".to_string(),
ReplicationTargetUsage {
ReplicationStats {
pending_count: 1,
..Default::default()
},
@@ -2568,7 +2565,7 @@ mod tests {
replication_stats: Some(ReplicationAllStats {
targets: HashMap::from([(
"arn:test:pending".to_string(),
ReplicationTargetUsage {
ReplicationStats {
pending_count: 1,
..Default::default()
},
@@ -2717,7 +2714,7 @@ mod tests {
targets: HashMap::from([
(
"arn:self-only".to_string(),
ReplicationTargetUsage {
ReplicationStats {
pending_size: 7,
pending_count: 1,
..Default::default()
@@ -2725,7 +2722,7 @@ mod tests {
),
(
"arn:shared".to_string(),
ReplicationTargetUsage {
ReplicationStats {
failed_size: 3,
failed_count: 1,
missed_threshold_size: 2,
@@ -2744,7 +2741,7 @@ mod tests {
targets: HashMap::from([
(
"arn:shared".to_string(),
ReplicationTargetUsage {
ReplicationStats {
failed_size: 5,
failed_count: 2,
after_threshold_size: 4,
@@ -2754,7 +2751,7 @@ mod tests {
),
(
"arn:other-only".to_string(),
ReplicationTargetUsage {
ReplicationStats {
replicated_size: 11,
replicated_count: 3,
..Default::default()
@@ -2996,9 +2993,7 @@ 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}"), ReplicationTargetUsage::default());
stats.targets.insert(format!("target-{index}"), ReplicationStats::default());
}
let encoded = rmp_serde::to_vec_named(&stats).expect("large replication target fixture should encode");
let decoded = rmp_serde::from_slice::<ReplicationAllStats>(&encoded)
@@ -3007,47 +3002,6 @@ 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 {
File diff suppressed because it is too large Load Diff
+2 -2
View File
@@ -33,7 +33,7 @@ use crate::bucket::utils::check_put_object_part_args;
use crate::bucket::utils::{check_valid_bucket_name, check_valid_bucket_name_strict, is_meta_bucketname};
use crate::cluster::rpc::{RemoteClient, S3PeerSys};
use crate::config::storageclass;
use crate::core::pools::PoolMeta;
use crate::core::pools::{DecommissionCanceler, PoolMeta};
use crate::disk::endpoint::{Endpoint, EndpointType};
use crate::disk::{DiskAPI, DiskInfo, DiskInfoOptions};
use crate::error::{Error, Result};
@@ -176,7 +176,7 @@ pub struct ECStore {
// pub local_disks: Vec<DiskStore>,
pub pool_meta: RwLock<PoolMeta>,
pub rebalance_meta: RwLock<Option<RebalanceMeta>>,
pub decommission_cancelers: RwLock<Vec<Option<CancellationToken>>>,
pub decommission_cancelers: RwLock<Vec<Option<DecommissionCanceler>>>,
/// Serializes rebalance/decommission start transitions.
///
/// Lock order: acquire `start_gate` before `pool_meta`, `rebalance_meta`,
@@ -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, ReplicationTargetUsage};
use rustfs_data_usage::{ReplicationAllStats, ReplicationStats};
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(),
ReplicationTargetUsage {
ReplicationStats {
after_threshold_count: 1,
..Default::default()
},
@@ -13,7 +13,7 @@
// limitations under the License.
use super::*;
use rustfs_data_usage::{ReplicationAllStats, ReplicationTargetUsage};
use rustfs_data_usage::{ReplicationAllStats, ReplicationStats};
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(),
ReplicationTargetUsage {
ReplicationStats {
replicated_size: 2048,
replicated_count: 2,
..Default::default()