Compare commits

..

2 Commits

Author SHA1 Message Date
overtrue a0dcbcda9b style: cargo fmt 2026-08-22 18:48:01 +08:00
overtrue 76e7d979d2 refactor(data-usage): ReplicationStats -> ReplicationTargetUsage
Rename the data-usage crate's ReplicationStats to ReplicationTargetUsage.
Serde field names are byte-identical (only the Rust type name changed;
field identifiers that rmp encodes are untouched). An rmp round-trip test
guards against future drift.

Scanner test imports updated to match.
2026-08-22 18:48:00 +08:00
13 changed files with 108 additions and 61 deletions
+64 -18
View File
@@ -585,9 +585,12 @@ impl VersionsHistogram {
} }
} }
/// Replication statistics for a single target /// Replication statistics for a single target.
#[derive(Debug, Default, Clone, Serialize, Deserialize)] ///
pub struct ReplicationStats { /// 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 pending_size: u64,
pub replicated_size: u64, pub replicated_size: u64,
pub failed_size: u64, pub failed_size: u64,
@@ -600,7 +603,7 @@ pub struct ReplicationStats {
pub replicated_count: u64, pub replicated_count: u64,
} }
impl ReplicationStats { impl ReplicationTargetUsage {
pub fn is_empty(&self) -> bool { pub fn is_empty(&self) -> bool {
let Self { let Self {
pending_size, pending_size,
@@ -636,7 +639,7 @@ impl ReplicationStats {
/// Replication statistics for all targets /// Replication statistics for all targets
#[derive(Debug, Default, Clone, Serialize, Deserialize)] #[derive(Debug, Default, Clone, Serialize, Deserialize)]
pub struct ReplicationAllStats { pub struct ReplicationAllStats {
pub targets: HashMap<String, ReplicationStats>, pub targets: HashMap<String, ReplicationTargetUsage>,
pub replica_size: u64, pub replica_size: u64,
pub replica_count: u64, pub replica_count: u64,
} }
@@ -649,7 +652,7 @@ impl ReplicationAllStats {
targets, targets,
} = self; } = 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")] #[deprecated(note = "use is_empty instead")]
@@ -2466,7 +2469,7 @@ mod tests {
#[test] #[test]
fn replication_stats_empty_checks_every_field() { fn replication_stats_empty_checks_every_field() {
type SetField = fn(&mut ReplicationStats); type SetField = fn(&mut ReplicationTargetUsage);
let cases: [(&str, SetField); 10] = [ let cases: [(&str, SetField); 10] = [
("pending_size", |stats| stats.pending_size = 1), ("pending_size", |stats| stats.pending_size = 1),
@@ -2481,9 +2484,9 @@ mod tests {
("replicated_count", |stats| stats.replicated_count = 1), ("replicated_count", |stats| stats.replicated_count = 1),
]; ];
assert!(ReplicationStats::default().is_empty()); assert!(ReplicationTargetUsage::default().is_empty());
for (field, set_nonzero) in cases { for (field, set_nonzero) in cases {
let mut stats = ReplicationStats::default(); let mut stats = ReplicationTargetUsage::default();
set_nonzero(&mut stats); set_nonzero(&mut stats);
assert!(!stats.is_empty(), "{field} must make replication stats non-empty"); assert!(!stats.is_empty(), "{field} must make replication stats non-empty");
} }
@@ -2514,17 +2517,17 @@ mod tests {
} }
let empty_targets = ReplicationAllStats { 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() ..Default::default()
}; };
assert!(empty_targets.is_empty(), "all-empty targets must keep aggregate stats empty"); assert!(empty_targets.is_empty(), "all-empty targets must keep aggregate stats empty");
let stats = ReplicationAllStats { let stats = ReplicationAllStats {
targets: HashMap::from([ targets: HashMap::from([
("arn:test:empty".to_string(), ReplicationStats::default()), ("arn:test:empty".to_string(), ReplicationTargetUsage::default()),
( (
"arn:test:non-empty".to_string(), "arn:test:non-empty".to_string(),
ReplicationStats { ReplicationTargetUsage {
pending_count: 1, pending_count: 1,
..Default::default() ..Default::default()
}, },
@@ -2565,7 +2568,7 @@ mod tests {
replication_stats: Some(ReplicationAllStats { replication_stats: Some(ReplicationAllStats {
targets: HashMap::from([( targets: HashMap::from([(
"arn:test:pending".to_string(), "arn:test:pending".to_string(),
ReplicationStats { ReplicationTargetUsage {
pending_count: 1, pending_count: 1,
..Default::default() ..Default::default()
}, },
@@ -2714,7 +2717,7 @@ mod tests {
targets: HashMap::from([ targets: HashMap::from([
( (
"arn:self-only".to_string(), "arn:self-only".to_string(),
ReplicationStats { ReplicationTargetUsage {
pending_size: 7, pending_size: 7,
pending_count: 1, pending_count: 1,
..Default::default() ..Default::default()
@@ -2722,7 +2725,7 @@ mod tests {
), ),
( (
"arn:shared".to_string(), "arn:shared".to_string(),
ReplicationStats { ReplicationTargetUsage {
failed_size: 3, failed_size: 3,
failed_count: 1, failed_count: 1,
missed_threshold_size: 2, missed_threshold_size: 2,
@@ -2741,7 +2744,7 @@ mod tests {
targets: HashMap::from([ targets: HashMap::from([
( (
"arn:shared".to_string(), "arn:shared".to_string(),
ReplicationStats { ReplicationTargetUsage {
failed_size: 5, failed_size: 5,
failed_count: 2, failed_count: 2,
after_threshold_size: 4, after_threshold_size: 4,
@@ -2751,7 +2754,7 @@ mod tests {
), ),
( (
"arn:other-only".to_string(), "arn:other-only".to_string(),
ReplicationStats { ReplicationTargetUsage {
replicated_size: 11, replicated_size: 11,
replicated_count: 3, replicated_count: 3,
..Default::default() ..Default::default()
@@ -2993,7 +2996,9 @@ mod tests {
fn replication_target_deserialization_preserves_large_historical_maps() { fn replication_target_deserialization_preserves_large_historical_maps() {
let mut stats = ReplicationAllStats::default(); let mut stats = ReplicationAllStats::default();
for index in 0..=1024 { 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 encoded = rmp_serde::to_vec_named(&stats).expect("large replication target fixture should encode");
let decoded = rmp_serde::from_slice::<ReplicationAllStats>(&encoded) let decoded = rmp_serde::from_slice::<ReplicationAllStats>(&encoded)
@@ -3002,6 +3007,47 @@ mod tests {
assert_eq!(decoded.targets.len(), stats.targets.len()); 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] #[test]
fn checked_merge_rejects_noncanonical_histograms_without_mutation() { fn checked_merge_rejects_noncanonical_histograms_without_mutation() {
let mut entry = DataUsageEntry { let mut entry = DataUsageEntry {
@@ -37,7 +37,7 @@ async fn test_bucket_default_sse_s3_put_object() -> Result<(), Box<dyn std::erro
let mut kms_env = LocalKMSTestEnvironment::new().await?; let mut kms_env = LocalKMSTestEnvironment::new().await?;
let _default_key_id = kms_env.start_rustfs_for_local_kms().await?; let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
kms_env.wait_for_kms_ready().await?; tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
let s3_client = kms_env.base_env.create_s3_client(); let s3_client = kms_env.base_env.create_s3_client();
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?; kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -159,7 +159,7 @@ async fn test_bucket_default_sse_kms_put_object() -> Result<(), Box<dyn std::err
let mut kms_env = LocalKMSTestEnvironment::new().await?; let mut kms_env = LocalKMSTestEnvironment::new().await?;
let default_key_id = kms_env.start_rustfs_for_local_kms().await?; let default_key_id = kms_env.start_rustfs_for_local_kms().await?;
kms_env.wait_for_kms_ready().await?; tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
let s3_client = kms_env.base_env.create_s3_client(); let s3_client = kms_env.base_env.create_s3_client();
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?; kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -278,7 +278,7 @@ async fn test_bucket_default_sse_kms_multipart_crc32() -> Result<(), Box<dyn std
let mut kms_env = LocalKMSTestEnvironment::new().await?; let mut kms_env = LocalKMSTestEnvironment::new().await?;
let default_key_id = kms_env.start_rustfs_for_local_kms().await?; let default_key_id = kms_env.start_rustfs_for_local_kms().await?;
kms_env.wait_for_kms_ready().await?; tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
let s3_client = kms_env.base_env.create_s3_client(); let s3_client = kms_env.base_env.create_s3_client();
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?; kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -475,7 +475,7 @@ async fn test_explicit_encryption_overrides_bucket_default() -> Result<(), Box<d
let mut kms_env = LocalKMSTestEnvironment::new().await?; let mut kms_env = LocalKMSTestEnvironment::new().await?;
let default_key_id = kms_env.start_rustfs_for_local_kms().await?; let default_key_id = kms_env.start_rustfs_for_local_kms().await?;
kms_env.wait_for_kms_ready().await?; tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
let s3_client = kms_env.base_env.create_s3_client(); let s3_client = kms_env.base_env.create_s3_client();
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?; kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -570,7 +570,7 @@ async fn test_sse_kms_without_key_id_populates_default() -> Result<(), Box<dyn s
let mut kms_env = LocalKMSTestEnvironment::new().await?; let mut kms_env = LocalKMSTestEnvironment::new().await?;
let default_key_id = kms_env.start_rustfs_for_local_kms().await?; let default_key_id = kms_env.start_rustfs_for_local_kms().await?;
kms_env.wait_for_kms_ready().await?; tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
let s3_client = kms_env.base_env.create_s3_client(); let s3_client = kms_env.base_env.create_s3_client();
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?; kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -61,7 +61,7 @@ async fn test_metadata_replace_self_copy_of_sse_object_stays_decryptable() {
) )
.await .await
.expect("failed to start RustFS with local KMS"); .expect("failed to start RustFS with local KMS");
kms_env.wait_for_kms_ready().await.expect("KMS ready"); tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
let client = kms_env.base_env.create_s3_client(); let client = kms_env.base_env.create_s3_client();
// Deliberately an UNVERSIONED bucket: that is the branch where the store layer can service // Deliberately an UNVERSIONED bucket: that is the branch where the store layer can service
@@ -160,7 +160,7 @@ async fn test_metadata_replace_self_copy_dropping_sse_rewrites_plaintext() {
) )
.await .await
.expect("failed to start RustFS with local KMS"); .expect("failed to start RustFS with local KMS");
kms_env.wait_for_kms_ready().await.expect("KMS ready"); tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
let client = kms_env.base_env.create_s3_client(); let client = kms_env.base_env.create_s3_client();
// Unversioned, and deliberately WITHOUT a bucket default-encryption rule, so the copy below // Unversioned, and deliberately WITHOUT a bucket default-encryption rule, so the copy below
@@ -256,7 +256,7 @@ async fn test_metadata_replace_self_copy_under_bucket_default_sse_stays_decrypta
) )
.await .await
.expect("failed to start RustFS with local KMS"); .expect("failed to start RustFS with local KMS");
kms_env.wait_for_kms_ready().await.expect("KMS ready"); tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
let client = kms_env.base_env.create_s3_client(); let client = kms_env.base_env.create_s3_client();
let bucket = "copy-object-self-copy-bucket-default-sse-test"; let bucket = "copy-object-self-copy-bucket-default-sse-test";
@@ -56,7 +56,7 @@ async fn test_self_copy_of_historical_sse_s3_version_is_readable() {
) )
.await .await
.expect("failed to start RustFS with local KMS"); .expect("failed to start RustFS with local KMS");
kms_env.wait_for_kms_ready().await.expect("KMS ready"); tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
let client = kms_env.base_env.create_s3_client(); let client = kms_env.base_env.create_s3_client();
let bucket = "copy-object-version-restore-sse-test"; let bucket = "copy-object-version-restore-sse-test";
@@ -87,7 +87,7 @@ async fn test_head_reports_managed_metadata_for_sse_s3() -> Result<(), Box<dyn s
let mut kms_env = LocalKMSTestEnvironment::new().await?; let mut kms_env = LocalKMSTestEnvironment::new().await?;
let _default_key = kms_env.start_rustfs_for_local_kms().await?; let _default_key = kms_env.start_rustfs_for_local_kms().await?;
kms_env.wait_for_kms_ready().await?; tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
let s3_client = kms_env.base_env.create_s3_client(); let s3_client = kms_env.base_env.create_s3_client();
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?; kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -147,7 +147,7 @@ async fn test_head_reports_managed_metadata_for_sse_kms_and_copy() -> Result<(),
let mut kms_env = LocalKMSTestEnvironment::new().await?; let mut kms_env = LocalKMSTestEnvironment::new().await?;
let default_key_id = kms_env.start_rustfs_for_local_kms().await?; let default_key_id = kms_env.start_rustfs_for_local_kms().await?;
kms_env.wait_for_kms_ready().await?; tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
let s3_client = kms_env.base_env.create_s3_client(); let s3_client = kms_env.base_env.create_s3_client();
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?; kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -250,7 +250,7 @@ async fn test_multipart_upload_writes_encrypted_data() -> Result<(), Box<dyn std
let mut kms_env = LocalKMSTestEnvironment::new().await?; let mut kms_env = LocalKMSTestEnvironment::new().await?;
let default_key_id = kms_env.start_rustfs_for_local_kms().await?; let default_key_id = kms_env.start_rustfs_for_local_kms().await?;
kms_env.wait_for_kms_ready().await?; tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
let s3_client = kms_env.base_env.create_s3_client(); let s3_client = kms_env.base_env.create_s3_client();
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?; kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -24,6 +24,7 @@ use super::common::{
test_sse_kms_encryption, test_sse_s3_encryption, test_sse_kms_encryption, test_sse_s3_encryption,
}; };
use crate::common::{TEST_BUCKET, init_logging}; use crate::common::{TEST_BUCKET, init_logging};
use tokio::time::{Duration, sleep};
use tracing::info; use tracing::info;
/// Comprehensive test: Full KMS workflow with all encryption types /// Comprehensive test: Full KMS workflow with all encryption types
@@ -34,7 +35,7 @@ async fn test_comprehensive_kms_full_workflow() -> Result<(), Box<dyn std::error
let mut kms_env = LocalKMSTestEnvironment::new().await?; let mut kms_env = LocalKMSTestEnvironment::new().await?;
let _default_key_id = kms_env.start_rustfs_for_local_kms().await?; let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
kms_env.wait_for_kms_ready().await?; sleep(Duration::from_secs(3)).await;
let s3_client = kms_env.base_env.create_s3_client(); let s3_client = kms_env.base_env.create_s3_client();
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?; kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -102,7 +103,7 @@ async fn test_comprehensive_stress_test() -> Result<(), Box<dyn std::error::Erro
let mut kms_env = LocalKMSTestEnvironment::new().await?; let mut kms_env = LocalKMSTestEnvironment::new().await?;
let _default_key_id = kms_env.start_rustfs_for_local_kms().await?; let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
kms_env.wait_for_kms_ready().await?; sleep(Duration::from_secs(3)).await;
let s3_client = kms_env.base_env.create_s3_client(); let s3_client = kms_env.base_env.create_s3_client();
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?; kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -136,7 +137,7 @@ async fn test_comprehensive_key_isolation() -> Result<(), Box<dyn std::error::Er
let mut kms_env = LocalKMSTestEnvironment::new().await?; let mut kms_env = LocalKMSTestEnvironment::new().await?;
let _default_key_id = kms_env.start_rustfs_for_local_kms().await?; let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
kms_env.wait_for_kms_ready().await?; sleep(Duration::from_secs(3)).await;
let s3_client = kms_env.base_env.create_s3_client(); let s3_client = kms_env.base_env.create_s3_client();
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?; kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -207,7 +208,7 @@ async fn test_comprehensive_concurrent_operations() -> Result<(), Box<dyn std::e
let mut kms_env = LocalKMSTestEnvironment::new().await?; let mut kms_env = LocalKMSTestEnvironment::new().await?;
let _default_key_id = kms_env.start_rustfs_for_local_kms().await?; let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
kms_env.wait_for_kms_ready().await?; sleep(Duration::from_secs(3)).await;
let s3_client = kms_env.base_env.create_s3_client(); let s3_client = kms_env.base_env.create_s3_client();
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?; kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -252,7 +253,7 @@ async fn test_comprehensive_performance_benchmark() -> Result<(), Box<dyn std::e
let mut kms_env = LocalKMSTestEnvironment::new().await?; let mut kms_env = LocalKMSTestEnvironment::new().await?;
let _default_key_id = kms_env.start_rustfs_for_local_kms().await?; let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
kms_env.wait_for_kms_ready().await?; sleep(Duration::from_secs(3)).await;
let s3_client = kms_env.base_env.create_s3_client(); let s3_client = kms_env.base_env.create_s3_client();
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?; kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -44,7 +44,7 @@ async fn test_kms_zero_byte_file_encryption() -> Result<(), Box<dyn std::error::
let mut kms_env = LocalKMSTestEnvironment::new().await?; let mut kms_env = LocalKMSTestEnvironment::new().await?;
let _default_key_id = kms_env.start_rustfs_for_local_kms().await?; let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
kms_env.wait_for_kms_ready().await?; tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
let s3_client = kms_env.base_env.create_s3_client(); let s3_client = kms_env.base_env.create_s3_client();
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?; kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -117,7 +117,7 @@ async fn test_kms_single_byte_file_encryption() -> Result<(), Box<dyn std::error
let mut kms_env = LocalKMSTestEnvironment::new().await?; let mut kms_env = LocalKMSTestEnvironment::new().await?;
let _default_key_id = kms_env.start_rustfs_for_local_kms().await?; let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
kms_env.wait_for_kms_ready().await?; tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
let s3_client = kms_env.base_env.create_s3_client(); let s3_client = kms_env.base_env.create_s3_client();
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?; kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -209,7 +209,7 @@ async fn test_kms_multipart_boundary_conditions() -> Result<(), Box<dyn std::err
let mut kms_env = LocalKMSTestEnvironment::new().await?; let mut kms_env = LocalKMSTestEnvironment::new().await?;
let _default_key_id = kms_env.start_rustfs_for_local_kms().await?; let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
kms_env.wait_for_kms_ready().await?; tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
let s3_client = kms_env.base_env.create_s3_client(); let s3_client = kms_env.base_env.create_s3_client();
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?; kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -284,7 +284,7 @@ async fn test_kms_invalid_key_scenarios() -> Result<(), Box<dyn std::error::Erro
let mut kms_env = LocalKMSTestEnvironment::new().await?; let mut kms_env = LocalKMSTestEnvironment::new().await?;
let _default_key_id = kms_env.start_rustfs_for_local_kms().await?; let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
kms_env.wait_for_kms_ready().await?; tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
let s3_client = kms_env.base_env.create_s3_client(); let s3_client = kms_env.base_env.create_s3_client();
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?; kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -371,7 +371,7 @@ async fn test_kms_concurrent_encryption() -> Result<(), Box<dyn std::error::Erro
let mut kms_env = LocalKMSTestEnvironment::new().await?; let mut kms_env = LocalKMSTestEnvironment::new().await?;
let _default_key_id = kms_env.start_rustfs_for_local_kms().await?; let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
kms_env.wait_for_kms_ready().await?; tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
let s3_client = Arc::new(kms_env.base_env.create_s3_client()); let s3_client = Arc::new(kms_env.base_env.create_s3_client());
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?; kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -478,7 +478,7 @@ async fn test_kms_key_validation_security() -> Result<(), Box<dyn std::error::Er
let mut kms_env = LocalKMSTestEnvironment::new().await?; let mut kms_env = LocalKMSTestEnvironment::new().await?;
let _default_key_id = kms_env.start_rustfs_for_local_kms().await?; let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
kms_env.wait_for_kms_ready().await?; tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
let s3_client = kms_env.base_env.create_s3_client(); let s3_client = kms_env.base_env.create_s3_client();
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?; kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -37,7 +37,7 @@ async fn test_kms_key_directory_unavailable() -> Result<(), Box<dyn std::error::
let mut kms_env = LocalKMSTestEnvironment::new().await?; let mut kms_env = LocalKMSTestEnvironment::new().await?;
let _default_key_id = kms_env.start_rustfs_for_local_kms().await?; let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
kms_env.wait_for_kms_ready().await?; tokio::time::sleep(Duration::from_secs(3)).await;
let s3_client = kms_env.base_env.create_s3_client(); let s3_client = kms_env.base_env.create_s3_client();
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?; kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -127,7 +127,7 @@ async fn test_kms_corrupted_key_files() -> Result<(), Box<dyn std::error::Error
let mut kms_env = LocalKMSTestEnvironment::new().await?; let mut kms_env = LocalKMSTestEnvironment::new().await?;
let default_key_id = kms_env.start_rustfs_for_local_kms().await?; let default_key_id = kms_env.start_rustfs_for_local_kms().await?;
kms_env.wait_for_kms_ready().await?; tokio::time::sleep(Duration::from_secs(3)).await;
let s3_client = kms_env.base_env.create_s3_client(); let s3_client = kms_env.base_env.create_s3_client();
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?; kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -218,7 +218,7 @@ async fn test_kms_multipart_upload_interruption() -> Result<(), Box<dyn std::err
let mut kms_env = LocalKMSTestEnvironment::new().await?; let mut kms_env = LocalKMSTestEnvironment::new().await?;
let _default_key_id = kms_env.start_rustfs_for_local_kms().await?; let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
kms_env.wait_for_kms_ready().await?; tokio::time::sleep(Duration::from_secs(3)).await;
let s3_client = kms_env.base_env.create_s3_client(); let s3_client = kms_env.base_env.create_s3_client();
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?; kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -401,7 +401,7 @@ async fn test_kms_resource_constraints() -> Result<(), Box<dyn std::error::Error
let mut kms_env = LocalKMSTestEnvironment::new().await?; let mut kms_env = LocalKMSTestEnvironment::new().await?;
let _default_key_id = kms_env.start_rustfs_for_local_kms().await?; let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
kms_env.wait_for_kms_ready().await?; tokio::time::sleep(Duration::from_secs(3)).await;
let s3_client = kms_env.base_env.create_s3_client(); let s3_client = kms_env.base_env.create_s3_client();
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?; kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
+4 -4
View File
@@ -46,7 +46,7 @@ async fn test_local_kms_end_to_end() -> Result<(), Box<dyn std::error::Error + S
.expect("Failed to start RustFS with Local KMS"); .expect("Failed to start RustFS with Local KMS");
// Wait a moment for RustFS to fully start up and initialize KMS // Wait a moment for RustFS to fully start up and initialize KMS
kms_env.wait_for_kms_ready().await?; tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
info!("RustFS started with KMS auto-configuration, default_key_id: {}", default_key_id); info!("RustFS started with KMS auto-configuration, default_key_id: {}", default_key_id);
@@ -127,7 +127,7 @@ async fn test_local_kms_key_isolation() {
.expect("Failed to start RustFS with Local KMS"); .expect("Failed to start RustFS with Local KMS");
// Wait a moment for RustFS to fully start up and initialize KMS // Wait a moment for RustFS to fully start up and initialize KMS
kms_env.wait_for_kms_ready().await.expect("KMS ready"); tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
info!("RustFS started with KMS auto-configuration, default_key_id: {}", default_key_id); info!("RustFS started with KMS auto-configuration, default_key_id: {}", default_key_id);
@@ -227,7 +227,7 @@ async fn test_local_kms_large_file() {
.expect("Failed to start RustFS with Local KMS"); .expect("Failed to start RustFS with Local KMS");
// Wait a moment for RustFS to fully start up and initialize KMS // Wait a moment for RustFS to fully start up and initialize KMS
kms_env.wait_for_kms_ready().await.expect("KMS ready"); tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
info!("RustFS started with KMS auto-configuration, default_key_id: {}", default_key_id); info!("RustFS started with KMS auto-configuration, default_key_id: {}", default_key_id);
@@ -309,7 +309,7 @@ async fn test_local_kms_multipart_upload() {
.expect("Failed to start RustFS with Local KMS"); .expect("Failed to start RustFS with Local KMS");
// Wait for KMS initialization // Wait for KMS initialization
kms_env.wait_for_kms_ready().await.expect("KMS ready"); tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
info!("RustFS started with KMS auto-configuration, default_key_id: {}", default_key_id); info!("RustFS started with KMS auto-configuration, default_key_id: {}", default_key_id);
+3 -3
View File
@@ -19,7 +19,7 @@
//! multipart upload behaviour. //! multipart upload behaviour.
use crate::common::{TEST_BUCKET, init_logging}; use crate::common::{TEST_BUCKET, init_logging};
use serial_test::serial; use tokio::time::{Duration, sleep};
use tracing::{error, info}; use tracing::{error, info};
use super::common::{ use super::common::{
@@ -45,8 +45,8 @@ impl VaultKmsTestContext {
start_kms(&env.base_env.url, &env.base_env.access_key, &env.base_env.secret_key).await?; start_kms(&env.base_env.url, &env.base_env.access_key, &env.base_env.secret_key).await?;
// Wait for KMS to finish initialising. // Allow Vault to finish initialising token auth and transit engine.
super::common::wait_for_kms_ready(&env.base_env.url, &env.base_env.access_key, &env.base_env.secret_key).await?; sleep(Duration::from_secs(2)).await;
Ok(Self { env }) Ok(Self { env })
} }
@@ -33,7 +33,7 @@ async fn test_step1_basic_single_file_encryption() -> Result<(), Box<dyn std::er
let mut kms_env = LocalKMSTestEnvironment::new().await?; let mut kms_env = LocalKMSTestEnvironment::new().await?;
let _default_key_id = kms_env.start_rustfs_for_local_kms().await?; let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
kms_env.wait_for_kms_ready().await?; tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
let s3_client = kms_env.base_env.create_s3_client(); let s3_client = kms_env.base_env.create_s3_client();
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?; kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -89,7 +89,7 @@ async fn test_step2_basic_multipart_upload_without_encryption() -> Result<(), Bo
let mut kms_env = LocalKMSTestEnvironment::new().await?; let mut kms_env = LocalKMSTestEnvironment::new().await?;
let _default_key_id = kms_env.start_rustfs_for_local_kms().await?; let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
kms_env.wait_for_kms_ready().await?; tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
let s3_client = kms_env.base_env.create_s3_client(); let s3_client = kms_env.base_env.create_s3_client();
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?; kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -187,7 +187,7 @@ async fn test_step3_multipart_upload_with_sse_s3() -> Result<(), Box<dyn std::er
let mut kms_env = LocalKMSTestEnvironment::new().await?; let mut kms_env = LocalKMSTestEnvironment::new().await?;
let _default_key_id = kms_env.start_rustfs_for_local_kms().await?; let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
kms_env.wait_for_kms_ready().await?; tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
let s3_client = kms_env.base_env.create_s3_client(); let s3_client = kms_env.base_env.create_s3_client();
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?; kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -310,7 +310,7 @@ async fn test_step4_large_multipart_upload_with_encryption() -> Result<(), Box<d
let mut kms_env = LocalKMSTestEnvironment::new().await?; let mut kms_env = LocalKMSTestEnvironment::new().await?;
let _default_key_id = kms_env.start_rustfs_for_local_kms().await?; let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
kms_env.wait_for_kms_ready().await?; tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
let s3_client = kms_env.base_env.create_s3_client(); let s3_client = kms_env.base_env.create_s3_client();
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?; kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -435,7 +435,7 @@ async fn test_step5_all_encryption_types_multipart() -> Result<(), Box<dyn std::
let mut kms_env = LocalKMSTestEnvironment::new().await?; let mut kms_env = LocalKMSTestEnvironment::new().await?;
let _default_key_id = kms_env.start_rustfs_for_local_kms().await?; let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
kms_env.wait_for_kms_ready().await?; tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
let s3_client = kms_env.base_env.create_s3_client(); let s3_client = kms_env.base_env.create_s3_client();
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?; kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -16,7 +16,7 @@ use super::persistence::DataUsageCacheLoadAttempt;
use super::*; use super::*;
use crate::storage_api::scanner_io::{HTTPRangeSpec, ObjectIO}; use crate::storage_api::scanner_io::{HTTPRangeSpec, ObjectIO};
use crate::{ScannerGetObjectReader, ScannerPutObjReader}; use crate::{ScannerGetObjectReader, ScannerPutObjReader};
use rustfs_data_usage::{ReplicationAllStats, ReplicationStats}; use rustfs_data_usage::{ReplicationAllStats, ReplicationTargetUsage};
use serde_json::Value; use serde_json::Value;
use std::io::Cursor; use std::io::Cursor;
use std::pin::Pin; use std::pin::Pin;
@@ -1636,7 +1636,7 @@ fn size_recursive_prunes_empty_and_preserves_threshold_replication_stats() {
replication_stats: Some(ReplicationAllStats { replication_stats: Some(ReplicationAllStats {
targets: HashMap::from([( targets: HashMap::from([(
"arn:test:threshold".to_string(), "arn:test:threshold".to_string(),
ReplicationStats { ReplicationTargetUsage {
after_threshold_count: 1, after_threshold_count: 1,
..Default::default() ..Default::default()
}, },
@@ -13,7 +13,7 @@
// limitations under the License. // limitations under the License.
use super::*; use super::*;
use rustfs_data_usage::{ReplicationAllStats, ReplicationStats}; use rustfs_data_usage::{ReplicationAllStats, ReplicationTargetUsage};
const TEST_PLAN_DIGEST: DataUsageScanPlanDigest = DataUsageScanPlanDigest([7; 32]); 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 { replication_stats: Some(ReplicationAllStats {
targets: HashMap::from([( targets: HashMap::from([(
"arn:target".to_string(), "arn:target".to_string(),
ReplicationStats { ReplicationTargetUsage {
replicated_size: 2048, replicated_size: 2048,
replicated_count: 2, replicated_count: 2,
..Default::default() ..Default::default()