diff --git a/crates/e2e_test/src/reliant/tiering.rs b/crates/e2e_test/src/reliant/tiering.rs index 49211b233..6c7ec1371 100644 --- a/crates/e2e_test/src/reliant/tiering.rs +++ b/crates/e2e_test/src/reliant/tiering.rs @@ -69,6 +69,8 @@ const SOURCE_BUCKET: &str = "ilm7-hot"; const MANUAL_DUE_BUCKET: &str = "ilm7-manual-due"; const MANUAL_DRY_RUN_BUCKET: &str = "ilm7-manual-dry-run"; const MANUAL_NOT_DUE_BUCKET: &str = "ilm7-manual-not-due"; +const MANUAL_QUEUE_PRESSURE_BUCKET: &str = "ilm7-manual-queue-pressure"; +const MANUAL_QUEUE_PRESSURE_PREFIX: &str = "manual-queue-pressure/"; const OBJECT_KEY: &str = "tier/鲁A12345/report.bin"; const MANUAL_DUE_KEY: &str = "manual-due/report.bin"; const MANUAL_DRY_RUN_KEY: &str = "manual-dry-run/report.bin"; @@ -353,12 +355,23 @@ async fn manual_transition_run( bucket: &str, prefix: &str, dry_run: bool, +) -> Result> { + manual_transition_run_with_max(hot, bucket, prefix, dry_run, 10).await +} + +async fn manual_transition_run_with_max( + hot: &RustFSTestEnvironment, + bucket: &str, + prefix: &str, + dry_run: bool, + max_objects: u64, ) -> Result> { let bucket = urlencoding::encode(bucket); let prefix = urlencoding::encode(prefix); let tier = urlencoding::encode(TIER_NAME); - let path = - format!("/rustfs/admin/v3/ilm/transition/run?bucket={bucket}&prefix={prefix}&tier={tier}&dryRun={dry_run}&maxObjects=10"); + let path = format!( + "/rustfs/admin/v3/ilm/transition/run?bucket={bucket}&prefix={prefix}&tier={tier}&dryRun={dry_run}&maxObjects={max_objects}" + ); let (status, body) = signed_admin_request(&hot.url, Method::POST, &path, None, &hot.access_key, &hot.secret_key).await?; if !status.is_success() { return Err(format!("manual transition run failed: status={status}, body={body}").into()); @@ -685,3 +698,62 @@ async fn test_manual_transition_run_contract_no_status_cancel_fields() -> TestRe Ok(()) } + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn test_manual_transition_run_queue_pressure_partial() -> TestResult { + let mut cold = RustFSTestEnvironment::new().await?; + cold.access_key = "manualpressurecoldtieradmin".to_string(); + cold.secret_key = "manualpressurecoldtiersecret".to_string(); + cold.start_rustfs_server_without_cleanup(vec![]).await?; + let cold_client = cold.create_s3_client(); + cold_client.create_bucket().bucket(TIER_BUCKET).send().await?; + + let mut hot = RustFSTestEnvironment::new().await?; + hot.start_rustfs_server_with_env( + vec![], + &[ + ("RUSTFS_SCANNER_ENABLED", "false"), + ("RUSTFS_SCANNER_CYCLE", "3600"), + ("RUSTFS_MAX_TRANSITION_WORKERS", "1"), + ("RUSTFS_TRANSITION_QUEUE_CAPACITY", "1"), + ], + ) + .await?; + let hot_client = hot.create_s3_client(); + add_rustfs_tier(&hot, &cold).await?; + + hot_client.create_bucket().bucket(MANUAL_QUEUE_PRESSURE_BUCKET).send().await?; + for idx in 0..20 { + let key = format!("{MANUAL_QUEUE_PRESSURE_PREFIX}obj-{idx:02}"); + put_single_part_object(&hot_client, MANUAL_QUEUE_PRESSURE_BUCKET, &key, b"queue-pressure payload").await?; + } + put_lifecycle_transition_rule( + &hot_client, + MANUAL_QUEUE_PRESSURE_BUCKET, + "manual-queue-pressure", + MANUAL_QUEUE_PRESSURE_PREFIX, + 0, + ) + .await?; + + let response = + manual_transition_run_with_max(&hot, MANUAL_QUEUE_PRESSURE_BUCKET, MANUAL_QUEUE_PRESSURE_PREFIX, false, 20).await?; + + assert_eq!(response.state, "partial"); + assert_eq!(response.report.bucket, MANUAL_QUEUE_PRESSURE_BUCKET); + assert_eq!(response.report.prefix, MANUAL_QUEUE_PRESSURE_PREFIX); + assert!( + response.report.skipped_queue_full > 0, + "expected queue-pressure path to skip at least one object: {:#?}", + response.report + ); + assert!(!response.report.truncated_by_duration); + assert!(response.report.enqueued < 20, "partial run should not enqueue all items in this setup"); + + let remote_count = cold_tier_object_count(&cold_client).await?; + assert!( + remote_count < 20, + "queue pressure must leave at least one object unqueued, remote_count={remote_count}" + ); + Ok(()) +} diff --git a/crates/protos/src/lib.rs b/crates/protos/src/lib.rs index b3515a608..669145d2b 100644 --- a/crates/protos/src/lib.rs +++ b/crates/protos/src/lib.rs @@ -934,6 +934,9 @@ pub async fn evict_failed_connection_with_log_level(addr: &str, log_level: Conne #[cfg(test)] mod tests { use super::*; + use std::sync::Mutex; + + static INTERNODE_RPC_MSGPACK_ONLY_ENV_LOCK: Mutex<()> = Mutex::new(()); #[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)] struct CompatPayloadField { @@ -1297,6 +1300,8 @@ mod tests { #[test] fn internode_rpc_msgpack_only_reuses_cached_env_until_reset() { + let _guard = INTERNODE_RPC_MSGPACK_ONLY_ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner()); + reset_internode_rpc_msgpack_only_cache(); temp_env::with_vars( @@ -1326,6 +1331,8 @@ mod tests { #[test] fn internode_rpc_msgpack_only_requires_request_and_fleet_confirmation() { + let _guard = INTERNODE_RPC_MSGPACK_ONLY_ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner()); + for (requested, fleet_confirmed, expected) in [ (None, None, false), (Some("true"), None, false),