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 <heihutu@gmail.com>

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?;