From de2c337fae13e40d20a703ee05363a8cf6681794 Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 25 Jul 2026 22:05:47 +0800 Subject: [PATCH] test(e2e): cover async transition limit partial (#5233) Cover async manual transition jobs that finish as partial because maxObjects truncates the scan, including terminal cancel idempotence and unchanged report counters. Co-authored-by: heihutu --- crates/e2e_test/src/reliant/tiering.rs | 77 ++++++++++++++++++++++++++ 1 file changed, 77 insertions(+) diff --git a/crates/e2e_test/src/reliant/tiering.rs b/crates/e2e_test/src/reliant/tiering.rs index 3346e344e..3a8be7800 100644 --- a/crates/e2e_test/src/reliant/tiering.rs +++ b/crates/e2e_test/src/reliant/tiering.rs @@ -71,7 +71,9 @@ 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_ASYNC_STATUS_BUCKET: &str = "ilm7-manual-async-status"; +const MANUAL_ASYNC_LIMIT_BUCKET: &str = "ilm7-manual-async-limit"; const MANUAL_QUEUE_PRESSURE_PREFIX: &str = "manual-queue-pressure/"; +const MANUAL_ASYNC_LIMIT_PREFIX: &str = "manual-async-limit/"; 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"; @@ -813,6 +815,81 @@ async fn test_manual_transition_async_job_status_polling() -> TestResult { Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn test_manual_transition_async_limit_reports_terminal_partial() -> TestResult { + let mut cold = RustFSTestEnvironment::new().await?; + cold.access_key = "manualasynclimitcoldtieradmin".to_string(); + cold.secret_key = "manualasynclimitcoldtiersecret".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")]) + .await?; + let hot_client = hot.create_s3_client(); + add_rustfs_tier(&hot, &cold).await?; + + hot_client.create_bucket().bucket(MANUAL_ASYNC_LIMIT_BUCKET).send().await?; + for idx in 0..2 { + let key = format!("{MANUAL_ASYNC_LIMIT_PREFIX}obj-{idx:02}"); + put_single_part_object(&hot_client, MANUAL_ASYNC_LIMIT_BUCKET, &key, b"async limit payload").await?; + } + put_lifecycle_transition_rule(&hot_client, MANUAL_ASYNC_LIMIT_BUCKET, "manual-async-limit", MANUAL_ASYNC_LIMIT_PREFIX, 1) + .await?; + + let before_remote_count = cold_tier_object_count(&cold_client).await?; + let accepted = manual_transition_async_run(&hot, MANUAL_ASYNC_LIMIT_BUCKET, MANUAL_ASYNC_LIMIT_PREFIX, false, 1).await?; + assert_eq!(accepted.state, "accepted"); + assert_eq!(accepted.mode, "durable_job"); + assert_eq!(accepted.report.bucket, MANUAL_ASYNC_LIMIT_BUCKET); + assert_eq!(accepted.report.prefix, MANUAL_ASYNC_LIMIT_PREFIX); + assert_eq!(accepted.report.scanned, 0); + let job_id = accepted.job_id.as_deref().ok_or("async response must include job_id")?; + let status_endpoint = accepted + .status_endpoint + .as_deref() + .ok_or("async response must include status_endpoint")?; + + let terminal = wait_for_manual_transition_job_terminal(&hot, status_endpoint, StdDuration::from_secs(30)).await?; + assert_eq!(terminal.job_id, job_id); + assert_eq!(terminal.status_endpoint, status_endpoint); + assert_eq!(terminal.status, "partial", "terminal limit job response: {terminal:#?}"); + assert!(!terminal.cancel_requested); + assert_eq!(terminal.failure_reason, None); + assert_eq!(terminal.report.bucket, MANUAL_ASYNC_LIMIT_BUCKET); + assert_eq!(terminal.report.prefix, MANUAL_ASYNC_LIMIT_PREFIX); + assert_eq!(terminal.report.tier.as_deref(), Some(TIER_NAME)); + assert!(!terminal.report.dry_run); + assert_eq!(terminal.report.scanned, 1, "terminal limit job response: {terminal:#?}"); + assert_eq!(terminal.report.eligible, 0, "terminal limit job response: {terminal:#?}"); + assert_eq!(terminal.report.enqueued, 0, "terminal limit job response: {terminal:#?}"); + assert_eq!(terminal.report.skipped_not_transition, 1, "terminal limit job response: {terminal:#?}"); + assert_eq!(terminal.report.skipped_queue_full, 0); + assert_eq!(terminal.report.skipped_queue_closed, 0); + assert_eq!(terminal.report.skipped_queue_timeout, 0); + assert!(terminal.report.truncated_by_limit); + assert!(!terminal.report.truncated_by_duration); + + let after_cancel = manual_transition_job_cancel(&hot, status_endpoint).await?; + assert_eq!(after_cancel.status, "partial"); + assert!(!after_cancel.cancel_requested); + assert_eq!(after_cancel.report.bucket, terminal.report.bucket); + assert_eq!(after_cancel.report.prefix, terminal.report.prefix); + assert_eq!(after_cancel.report.tier, terminal.report.tier); + assert_eq!(after_cancel.report.scanned, terminal.report.scanned); + assert_eq!(after_cancel.report.skipped_not_transition, terminal.report.skipped_not_transition); + assert_eq!(after_cancel.report.truncated_by_limit, terminal.report.truncated_by_limit); + + let second_cancel = manual_transition_job_cancel(&hot, status_endpoint).await?; + assert_eq!(second_cancel.status, "partial"); + assert!(!second_cancel.cancel_requested); + assert_eq!(second_cancel.report.scanned, terminal.report.scanned); + assert_eq!(second_cancel.report.truncated_by_limit, terminal.report.truncated_by_limit); + assert_eq!(cold_tier_object_count(&cold_client).await?, before_remote_count); + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn test_manual_transition_run_contract_no_status_cancel_fields() -> TestResult { let mut cold = RustFSTestEnvironment::new().await?;