feat(ilm): bound manual transition duration (#5174)

* feat(ilm): bound manual transition duration

Add maxDurationSeconds handling for manual transition runs so operators can bound long scoped scans without changing the existing enqueue_only job contract.

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

* docs(ilm): document manual transition run limits

Document the enqueue_only manual transition contract, secret-handling guidance, and best-effort duration budget while pinning the new duration flag in the e2e response model.

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

---------

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

crates/e2e_test/src/reliant/tiering.rs

@@ -345,6 +345,7 @@ struct ManualTransitionRunReport {
     skipped_queue_closed: u64,
     skipped_queue_timeout: u64,
     truncated_by_limit: bool,
+    truncated_by_duration: bool,
 }
 
 async fn manual_transition_run(
@@ -590,6 +591,7 @@ async fn test_manual_transition_run_black_box_semantics() -> TestResult {
     assert_eq!(due.report.skipped_directory, 0);
     assert_eq!(due.report.skipped_replication, 0);
     assert!(!due.report.truncated_by_limit);
+    assert!(!due.report.truncated_by_duration);
     wait_for_transition(&hot_client, MANUAL_DUE_BUCKET, MANUAL_DUE_KEY, StdDuration::from_secs(90)).await?;
     let remote_count_after_due = cold_tier_object_count(&cold_client).await?;
 
@@ -615,6 +617,7 @@ async fn test_manual_transition_run_black_box_semantics() -> TestResult {
     assert_eq!(dry.report.dry_run_eligible, 1, "dry-run report: {:#?}", dry.report);
     assert_eq!(dry.report.enqueued, 0, "dry-run report: {:#?}", dry.report);
     assert_eq!(dry.report.skipped_not_transition, 1, "dry-run report: {:#?}", dry.report);
+    assert!(!dry.report.truncated_by_duration);
     assert_eq!(
         cold_tier_object_count(&cold_client).await?,
         before_dry_run_remote_count,
@@ -639,6 +642,7 @@ async fn test_manual_transition_run_black_box_semantics() -> TestResult {
     assert_eq!(not_due.report.skipped_queue_full, 0);
     assert_eq!(not_due.report.skipped_queue_closed, 0);
     assert_eq!(not_due.report.skipped_queue_timeout, 0);
+    assert!(!not_due.report.truncated_by_duration);
     assert_remains_not_transitioned(&hot_client, MANUAL_NOT_DUE_BUCKET, MANUAL_NOT_DUE_KEY, StdDuration::from_secs(2)).await?;
 
     Ok(())

crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs

@@ -2440,6 +2440,7 @@ pub struct ManualTransitionRunOptions {
     pub tier: Option<String>,
     pub dry_run: bool,
     pub max_objects: Option<u64>,
+    pub max_duration: Option<std::time::Duration>,
 }
 
 #[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize)]
@@ -2463,6 +2464,7 @@ pub struct ManualTransitionRunReport {
     pub skipped_queue_closed: u64,
     pub skipped_queue_timeout: u64,
     pub truncated_by_limit: bool,
+    pub truncated_by_duration: bool,
     #[serde(skip_serializing)]
     pub next_marker: Option<String>,
     #[serde(skip_serializing)]
@@ -2483,6 +2485,10 @@ impl ManualTransitionRunReport {
     pub fn has_partial_enqueue(&self) -> bool {
         self.skipped_queue_full > 0 || self.skipped_queue_closed > 0 || self.skipped_queue_timeout > 0
     }
+
+    pub fn was_truncated(&self) -> bool {
+        self.truncated_by_limit || self.truncated_by_duration
+    }
 }
 
 pub async fn enqueue_transition_for_existing_objects(api: Arc<ECStore>, bucket: &str) -> Result<(), Error> {
@@ -2507,6 +2513,7 @@ pub async fn enqueue_transition_for_existing_objects_scoped(
     let mut previous_marker = marker.clone();
     let mut previous_version_marker = version_marker.clone();
     let src = LcEventSrc::Scanner;
+    let deadline = options.max_duration.map(|duration| tokio::time::Instant::now() + duration);
 
     loop {
         let page = api
@@ -2515,6 +2522,12 @@ pub async fn enqueue_transition_for_existing_objects_scoped(
             .await?;
 
         for (index, object) in page.objects.iter().enumerate() {
+            if manual_transition_duration_elapsed(deadline) {
+                report.truncated_by_duration = true;
+                report.next_marker.clone_from(&previous_marker);
+                report.next_version_idmarker.clone_from(&previous_version_marker);
+                return Ok(report);
+            }
             report.scanned = report.scanned.saturating_add(1);
             enqueue_transition_with_lifecycle_report(object, &lc, &src, &options, &mut report).await;
             if report.has_partial_enqueue() {
@@ -2537,6 +2550,12 @@ pub async fn enqueue_transition_for_existing_objects_scoped(
         if !page.is_truncated {
             return Ok(report);
         }
+        if manual_transition_duration_elapsed(deadline) {
+            report.truncated_by_duration = true;
+            report.next_marker.clone_from(&previous_marker);
+            report.next_version_idmarker.clone_from(&previous_version_marker);
+            return Ok(report);
+        }
 
         marker = page.next_marker;
         version_marker = page.next_version_idmarker;
@@ -2735,6 +2754,10 @@ fn manual_transition_has_more_after_limit(page_index: usize, page_len: usize, pa
     page_index.saturating_add(1) < page_len || page_is_truncated
 }
 
+fn manual_transition_duration_elapsed(deadline: Option<tokio::time::Instant>) -> bool {
+    deadline.is_some_and(|deadline| tokio::time::Instant::now() >= deadline)
+}
+
 fn manual_transition_version_marker(oi: &ObjectInfo) -> String {
     oi.version_id
         .map(|version| version.to_string())
@@ -3652,7 +3675,7 @@ mod tests {
         eval_action_from_lifecycle, jitter_tier_free_version_recovery_delay, lifecycle_action_blocked_by_replication,
         lifecycle_delete_all_versions_replication_scan, lifecycle_deleted_object, lifecycle_replication_blocks_action,
         lifecycle_rule_has_date_expiration, lifecycle_version_purge_state_from_completed_targets,
-        manual_transition_has_more_after_limit, manual_transition_version_marker,
+        manual_transition_duration_elapsed, manual_transition_has_more_after_limit, manual_transition_version_marker,
         mark_delete_opts_skip_decommissioned_on_remote_success, merge_stale_multipart_candidate, replication_state_for_delete,
         resolve_transition_queue_capacity, resolve_transition_queue_send_timeout, resolve_transition_worker_count,
         resolve_transition_workers_absolute_max, run_tier_free_version_recovery_loop, select_restore_s3_location,
@@ -6354,6 +6377,21 @@ mod tests {
         assert!(manual_transition_has_more_after_limit(9, 10, true));
     }
 
+    #[test]
+    fn manual_transition_duration_budget_detects_elapsed_deadline() {
+        let report = ManualTransitionRunReport {
+            truncated_by_duration: true,
+            ..Default::default()
+        };
+
+        assert!(!manual_transition_duration_elapsed(None));
+        assert!(manual_transition_duration_elapsed(Some(tokio::time::Instant::now())));
+        assert!(!manual_transition_duration_elapsed(Some(
+            tokio::time::Instant::now() + StdDuration::from_secs(60)
+        )));
+        assert!(report.was_truncated());
+    }
+
     #[tokio::test]
     async fn existing_object_lifecycle_allows_expired_marker_after_replication_completed() {
         let lc = expired_delete_marker_lifecycle();

docs/operations/tier-ilm-debugging.md

@@ -73,6 +73,31 @@ If both `x-rustfs-internal-transitioned-versionID` and
 `x-minio-internal-transitioned-versionID` are the **empty string**, the object
 was transitioned to a non-versioned tier bucket and no versionId must be sent.
 
+## Manual transition run
+
+Manual transition run is an operator trigger for the existing lifecycle transition evaluator. It does not force objects that are not due under the bucket lifecycle rule, and it does not bypass versioning, replication, delete-marker, directory-marker, tier, or in-flight transition checks.
+
+The admin endpoint is:
+
+```text
+POST /rustfs/admin/v3/ilm/transition/run?bucket=<bucket>&prefix=<prefix>&tier=<tier>&dryRun=true&maxObjects=10000&maxDurationSeconds=30
+```
+
+Only `bucket` is required. `prefix`, `tier`, `dryRun`, `maxObjects`, and `maxDurationSeconds` narrow or bound the run. `maxObjects` defaults to `10000` and is capped at `100000`. `maxDurationSeconds` is optional, capped at `3600`, and enforced as a best-effort budget checked between listed object versions and pages; an in-flight listing call is not cancelled.
+
+The current contract is `enqueue_only`: the response reports what this bounded scan evaluated and enqueued into the in-memory transition queue. `state=completed` means the bounded scan reached the end of its current scope without queue pressure or configured budget truncation. It does not mean every remote tier PUT has completed. `state=partial` means the run stopped early because it hit `maxObjects`, `maxDurationSeconds`, or queue pressure (`skipped_queue_full`, `skipped_queue_closed`, or `skipped_queue_timeout`).
+
+`job_id` and `status_endpoint` are currently `null`. There is no durable background job, no restart recovery cursor, no cluster-wide single-flight admission, no status polling endpoint, and no cancel endpoint for this trigger yet. Re-running the command is safe only in the normal lifecycle sense: already in-flight object versions are deduplicated in the local process, but separate nodes do not yet share a durable manual-run admission record.
+
+Recommended operator flow:
+
+```bash
+rc admin ilm transition run local/mybucket --prefix logs/ --tier cold --dry-run --max-objects 1000 --max-duration-seconds 30
+rc admin ilm transition run local/mybucket --prefix logs/ --tier cold --max-objects 1000 --max-duration-seconds 30
+```
+
+Inspect the aggregate counters before widening scope. Full object-key lists are intentionally not returned by the admin response. If `RUSTFS_RPC_SECRET` or other credentials were pasted into an issue, chat, log, or ticket while debugging tiering, rotate them on every node, restart the cluster with the new value, and redact the exposed copy before sharing more diagnostics.
+
 ## Historical fixes (for context, already merged)
 
 - Expire/GET race (`NoSuchVersion` during expiry of a tiered object):

rustfs/src/admin/handlers/ilm_transition.rs

@@ -32,6 +32,7 @@ use serde::{Deserialize, Serialize};
 const JSON_CONTENT_TYPE: &str = "application/json";
 const DEFAULT_MANUAL_TRANSITION_MAX_OBJECTS: u64 = 10_000;
 const MAX_MANUAL_TRANSITION_OBJECTS: u64 = 100_000;
+const MAX_MANUAL_TRANSITION_DURATION_SECONDS: u64 = 3600;
 
 #[derive(Debug, Deserialize, Default)]
 #[serde(deny_unknown_fields)]
@@ -46,6 +47,8 @@ struct ManualTransitionRunQuery {
     dry_run: Option<bool>,
     #[serde(rename = "maxObjects")]
     max_objects: Option<u64>,
+    #[serde(rename = "maxDurationSeconds")]
+    max_duration_seconds: Option<u64>,
 }
 
 #[derive(Debug, Serialize)]
@@ -63,7 +66,6 @@ pub fn register_ilm_transition_route(r: &mut S3Router<AdminOperation>) -> std::i
         format!("{ADMIN_PREFIX}/v3/ilm/transition/run").as_str(),
         AdminOperation(&ManualTransitionRunHandler {}),
     )?;
-
     Ok(())
 }
 
@@ -88,6 +90,12 @@ fn parse_manual_transition_query(query: Option<&str>) -> S3Result<(String, Manua
     if max_objects == 0 || max_objects > MAX_MANUAL_TRANSITION_OBJECTS {
         return Err(s3_error!(InvalidArgument, "maxObjects is outside the allowed range"));
     }
+    if query
+        .max_duration_seconds
+        .is_some_and(|duration| duration == 0 || duration > MAX_MANUAL_TRANSITION_DURATION_SECONDS)
+    {
+        return Err(s3_error!(InvalidArgument, "maxDurationSeconds is outside the allowed range"));
+    }
 
     Ok((
         bucket.to_string(),
@@ -98,6 +106,7 @@ fn parse_manual_transition_query(query: Option<&str>) -> S3Result<(String, Manua
             tier: query.tier.map(|tier| tier.trim().to_string()).filter(|tier| !tier.is_empty()),
             dry_run: query.dry_run.unwrap_or(false),
             max_objects: Some(max_objects),
+            max_duration: query.max_duration_seconds.map(std::time::Duration::from_secs),
         },
     ))
 }
@@ -126,7 +135,7 @@ async fn authorize_manual_transition_request(req: &S3Request<Body>) -> S3Result<
 }
 
 fn response_state(report: &ManualTransitionRunReport) -> &'static str {
-    if report.truncated_by_limit || report.has_partial_enqueue() {
+    if report.was_truncated() || report.has_partial_enqueue() {
         "partial"
     } else {
         "completed"
@@ -187,6 +196,15 @@ mod tests {
         assert_eq!(options.tier.as_deref(), Some("warm"));
         assert!(!options.dry_run);
         assert_eq!(options.max_objects, Some(DEFAULT_MANUAL_TRANSITION_MAX_OBJECTS));
+        assert_eq!(options.max_duration, None);
+    }
+
+    #[test]
+    fn manual_transition_query_accepts_duration_budget() {
+        let (_bucket, options) =
+            parse_manual_transition_query(Some("bucket=data&maxDurationSeconds=30")).expect("valid query should parse");
+
+        assert_eq!(options.max_duration, Some(std::time::Duration::from_secs(30)));
     }
 
     #[test]
@@ -203,6 +221,18 @@ mod tests {
         assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
     }
 
+    #[test]
+    fn manual_transition_query_rejects_invalid_duration_budget() {
+        let err = parse_manual_transition_query(Some("bucket=data&maxDurationSeconds=0")).expect_err("zero budget must fail");
+
+        assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
+
+        let err = parse_manual_transition_query(Some("bucket=data&maxDurationSeconds=3601"))
+            .expect_err("budget above the cap must fail");
+
+        assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
+    }
+
     #[test]
     fn manual_transition_response_reports_partial_for_queue_pressure() {
         let report = ManualTransitionRunReport {
@@ -213,6 +243,16 @@ mod tests {
         assert_eq!(response_state(&report), "partial");
     }
 
+    #[test]
+    fn manual_transition_response_reports_partial_for_duration_budget() {
+        let report = ManualTransitionRunReport {
+            truncated_by_duration: true,
+            ..Default::default()
+        };
+
+        assert_eq!(response_state(&report), "partial");
+    }
+
     #[test]
     fn manual_transition_response_omits_raw_resume_markers() {
         let report = ManualTransitionRunReport {