mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-22 04:16:38 +00:00
Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| da3fd83aa7 | |||
| a2e7036cb1 |
@@ -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 {
|
||||
|
||||
@@ -36,8 +36,7 @@ use crate::disk::error::DiskError;
|
||||
use crate::disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET};
|
||||
use crate::error::{Error, Result};
|
||||
use crate::error::{
|
||||
StorageError, is_err_bucket_exists, is_err_bucket_not_found, is_err_object_not_found, is_err_operation_canceled,
|
||||
is_err_version_not_found,
|
||||
StorageError, is_err_bucket_exists, is_err_bucket_not_found, is_err_object_not_found, is_err_version_not_found,
|
||||
};
|
||||
use crate::layout::endpoints::EndpointServerPools;
|
||||
use crate::object_api::{GetObjectReader, ObjectOptions};
|
||||
@@ -774,76 +773,7 @@ async fn load_decommission_entry_exact_versions(
|
||||
}
|
||||
|
||||
fn resolve_decommission_check_after_list_result(list_result: Result<()>, entry_error: Option<Error>) -> Result<()> {
|
||||
match list_result {
|
||||
Ok(()) => entry_error.map_or(Ok(()), Err),
|
||||
Err(list_err) => resolve_decommission_listing_error(Some(list_err), entry_error).map_or(Ok(()), Err),
|
||||
}
|
||||
}
|
||||
|
||||
fn resolve_decommission_listing_error(listing_error: Option<Error>, entry_error: Option<Error>) -> Option<Error> {
|
||||
match (listing_error, entry_error) {
|
||||
(Some(listing_error), Some(entry_error)) if is_err_operation_canceled(&listing_error) => Some(entry_error),
|
||||
(Some(listing_error), Some(entry_error)) if is_err_operation_canceled(&entry_error) => Some(listing_error),
|
||||
(Some(listing_error), _) => Some(listing_error),
|
||||
(None, entry_error) => entry_error,
|
||||
}
|
||||
}
|
||||
|
||||
fn decommission_unresolved_listing_error(
|
||||
bucket: &str,
|
||||
prefix: &str,
|
||||
candidate: Option<&str>,
|
||||
candidate_count: usize,
|
||||
disk_error_count: usize,
|
||||
pool_index: usize,
|
||||
set_index: usize,
|
||||
) -> Error {
|
||||
let location = candidate.unwrap_or(prefix);
|
||||
Error::other(format!(
|
||||
"decommission listing could not resolve metadata for {bucket}/{location} on pool {pool_index} set {set_index} ({candidate_count} candidate(s), {disk_error_count} disk error(s))"
|
||||
))
|
||||
}
|
||||
|
||||
fn resolve_decommission_partial_listing_entry(
|
||||
entries: MetaCacheEntries,
|
||||
resolver: MetadataResolutionParams,
|
||||
bucket: &str,
|
||||
prefix: &str,
|
||||
disk_error_count: usize,
|
||||
pool_index: usize,
|
||||
set_index: usize,
|
||||
) -> Result<MetaCacheEntry> {
|
||||
let candidate_count = entries.as_ref().iter().flatten().count();
|
||||
if let Some(entry) = entries.resolve(resolver) {
|
||||
return Ok(entry);
|
||||
}
|
||||
|
||||
let candidate = entries.as_ref().iter().flatten().map(|entry| entry.name.as_str()).next();
|
||||
Err(decommission_unresolved_listing_error(
|
||||
bucket,
|
||||
prefix,
|
||||
candidate,
|
||||
candidate_count,
|
||||
disk_error_count,
|
||||
pool_index,
|
||||
set_index,
|
||||
))
|
||||
}
|
||||
|
||||
async fn record_decommission_entry_error(
|
||||
entry_error: &Arc<tokio::sync::Mutex<Option<Error>>>,
|
||||
rx: &CancellationToken,
|
||||
err: Error,
|
||||
) {
|
||||
if rx.is_cancelled() {
|
||||
return;
|
||||
}
|
||||
|
||||
let mut first_err = entry_error.lock().await;
|
||||
if first_err.is_none() && !rx.is_cancelled() {
|
||||
*first_err = Some(err);
|
||||
rx.cancel();
|
||||
}
|
||||
if let Some(err) = entry_error { Err(err) } else { list_result }
|
||||
}
|
||||
|
||||
fn resolve_decommission_pool_meta_reload_result(result: Result<()>, stage: &str) -> Result<()> {
|
||||
@@ -3608,7 +3538,6 @@ impl ECStore {
|
||||
let rx_clone = rx.clone();
|
||||
let bi = bi.clone();
|
||||
let set_id = set_idx;
|
||||
let listing_entry_error = entry_error.clone();
|
||||
let worker = tokio::spawn(async move {
|
||||
let _listing_permit = listing_permit;
|
||||
run_decommission_listing_with_retry(
|
||||
@@ -3622,11 +3551,7 @@ impl ECStore {
|
||||
let set = set.clone();
|
||||
let rx = rx_clone.clone();
|
||||
let bucket = bi.clone();
|
||||
let entry_error = listing_entry_error.clone();
|
||||
async move {
|
||||
set.list_objects_to_decommission(rx, bucket, callback, entry_error.clone(), idx, set_id)
|
||||
.await
|
||||
}
|
||||
async move { set.list_objects_to_decommission(rx, bucket, callback).await }
|
||||
},
|
||||
)
|
||||
.await
|
||||
@@ -3656,7 +3581,11 @@ impl ECStore {
|
||||
|
||||
wait_decommission_worker_drain(&workers, worker_limit).await?;
|
||||
|
||||
if let Some(err) = resolve_decommission_listing_error(listing_worker_error, entry_error.lock().await.clone()) {
|
||||
if let Some(err) = listing_worker_error {
|
||||
return Err(err);
|
||||
}
|
||||
|
||||
if let Some(err) = entry_error.lock().await.clone() {
|
||||
return Err(err);
|
||||
}
|
||||
|
||||
@@ -4262,7 +4191,7 @@ impl ECStore {
|
||||
let buckets = self.get_buckets_to_decommission().await?;
|
||||
let pool = self.pools[idx].clone();
|
||||
|
||||
for (set_index, set) in pool.disk_set.iter().enumerate() {
|
||||
for set in &pool.disk_set {
|
||||
for bucket_info in &buckets {
|
||||
let mut lifecycle_config = None;
|
||||
let mut object_lock_config = None;
|
||||
@@ -4357,7 +4286,7 @@ impl ECStore {
|
||||
});
|
||||
|
||||
let list_result = set
|
||||
.list_objects_to_decommission(callback_rx, bucket_info.clone(), callback, entry_error.clone(), idx, set_index)
|
||||
.list_objects_to_decommission(callback_rx, bucket_info.clone(), callback)
|
||||
.await;
|
||||
let entry_error = entry_error.lock().await.clone();
|
||||
resolve_decommission_check_after_list_result(list_result, entry_error)?;
|
||||
@@ -5092,15 +5021,12 @@ mod tests {
|
||||
pub type ListCallback = Arc<dyn Fn(MetaCacheEntry) -> BoxFuture<'static, ()> + Send + Sync + 'static>;
|
||||
|
||||
impl SetDisks {
|
||||
#[tracing::instrument(skip(self, rx, cb_func, entry_error))]
|
||||
#[tracing::instrument(skip(self, rx, cb_func))]
|
||||
async fn list_objects_to_decommission(
|
||||
self: &Arc<Self>,
|
||||
rx: CancellationToken,
|
||||
bucket_info: DecomBucketInfo,
|
||||
cb_func: ListCallback,
|
||||
entry_error: Arc<tokio::sync::Mutex<Option<Error>>>,
|
||||
pool_index: usize,
|
||||
set_index: usize,
|
||||
) -> Result<()> {
|
||||
let (disks, _) = self.get_online_disks_with_healing(false).await;
|
||||
ensure_decommission_listing_disks_available(!disks.is_empty(), &bucket_info.name)?;
|
||||
@@ -5115,12 +5041,6 @@ impl SetDisks {
|
||||
};
|
||||
|
||||
let cb1 = cb_func.clone();
|
||||
let unresolved_error = entry_error.clone();
|
||||
let unresolved_rx = rx.clone();
|
||||
let unresolved_bucket = bucket_info.name.clone();
|
||||
let unresolved_prefix = bucket_info.prefix.clone();
|
||||
let unresolved_pool_index = pool_index;
|
||||
let unresolved_set_index = set_index;
|
||||
|
||||
list_path_raw(
|
||||
rx,
|
||||
@@ -5133,51 +5053,20 @@ impl SetDisks {
|
||||
skip_walkdir_total_timeout: true,
|
||||
walkdir_stall_timeout: Some(DECOMMISSION_BACKGROUND_WALKDIR_STALL_TIMEOUT),
|
||||
agreed: Some(Box::new(move |entry: MetaCacheEntry| Box::pin(cb1(entry)))),
|
||||
partial: Some(Box::new(move |entries: MetaCacheEntries, errs: &[Option<DiskError>]| {
|
||||
partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option<DiskError>]| {
|
||||
let resolver = resolver.clone();
|
||||
let cb_func = cb_func.clone();
|
||||
let bucket = unresolved_bucket.clone();
|
||||
let prefix = unresolved_prefix.clone();
|
||||
let unresolved_error = unresolved_error.clone();
|
||||
let unresolved_rx = unresolved_rx.clone();
|
||||
let pool_index = unresolved_pool_index;
|
||||
let set_index = unresolved_set_index;
|
||||
let disk_error_count = errs.iter().flatten().count();
|
||||
if unresolved_rx.is_cancelled() {
|
||||
return Box::pin(async {});
|
||||
}
|
||||
|
||||
match resolve_decommission_partial_listing_entry(
|
||||
entries,
|
||||
resolver,
|
||||
&bucket,
|
||||
&prefix,
|
||||
disk_error_count,
|
||||
pool_index,
|
||||
set_index,
|
||||
) {
|
||||
Ok(entry) => {
|
||||
match entries.resolve(resolver) {
|
||||
Some(entry) => {
|
||||
warn!("decommission_pool: list_objects_to_decommission get {}", &entry.name);
|
||||
Box::pin(async move {
|
||||
cb_func(entry).await;
|
||||
})
|
||||
}
|
||||
Err(err) => Box::pin(async move {
|
||||
if unresolved_rx.is_cancelled() {
|
||||
return;
|
||||
}
|
||||
warn!(
|
||||
event = EVENT_DECOMMISSION_BUCKET,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
subsystem = LOG_SUBSYSTEM_POOLS,
|
||||
bucket = %bucket,
|
||||
prefix = %prefix,
|
||||
state = "unresolved_entry",
|
||||
error = %err,
|
||||
"Decommission listing failed closed on unresolved metadata"
|
||||
);
|
||||
record_decommission_entry_error(&unresolved_error, &unresolved_rx, err).await;
|
||||
}),
|
||||
None => {
|
||||
warn!("decommission_pool: list_objects_to_decommission get none");
|
||||
Box::pin(async {})
|
||||
}
|
||||
}
|
||||
})),
|
||||
..Default::default()
|
||||
@@ -5185,10 +5074,6 @@ impl SetDisks {
|
||||
)
|
||||
.await?;
|
||||
|
||||
if let Some(err) = entry_error.lock().await.clone() {
|
||||
return Err(err);
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
@@ -5394,12 +5279,11 @@ mod pools_tests {
|
||||
has_active_decommission_canceler, is_decommission_active, is_decommission_cancel_requested,
|
||||
load_decommission_entry_versions, local_decommission_queue_prefix, mark_decommission_bucket_done,
|
||||
merge_pool_status_refresh, missing_decommission_worker_prefix, observe_decommission_terminal_reload_result,
|
||||
pool_meta_has_active_decommission, record_decommission_entry_error, require_decommission_store,
|
||||
resolve_decommission_bucket_done_save_result, resolve_decommission_bucket_state,
|
||||
resolve_decommission_check_after_list_result, resolve_decommission_entry_cleanup_delete_result,
|
||||
resolve_decommission_entry_exact_versions, resolve_decommission_entry_reload_result, resolve_decommission_listing_error,
|
||||
resolve_decommission_listing_worker_result, resolve_decommission_optional_bucket_config_result,
|
||||
resolve_decommission_partial_listing_entry, resolve_decommission_pool_meta_reload_result,
|
||||
pool_meta_has_active_decommission, require_decommission_store, resolve_decommission_bucket_done_save_result,
|
||||
resolve_decommission_bucket_state, resolve_decommission_check_after_list_result,
|
||||
resolve_decommission_entry_cleanup_delete_result, resolve_decommission_entry_exact_versions,
|
||||
resolve_decommission_entry_reload_result, resolve_decommission_listing_worker_result,
|
||||
resolve_decommission_optional_bucket_config_result, resolve_decommission_pool_meta_reload_result,
|
||||
resolve_decommission_preflight_heal_result, resolve_decommission_progress_save_result,
|
||||
resolve_decommission_spawn_failure_result, resolve_decommission_terminal_mark_after_error_result,
|
||||
resolve_decommission_terminal_mark_result, resolve_decommission_update_after_result,
|
||||
@@ -5418,9 +5302,7 @@ mod pools_tests {
|
||||
use crate::error::{Error, StorageError};
|
||||
use crate::layout::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints};
|
||||
use crate::services::rebalance::{RebalStatus, RebalanceInfo, RebalanceMeta, RebalanceStats};
|
||||
use rustfs_filemeta::{
|
||||
FileInfo, FileInfoVersions, MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams, ObjectPartInfo,
|
||||
};
|
||||
use rustfs_filemeta::{FileInfo, FileInfoVersions, MetaCacheEntry, ObjectPartInfo};
|
||||
use rustfs_rio::Index;
|
||||
use std::sync::{
|
||||
Arc,
|
||||
@@ -6439,65 +6321,6 @@ mod pools_tests {
|
||||
assert!(matches!(err, Error::SlowDown));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_resolve_decommission_partial_listing_entry_rejects_unresolved_metadata() {
|
||||
let err = resolve_decommission_partial_listing_entry(
|
||||
MetaCacheEntries(vec![None]),
|
||||
MetadataResolutionParams {
|
||||
dir_quorum: 2,
|
||||
obj_quorum: 2,
|
||||
bucket: "bucket-a".to_string(),
|
||||
..Default::default()
|
||||
},
|
||||
"bucket-a",
|
||||
"prefix/",
|
||||
1,
|
||||
2,
|
||||
3,
|
||||
)
|
||||
.expect_err("unresolved partial listing must fail closed");
|
||||
|
||||
let message = err.to_string();
|
||||
assert!(message.contains("decommission listing could not resolve metadata"));
|
||||
assert!(message.contains("bucket-a/prefix/"));
|
||||
assert!(message.contains("pool 2 set 3"));
|
||||
assert!(message.contains("1 disk error(s)"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_record_decommission_entry_error_cancels_listing_and_preserves_first_error() {
|
||||
let entry_error = Arc::new(tokio::sync::Mutex::new(None));
|
||||
let rx = CancellationToken::new();
|
||||
|
||||
record_decommission_entry_error(&entry_error, &rx, Error::SlowDown).await;
|
||||
record_decommission_entry_error(&entry_error, &rx, Error::OperationCanceled).await;
|
||||
|
||||
assert!(rx.is_cancelled());
|
||||
assert!(matches!(*entry_error.lock().await, Some(Error::SlowDown)));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_record_decommission_entry_error_ignores_already_canceled_listing() {
|
||||
let entry_error = Arc::new(tokio::sync::Mutex::new(None));
|
||||
let rx = CancellationToken::new();
|
||||
rx.cancel();
|
||||
|
||||
record_decommission_entry_error(&entry_error, &rx, Error::SlowDown).await;
|
||||
|
||||
assert!(entry_error.lock().await.is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_resolve_decommission_listing_error_preserves_real_listing_failure() {
|
||||
let err = resolve_decommission_listing_error(Some(Error::SlowDown), Some(Error::OperationCanceled))
|
||||
.expect("listing failure should be returned");
|
||||
assert!(matches!(err, Error::SlowDown));
|
||||
|
||||
let err = resolve_decommission_listing_error(Some(Error::OperationCanceled), Some(Error::SlowDown))
|
||||
.expect("entry failure should be returned");
|
||||
assert!(matches!(err, Error::SlowDown));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_resolve_decommission_check_after_list_result_returns_list_result_without_entry_error() {
|
||||
let err = resolve_decommission_check_after_list_result(Err(Error::OperationCanceled), None)
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -206,13 +206,6 @@ def check_runner_selection(root: Path) -> list[str]:
|
||||
return errors
|
||||
|
||||
|
||||
def check_s3_tests_runner(root: Path) -> list[str]:
|
||||
runner = (root / "scripts/s3-tests/run.sh").read_text()
|
||||
if "--showlocals" in runner:
|
||||
return ["scripts/s3-tests/run.sh: pytest failure diagnostics must not dump local values"]
|
||||
return []
|
||||
|
||||
|
||||
def profile_selection(root: Path, profile: str) -> str:
|
||||
if not re.fullmatch(r"e2e-[a-z0-9-]+", profile):
|
||||
raise ValueError(f"invalid e2e profile name: {profile}")
|
||||
@@ -279,7 +272,6 @@ def validate(root: Path) -> list[str]:
|
||||
errors.extend(check_e2e_modules(root))
|
||||
errors.extend(check_fuzz_targets(root))
|
||||
errors.extend(check_runner_selection(root))
|
||||
errors.extend(check_s3_tests_runner(root))
|
||||
errors.extend(check_profile_definitions(root))
|
||||
return errors
|
||||
|
||||
@@ -349,23 +341,6 @@ class SelfTests(unittest.TestCase):
|
||||
)
|
||||
self.assertEqual(len(check_fuzz_targets(root)), 1)
|
||||
|
||||
def test_s3_runner_rejects_unbounded_failure_locals(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
runner = root / "scripts/s3-tests/run.sh"
|
||||
runner.parent.mkdir(parents=True)
|
||||
runner.write_text("tox -- -vv -ra --tb=long\n")
|
||||
self.assertEqual(check_s3_tests_runner(root), [])
|
||||
runner.write_text("tox -- -vv -ra --showlocals --tb=long\n")
|
||||
self.assertEqual(len(check_s3_tests_runner(root)), 1)
|
||||
with (
|
||||
mock.patch(__name__ + ".check_e2e_modules", return_value=[]),
|
||||
mock.patch(__name__ + ".check_fuzz_targets", return_value=[]),
|
||||
mock.patch(__name__ + ".check_runner_selection", return_value=[]),
|
||||
mock.patch(__name__ + ".check_profile_definitions", return_value=[]),
|
||||
):
|
||||
self.assertEqual(len(validate(root)), 1)
|
||||
|
||||
def test_profile_listing_enforces_selection(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
@@ -436,7 +411,7 @@ def main() -> int:
|
||||
for error in errors:
|
||||
print(f"ERROR: {error}", file=sys.stderr)
|
||||
return 1
|
||||
print("OK: e2e modules, runner selection, fuzz matrices, profiles, and bounded diagnostics are wired")
|
||||
print("OK: e2e modules, runner selection, fuzz matrices, and profile guards are wired")
|
||||
return 0
|
||||
|
||||
|
||||
|
||||
@@ -1028,11 +1028,10 @@ else
|
||||
fi
|
||||
|
||||
# Run tests from s3tests/functional
|
||||
# Failure locals can contain multi-MiB request bodies; keep tracebacks without expanding local values.
|
||||
set +e
|
||||
S3TEST_CONF="${CONF_OUTPUT_PATH}" \
|
||||
tox -- \
|
||||
-vv -ra --tb=long \
|
||||
-vv -ra --showlocals --tb=long \
|
||||
--maxfail="${MAXFAIL}" \
|
||||
--timeout="${TEST_TIMEOUT}" \
|
||||
--junitxml="${ARTIFACTS_DIR}/junit.xml" \
|
||||
|
||||
Reference in New Issue
Block a user