diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 66818d5d6..692cb025c 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -2440,6 +2440,7 @@ pub struct ManualTransitionRunOptions { pub tier: Option, pub dry_run: bool, pub max_objects: Option, + pub max_duration: Option, } #[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, #[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, 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) -> 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(); diff --git a/rustfs/src/admin/handlers/ilm_transition.rs b/rustfs/src/admin/handlers/ilm_transition.rs index 659d83300..493613e68 100644 --- a/rustfs/src/admin/handlers/ilm_transition.rs +++ b/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, #[serde(rename = "maxObjects")] max_objects: Option, + #[serde(rename = "maxDurationSeconds")] + max_duration_seconds: Option, } #[derive(Debug, Serialize)] @@ -63,7 +66,6 @@ pub fn register_ilm_transition_route(r: &mut S3Router) -> 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) -> 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 {