test(ilm): cover manual transition e2e gaps (#5219)

* test(ilm): cover manual transition queue pressure

Co-Authored-By: heihutu <heihutu@gmail.com>

* test(protos): serialize msgpack-only env tests

Co-Authored-By: heihutu <heihutu@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>

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<ManualTransitionRunResponse, Box<dyn std::error::Error + Send + Sync>> {
+    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<ManualTransitionRunResponse, Box<dyn std::error::Error + Send + Sync>> {
     let bucket = urlencoding::encode(bucket);
     let prefix = urlencoding::encode(prefix);
     let tier = urlencoding::encode(TIER_NAME);
-    let path =
+    let path = format!(
-        format!("/rustfs/admin/v3/ilm/transition/run?bucket={bucket}&prefix={prefix}&tier={tier}&dryRun={dry_run}&maxObjects=10");
+        "/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(())
+}

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),