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>
This commit is contained in:
houseme
2026-07-24 15:20:36 +08:00
parent 66d5bc1e16
commit 2ead2f5689
2 changed files with 81 additions and 3 deletions
@@ -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();
+42 -2
View File
@@ -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 {