test(ilm): assert async transition mode contract

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";
@@ -281,12 +283,12 @@ async fn put_multipart_object(client: &Client, bucket: &str, key: &str, data: &[
     Ok(())
 }
 
-async fn put_single_part_object(client: &Client, bucket: &str, key: &str, body: &'static [u8]) -> TestResult {
+async fn put_single_part_object(client: &Client, bucket: &str, key: &str, body: &[u8]) -> TestResult {
     client
         .put_object()
         .bucket(bucket)
         .key(key)
-        .body(ByteStream::from_static(body))
+        .body(ByteStream::from(body.to_vec()))
         .send()
         .await?;
     Ok(())
@@ -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_objects(hot, bucket, prefix, dry_run, 10).await
+}
+
+async fn manual_transition_run_with_max_objects(
+    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 =
-        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());
@@ -647,3 +660,109 @@ async fn test_manual_transition_run_black_box_semantics() -> TestResult {
 
     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}");
+        let body = vec![0x5a; 1024 * 1024];
+        put_single_part_object(&hot_client, MANUAL_QUEUE_PRESSURE_BUCKET, &key, &body).await?;
+    }
+
+    put_lifecycle_transition_rule(
+        &hot_client,
+        MANUAL_QUEUE_PRESSURE_BUCKET,
+        "manual-queue-pressure",
+        MANUAL_QUEUE_PRESSURE_PREFIX,
+        0,
+    )
+    .await?;
+
+    let result =
+        manual_transition_run_with_max_objects(&hot, MANUAL_QUEUE_PRESSURE_BUCKET, MANUAL_QUEUE_PRESSURE_PREFIX, false, 20)
+            .await?;
+    assert_eq!(result.state, "partial");
+    assert!(
+        result.report.skipped_queue_full > 0,
+        "queue-pressure partial should record queue full skips: {:#?}",
+        result.report
+    );
+    assert!(
+        !result.report.truncated_by_duration,
+        "queue-pressure partial should be pressure-driven: {:#?}",
+        result.report
+    );
+
+    let remaining = cold_tier_object_count(&cold_client).await?;
+    assert!(
+        remaining < 20,
+        "queue-pressure partial should skip at least one enqueue, remote count should be <20, got {remaining}"
+    );
+
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn test_manual_transition_run_async_not_implemented() -> TestResult {
+    let mut hot = RustFSTestEnvironment::new().await?;
+    hot.start_rustfs_server_with_env(vec![], &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")])
+        .await?;
+
+    let mode_async =
+        manual_transition_run_with_query(&hot, "/rustfs/admin/v3/ilm/transition/run?bucket=manual-async-test&mode=async").await?;
+    let (mode_async_status, mode_async_body) = mode_async;
+    assert_eq!(
+        mode_async_status,
+        reqwest::StatusCode::NOT_IMPLEMENTED,
+        "mode=async must remain unimplemented"
+    );
+    assert!(
+        mode_async_body.contains("NotImplemented"),
+        "mode=async body must advertise NotImplemented, body: {mode_async_body}"
+    );
+
+    let async_bool =
+        manual_transition_run_with_query(&hot, "/rustfs/admin/v3/ilm/transition/run?bucket=manual-async-test&async=true").await?;
+    let (async_bool_status, async_bool_body) = async_bool;
+    assert_eq!(
+        async_bool_status,
+        reqwest::StatusCode::NOT_IMPLEMENTED,
+        "async=true must remain unimplemented"
+    );
+    assert!(
+        async_bool_body.contains("NotImplemented"),
+        "async=true body must advertise NotImplemented, body: {async_bool_body}"
+    );
+
+    hot.stop_server();
+    Ok(())
+}
+
+async fn manual_transition_run_with_query(
+    hot: &RustFSTestEnvironment,
+    path: &str,
+) -> Result<(reqwest::StatusCode, String), Box<dyn std::error::Error + Send + Sync>> {
+    signed_admin_request(&hot.url, Method::POST, path, None, &hot.access_key, &hot.secret_key).await
+}

rustfs/src/admin/handlers/ilm_transition.rs

@@ -32,6 +32,7 @@ use rustfs_utils::{
 use s3s::header::CONTENT_TYPE;
 use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error};
 use serde::{Deserialize, Serialize};
+use std::sync::{Mutex, MutexGuard, OnceLock};
 use tracing::{error, info, warn};
 
 const JSON_CONTENT_TYPE: &str = "application/json";
@@ -42,6 +43,82 @@ const LOG_COMPONENT_ADMIN: &str = "admin";
 const LOG_SUBSYSTEM_ILM_TRANSITION: &str = "ilm_transition";
 const EVENT_ADMIN_ILM_TRANSITION_STATE: &str = "admin_ilm_transition_state";
 
+static ACTIVE_MANUAL_TRANSITION_SCOPES: OnceLock<Mutex<Vec<ManualTransitionRunScope>>> = OnceLock::new();
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+struct ManualTransitionRunScope {
+    bucket: String,
+    prefix: String,
+    tier: Option<String>,
+    dry_run: bool,
+}
+
+impl ManualTransitionRunScope {
+    fn new(bucket: &str, options: &ManualTransitionRunOptions) -> Self {
+        Self {
+            bucket: bucket.to_string(),
+            prefix: options.prefix.clone(),
+            tier: options.tier.as_ref().map(|tier| tier.to_ascii_uppercase()),
+            dry_run: options.dry_run,
+        }
+    }
+
+    fn overlaps(&self, other: &Self) -> bool {
+        self.bucket == other.bucket
+            && self.dry_run == other.dry_run
+            && prefixes_overlap(&self.prefix, &other.prefix)
+            && match (self.tier.as_deref(), other.tier.as_deref()) {
+                (Some(left), Some(right)) => left == right,
+                _ => true,
+            }
+    }
+}
+
+#[derive(Debug)]
+struct ManualTransitionRunAdmission {
+    scope: ManualTransitionRunScope,
+}
+
+impl Drop for ManualTransitionRunAdmission {
+    fn drop(&mut self) {
+        let mut scopes = lock_active_manual_transition_scopes();
+        if let Some(index) = scopes.iter().position(|scope| scope == &self.scope) {
+            scopes.swap_remove(index);
+        }
+    }
+}
+
+fn active_manual_transition_scopes() -> &'static Mutex<Vec<ManualTransitionRunScope>> {
+    ACTIVE_MANUAL_TRANSITION_SCOPES.get_or_init(|| Mutex::new(Vec::new()))
+}
+
+fn lock_active_manual_transition_scopes() -> MutexGuard<'static, Vec<ManualTransitionRunScope>> {
+    match active_manual_transition_scopes().lock() {
+        Ok(scopes) => scopes,
+        Err(poisoned) => poisoned.into_inner(),
+    }
+}
+
+fn prefixes_overlap(left: &str, right: &str) -> bool {
+    left.starts_with(right) || right.starts_with(left)
+}
+
+fn manual_transition_already_running_error() -> S3Error {
+    s3_error!(
+        OperationAborted,
+        "manual transition run already active for this bucket, prefix, tier, and dry-run mode"
+    )
+}
+
+fn acquire_manual_transition_admission(scope: ManualTransitionRunScope) -> S3Result<ManualTransitionRunAdmission> {
+    let mut scopes = lock_active_manual_transition_scopes();
+    if scopes.iter().any(|active| active.overlaps(&scope)) {
+        return Err(manual_transition_already_running_error());
+    }
+    scopes.push(scope.clone());
+    Ok(ManualTransitionRunAdmission { scope })
+}
+
 #[derive(Debug, Deserialize, Default)]
 #[serde(deny_unknown_fields)]
 struct ManualTransitionRunQuery {
@@ -296,6 +373,14 @@ impl Operation for ManualTransitionRunHandler {
         };
         let max_objects = options.max_objects;
         let max_duration_seconds = options.max_duration.map(|duration| duration.as_secs());
+        let scope = ManualTransitionRunScope::new(&bucket, &options);
+        let _admission = match acquire_manual_transition_admission(scope) {
+            Ok(admission) => admission,
+            Err(err) => {
+                log_manual_transition_rejected("already_running", &request_id, &actor, &remote_addr);
+                return Err(err);
+            }
+        };
 
         let report = match enqueue_transition_for_existing_objects_scoped(store, &bucket, options).await {
             Ok(report) => report,
@@ -392,6 +477,109 @@ mod tests {
         assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
     }
 
+    #[test]
+    fn manual_transition_scope_ignores_resume_and_budget_parameters() {
+        let (bucket, first) = parse_manual_transition_query(Some(
+            "bucket=data&prefix=logs/&tier=warm&marker=logs/a&versionMarker=v1&maxObjects=10",
+        ))
+        .expect("first query should parse");
+        let (_, second) = parse_manual_transition_query(Some(
+            "bucket=data&prefix=logs/&tier=WARM&marker=logs/z&versionMarker=v9&maxObjects=20",
+        ))
+        .expect("second query should parse");
+
+        assert_eq!(
+            ManualTransitionRunScope::new(&bucket, &first),
+            ManualTransitionRunScope::new(&bucket, &second)
+        );
+    }
+
+    #[test]
+    fn manual_transition_scope_distinguishes_dry_run_mode() {
+        let (bucket, real) =
+            parse_manual_transition_query(Some("bucket=data&prefix=logs/&tier=warm")).expect("real query should parse");
+        let (_, dry_run) = parse_manual_transition_query(Some("bucket=data&prefix=logs/&tier=warm&dryRun=true"))
+            .expect("dry-run query should parse");
+
+        assert_ne!(
+            ManualTransitionRunScope::new(&bucket, &real),
+            ManualTransitionRunScope::new(&bucket, &dry_run)
+        );
+    }
+
+    #[test]
+    fn manual_transition_admission_rejects_same_scope_until_guard_drops() {
+        let (bucket, options) =
+            parse_manual_transition_query(Some("bucket=admission-test&prefix=logs/&tier=warm")).expect("query should parse");
+        let scope = ManualTransitionRunScope::new(&bucket, &options);
+        let first = acquire_manual_transition_admission(scope.clone()).expect("first admission should succeed");
+
+        let err = acquire_manual_transition_admission(scope.clone()).expect_err("same scope must be rejected");
+
+        assert_eq!(err.code(), &S3ErrorCode::OperationAborted);
+        assert_eq!(err.status_code(), Some(StatusCode::CONFLICT));
+
+        let different = ManualTransitionRunScope::new(
+            "admission-test",
+            &ManualTransitionRunOptions {
+                prefix: "other/".into(),
+                ..options
+            },
+        );
+        let other = acquire_manual_transition_admission(different).expect("different scope should run independently");
+
+        drop(other);
+        drop(first);
+
+        acquire_manual_transition_admission(scope).expect("scope should be released after guard drops");
+    }
+
+    #[test]
+    fn manual_transition_admission_rejects_overlapping_prefix_or_tier() {
+        let (bucket, options) =
+            parse_manual_transition_query(Some("bucket=admission-overlap-test&prefix=logs/")).expect("query should parse");
+        let scope = ManualTransitionRunScope::new(&bucket, &options);
+        let active = acquire_manual_transition_admission(scope).expect("first admission should succeed");
+
+        let overlapping_prefix = ManualTransitionRunScope::new(
+            "admission-overlap-test",
+            &ManualTransitionRunOptions {
+                prefix: "logs/2026/".into(),
+                tier: Some("warm".into()),
+                ..ManualTransitionRunOptions::default()
+            },
+        );
+        let err =
+            acquire_manual_transition_admission(overlapping_prefix).expect_err("wildcard tier and nested prefix must conflict");
+
+        assert_eq!(err.status_code(), Some(StatusCode::CONFLICT));
+
+        let disjoint_prefix = ManualTransitionRunScope::new(
+            "admission-overlap-test",
+            &ManualTransitionRunOptions {
+                prefix: "archive/".into(),
+                tier: Some("warm".into()),
+                ..ManualTransitionRunOptions::default()
+            },
+        );
+        let disjoint = acquire_manual_transition_admission(disjoint_prefix).expect("disjoint prefix should run independently");
+
+        drop(disjoint);
+        drop(active);
+    }
+
+    #[test]
+    fn manual_transition_handler_acquires_admission_before_enqueue() {
+        let src = include_str!("ilm_transition.rs");
+        let handler_block = extract_block_between_markers(
+            src,
+            "impl Operation for ManualTransitionRunHandler",
+            "let report = match enqueue_transition_for_existing_objects_scoped",
+        );
+
+        assert!(handler_block.contains("acquire_manual_transition_admission"));
+    }
+
     #[test]
     fn manual_transition_query_rejects_server_info_style_unscoped_request() {
         let err = parse_manual_transition_query(Some("dryRun=true")).expect_err("bucket must be required");