From 77b5e1b64c2c4f8d19758f1e4356df53f0eab07b Mon Sep 17 00:00:00 2001 From: houseme Date: Sun, 26 Jul 2026 02:02:37 +0800 Subject: [PATCH] fix(ilm): report manual transition tier failures (#5238) Co-authored-by: heihutu --- crates/e2e_test/src/reliant/tiering.rs | 235 ++++++++++++++ .../bucket/lifecycle/bucket_lifecycle_ops.rs | 296 ++++++++++++++++-- .../bucket/lifecycle/manual_transition_job.rs | 206 +++++++++++- rustfs/src/admin/handlers/ilm_transition.rs | 70 ++++- 4 files changed, 753 insertions(+), 54 deletions(-) diff --git a/crates/e2e_test/src/reliant/tiering.rs b/crates/e2e_test/src/reliant/tiering.rs index 38cd30fc9..3cf8dfff4 100644 --- a/crates/e2e_test/src/reliant/tiering.rs +++ b/crates/e2e_test/src/reliant/tiering.rs @@ -74,16 +74,22 @@ const MANUAL_ASYNC_STATUS_BUCKET: &str = "ilm7-manual-async-status"; const MANUAL_CONTINUATION_BUCKET: &str = "ilm7-manual-continuation"; const MANUAL_ASYNC_LIMIT_BUCKET: &str = "ilm7-manual-async-limit"; const MANUAL_ASYNC_CONFLICT_BUCKET: &str = "ilm7-manual-async-conflict"; +const MANUAL_TIER_FAILURE_BUCKET: &str = "ilm7-manual-tier-failure"; +const MANUAL_WORKER_FAILURE_BUCKET: &str = "ilm7-manual-worker-failure"; const MANUAL_QUEUE_PRESSURE_PREFIX: &str = "manual-queue-pressure/"; const MANUAL_CONTINUATION_PREFIX: &str = "manual-continuation/"; const MANUAL_ASYNC_LIMIT_PREFIX: &str = "manual-async-limit/"; const MANUAL_ASYNC_CONFLICT_PREFIX: &str = "manual-async-conflict/"; const MANUAL_ASYNC_CONFLICT_NESTED_PREFIX: &str = "manual-async-conflict/nested/"; +const MANUAL_TIER_FAILURE_PREFIX: &str = "manual-tier-failure/"; +const MANUAL_WORKER_FAILURE_PREFIX: &str = "manual-worker-failure/"; 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"; const MANUAL_NOT_DUE_KEY: &str = "manual-not-due/report.bin"; const MANUAL_ASYNC_STATUS_KEY: &str = "manual-async-status/report.bin"; +const MANUAL_TIER_FAILURE_KEY: &str = "manual-tier-failure/report.bin"; +const MANUAL_WORKER_FAILURE_KEY: &str = "manual-worker-failure/report.bin"; const CONTENT_TYPE: &str = "application/x-ilm7"; const USER_META_KEY: &str = "ilm7-origin"; const USER_META_VAL: &str = "hermetic-transition"; @@ -177,6 +183,15 @@ async fn add_rustfs_tier(hot: &RustFSTestEnvironment, cold: &RustFSTestEnvironme Ok(()) } +async fn remove_rustfs_tier_force(hot: &RustFSTestEnvironment) -> TestResult { + let path = format!("/rustfs/admin/v3/tier/{TIER_NAME}?force=true"); + let (status, resp) = signed_admin_request(&hot.url, Method::DELETE, &path, None, &hot.access_key, &hot.secret_key).await?; + if !status.is_success() { + return Err(format!("RemoveTier(RustFS) failed: status={status}, body={resp}").into()); + } + Ok(()) +} + /// A current-version `Transition Days=0` rule scoped to the object's prefix. fn transition_rule() -> Result> { transition_rule_for("ilm7-transition", "tier/", 0) @@ -355,6 +370,14 @@ struct ManualTransitionRunReport { skipped_queue_full: u64, skipped_queue_closed: u64, skipped_queue_timeout: u64, + #[serde(default)] + transition_completed: u64, + #[serde(default)] + transition_failed: u64, + #[serde(default)] + tier_failure: u64, + #[serde(default)] + cancelled: bool, truncated_by_limit: bool, truncated_by_duration: bool, continuation_token: Option, @@ -718,6 +741,8 @@ async fn test_manual_transition_run_black_box_semantics() -> TestResult { assert_eq!(due.report.skipped_delete_marker, 0); assert_eq!(due.report.skipped_directory, 0); assert_eq!(due.report.skipped_replication, 0); + assert_eq!(due.report.tier_failure, 0); + assert!(!due.report.cancelled); 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?; @@ -745,6 +770,8 @@ 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_eq!(dry.report.tier_failure, 0); + assert!(!dry.report.cancelled); assert!(!dry.report.truncated_by_duration); assert_eq!( cold_tier_object_count(&cold_client).await?, @@ -767,6 +794,8 @@ async fn test_manual_transition_run_black_box_semantics() -> TestResult { assert_eq!(not_due.report.eligible, 0, "not-due report: {:#?}", not_due.report); assert_eq!(not_due.report.enqueued, 0, "not-due report: {:#?}", not_due.report); assert_eq!(not_due.report.skipped_not_transition, 1, "not-due report: {:#?}", not_due.report); + assert_eq!(not_due.report.tier_failure, 0); + assert!(!not_due.report.cancelled); 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); @@ -811,6 +840,8 @@ async fn test_manual_transition_async_job_status_polling() -> TestResult { assert!(accepted.report.dry_run); assert_eq!(accepted.report.scanned, 0); assert_eq!(accepted.report.eligible, 0); + assert_eq!(accepted.report.transition_completed, 0); + assert_eq!(accepted.report.transition_failed, 0); let job_id = accepted.job_id.as_deref().ok_or("async response must include job_id")?; let status_endpoint = accepted .status_endpoint @@ -840,6 +871,10 @@ async fn test_manual_transition_async_job_status_polling() -> TestResult { assert_eq!(terminal.report.skipped_queue_full, 0); assert_eq!(terminal.report.skipped_queue_closed, 0); assert_eq!(terminal.report.skipped_queue_timeout, 0); + assert_eq!(terminal.report.transition_completed, 0); + assert_eq!(terminal.report.transition_failed, 0); + assert_eq!(terminal.report.tier_failure, 0); + assert!(!terminal.report.cancelled); assert!(!terminal.report.truncated_by_limit); assert!(!terminal.report.truncated_by_duration); @@ -886,6 +921,8 @@ async fn test_manual_transition_async_limit_reports_terminal_partial() -> TestRe assert_eq!(accepted.report.bucket, MANUAL_ASYNC_LIMIT_BUCKET); assert_eq!(accepted.report.prefix, MANUAL_ASYNC_LIMIT_PREFIX); assert_eq!(accepted.report.scanned, 0); + assert_eq!(accepted.report.transition_completed, 0); + assert_eq!(accepted.report.transition_failed, 0); let job_id = accepted.job_id.as_deref().ok_or("async response must include job_id")?; let status_endpoint = accepted .status_endpoint @@ -909,6 +946,10 @@ async fn test_manual_transition_async_limit_reports_terminal_partial() -> TestRe assert_eq!(terminal.report.skipped_queue_full, 0); assert_eq!(terminal.report.skipped_queue_closed, 0); assert_eq!(terminal.report.skipped_queue_timeout, 0); + assert_eq!(terminal.report.transition_completed, 0); + assert_eq!(terminal.report.transition_failed, 0); + assert_eq!(terminal.report.tier_failure, 0); + assert!(!terminal.report.cancelled); assert!(terminal.report.truncated_by_limit); assert!(!terminal.report.truncated_by_duration); @@ -920,12 +961,20 @@ async fn test_manual_transition_async_limit_reports_terminal_partial() -> TestRe assert_eq!(after_cancel.report.tier, terminal.report.tier); assert_eq!(after_cancel.report.scanned, terminal.report.scanned); assert_eq!(after_cancel.report.skipped_not_transition, terminal.report.skipped_not_transition); + assert_eq!(after_cancel.report.transition_completed, terminal.report.transition_completed); + assert_eq!(after_cancel.report.transition_failed, terminal.report.transition_failed); + assert_eq!(after_cancel.report.tier_failure, terminal.report.tier_failure); + assert_eq!(after_cancel.report.cancelled, terminal.report.cancelled); assert_eq!(after_cancel.report.truncated_by_limit, terminal.report.truncated_by_limit); let second_cancel = manual_transition_job_cancel(&hot, status_endpoint).await?; assert_eq!(second_cancel.status, "partial"); assert!(!second_cancel.cancel_requested); assert_eq!(second_cancel.report.scanned, terminal.report.scanned); + assert_eq!(second_cancel.report.transition_completed, terminal.report.transition_completed); + assert_eq!(second_cancel.report.transition_failed, terminal.report.transition_failed); + assert_eq!(second_cancel.report.tier_failure, terminal.report.tier_failure); + assert_eq!(second_cancel.report.cancelled, terminal.report.cancelled); assert_eq!(second_cancel.report.truncated_by_limit, terminal.report.truncated_by_limit); assert_eq!(cold_tier_object_count(&cold_client).await?, before_remote_count); Ok(()) @@ -1011,11 +1060,191 @@ async fn test_manual_transition_async_overlapping_scope_conflict_reports_active_ 50, "terminal conflict winner response: {terminal:#?}" ); + assert_eq!( + terminal.report.transition_completed, terminal.report.enqueued, + "terminal conflict winner must wait for all queued transitions: {terminal:#?}" + ); + assert_eq!(terminal.report.transition_failed, 0, "terminal conflict winner response: {terminal:#?}"); + assert_eq!(terminal.report.tier_failure, 0, "terminal conflict winner response: {terminal:#?}"); assert!(cold_tier_object_count(&cold_client).await? <= 50); Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn test_manual_transition_async_tier_failure_reports_terminal_partial() -> TestResult { + let mut cold = RustFSTestEnvironment::new().await?; + cold.access_key = "manualtierfailurecoldadmin".to_string(); + cold.secret_key = "manualtierfailurecoldsecret".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")]) + .await?; + let hot_client = hot.create_s3_client(); + add_rustfs_tier(&hot, &cold).await?; + + hot_client.create_bucket().bucket(MANUAL_TIER_FAILURE_BUCKET).send().await?; + let due_mtime = OffsetDateTime::now_utc() - time::Duration::hours(25); + put_backdated_single_part_object( + &hot_client, + MANUAL_TIER_FAILURE_BUCKET, + MANUAL_TIER_FAILURE_KEY, + b"manual tier failure object", + due_mtime, + ) + .await?; + put_lifecycle_transition_rule( + &hot_client, + MANUAL_TIER_FAILURE_BUCKET, + "manual-tier-failure", + MANUAL_TIER_FAILURE_PREFIX, + 0, + ) + .await?; + remove_rustfs_tier_force(&hot).await?; + + let before_remote_count = cold_tier_object_count(&cold_client).await?; + let accepted = manual_transition_async_run(&hot, MANUAL_TIER_FAILURE_BUCKET, MANUAL_TIER_FAILURE_PREFIX, false, 10).await?; + assert_eq!(accepted.state, "accepted"); + assert_eq!(accepted.mode, "durable_job"); + assert_eq!(accepted.report.transition_completed, 0); + assert_eq!(accepted.report.transition_failed, 0); + let job_id = accepted.job_id.as_deref().ok_or("async response must include job_id")?; + let status_endpoint = accepted + .status_endpoint + .as_deref() + .ok_or("async response must include status_endpoint")?; + + let terminal = wait_for_manual_transition_job_terminal(&hot, status_endpoint, StdDuration::from_secs(30)).await?; + assert_eq!(terminal.job_id, job_id); + assert_eq!(terminal.status, "partial", "terminal tier failure job response: {terminal:#?}"); + assert!(!terminal.cancel_requested); + assert_eq!(terminal.failure_reason, None); + assert_eq!(terminal.report.bucket, MANUAL_TIER_FAILURE_BUCKET); + assert_eq!(terminal.report.prefix, MANUAL_TIER_FAILURE_PREFIX); + assert_eq!(terminal.report.tier.as_deref(), Some(TIER_NAME)); + assert!(!terminal.report.dry_run); + assert!(terminal.report.lifecycle_config_found); + assert_eq!(terminal.report.scanned, 1, "terminal tier failure job response: {terminal:#?}"); + assert_eq!(terminal.report.eligible, 0, "terminal tier failure job response: {terminal:#?}"); + assert_eq!(terminal.report.enqueued, 0, "terminal tier failure job response: {terminal:#?}"); + assert_eq!( + terminal.report.transition_completed, 0, + "terminal tier failure job response: {terminal:#?}" + ); + assert_eq!(terminal.report.transition_failed, 0, "terminal tier failure job response: {terminal:#?}"); + assert_eq!(terminal.report.tier_failure, 1, "terminal tier failure job response: {terminal:#?}"); + assert_eq!(terminal.report.skipped_queue_full, 0); + assert_eq!(terminal.report.skipped_queue_closed, 0); + assert_eq!(terminal.report.skipped_queue_timeout, 0); + assert!(!terminal.report.cancelled); + assert!(!terminal.report.truncated_by_limit); + assert!(!terminal.report.truncated_by_duration); + assert_eq!( + cold_tier_object_count(&cold_client).await?, + before_remote_count, + "tier failure must not create a remote object" + ); + assert_remains_not_transitioned( + &hot_client, + MANUAL_TIER_FAILURE_BUCKET, + MANUAL_TIER_FAILURE_KEY, + StdDuration::from_secs(2), + ) + .await?; + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn test_manual_transition_async_worker_failure_reports_terminal_partial() -> TestResult { + let mut cold = RustFSTestEnvironment::new().await?; + cold.access_key = "manualworkerfailurecoldadmin".to_string(); + cold.secret_key = "manualworkerfailurecoldsecret".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")]) + .await?; + let hot_client = hot.create_s3_client(); + add_rustfs_tier(&hot, &cold).await?; + cold.stop_server(); + + hot_client.create_bucket().bucket(MANUAL_WORKER_FAILURE_BUCKET).send().await?; + let due_mtime = OffsetDateTime::now_utc() - time::Duration::hours(25); + put_backdated_single_part_object( + &hot_client, + MANUAL_WORKER_FAILURE_BUCKET, + MANUAL_WORKER_FAILURE_KEY, + b"manual worker failure object", + due_mtime, + ) + .await?; + put_lifecycle_transition_rule( + &hot_client, + MANUAL_WORKER_FAILURE_BUCKET, + "manual-worker-failure", + MANUAL_WORKER_FAILURE_PREFIX, + 0, + ) + .await?; + + let accepted = + manual_transition_async_run(&hot, MANUAL_WORKER_FAILURE_BUCKET, MANUAL_WORKER_FAILURE_PREFIX, false, 10).await?; + assert_eq!(accepted.state, "accepted"); + assert_eq!(accepted.mode, "durable_job"); + assert_eq!(accepted.report.transition_completed, 0); + assert_eq!(accepted.report.transition_failed, 0); + let job_id = accepted.job_id.as_deref().ok_or("async response must include job_id")?; + let status_endpoint = accepted + .status_endpoint + .as_deref() + .ok_or("async response must include status_endpoint")?; + + let terminal = wait_for_manual_transition_job_terminal(&hot, status_endpoint, StdDuration::from_secs(30)).await?; + assert_eq!(terminal.job_id, job_id); + assert_eq!(terminal.status, "partial", "terminal worker failure job response: {terminal:#?}"); + assert!(!terminal.cancel_requested); + assert_eq!(terminal.failure_reason, None); + assert_eq!(terminal.report.bucket, MANUAL_WORKER_FAILURE_BUCKET); + assert_eq!(terminal.report.prefix, MANUAL_WORKER_FAILURE_PREFIX); + assert_eq!(terminal.report.tier.as_deref(), Some(TIER_NAME)); + assert!(!terminal.report.dry_run); + assert!(terminal.report.lifecycle_config_found); + assert_eq!(terminal.report.scanned, 1, "terminal worker failure job response: {terminal:#?}"); + assert_eq!(terminal.report.eligible, 1, "terminal worker failure job response: {terminal:#?}"); + assert_eq!(terminal.report.enqueued, 1, "terminal worker failure job response: {terminal:#?}"); + assert_eq!( + terminal.report.transition_completed, 0, + "terminal worker failure job response: {terminal:#?}" + ); + assert_eq!( + terminal.report.transition_failed, 1, + "terminal worker failure job response: {terminal:#?}" + ); + assert_eq!(terminal.report.tier_failure, 1, "terminal worker failure job response: {terminal:#?}"); + assert_eq!(terminal.report.skipped_queue_full, 0); + assert_eq!(terminal.report.skipped_queue_closed, 0); + assert_eq!(terminal.report.skipped_queue_timeout, 0); + assert!(!terminal.report.cancelled); + assert!(!terminal.report.truncated_by_limit); + assert!(!terminal.report.truncated_by_duration); + assert_remains_not_transitioned( + &hot_client, + MANUAL_WORKER_FAILURE_BUCKET, + MANUAL_WORKER_FAILURE_KEY, + StdDuration::from_secs(2), + ) + .await?; + + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn test_manual_transition_run_contract_no_status_cancel_fields() -> TestResult { let mut cold = RustFSTestEnvironment::new().await?; @@ -1092,6 +1321,8 @@ async fn test_manual_transition_run_continuation_token_resumes_without_raw_marke assert_eq!(first.report.scanned, 1, "first continuation page: {first:#?}"); assert_eq!(first.report.eligible, 1, "first continuation page: {first:#?}"); assert_eq!(first.report.dry_run_eligible, 1, "first continuation page: {first:#?}"); + assert_eq!(first.report.tier_failure, 0); + assert!(!first.report.cancelled); assert!(first.report.truncated_by_limit); let continuation = first .report @@ -1119,6 +1350,8 @@ async fn test_manual_transition_run_continuation_token_resumes_without_raw_marke assert_eq!(second.report.scanned, 1, "second continuation page: {second:#?}"); assert_eq!(second.report.eligible, 1, "second continuation page: {second:#?}"); assert_eq!(second.report.dry_run_eligible, 1, "second continuation page: {second:#?}"); + assert_eq!(second.report.tier_failure, 0); + assert!(!second.report.cancelled); assert!(!second.report.truncated_by_limit); assert!(second.report.continuation_token.is_none()); @@ -1173,6 +1406,8 @@ async fn test_manual_transition_run_queue_pressure_partial() -> TestResult { "expected queue-pressure path to skip at least one object: {:#?}", response.report ); + assert_eq!(response.report.tier_failure, 0); + assert!(!response.report.cancelled); assert!(!response.report.truncated_by_duration); assert!(response.report.enqueued < 20, "partial run should not enqueue all items in this setup"); diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index c89d171f4..0f64a6ac2 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -23,10 +23,11 @@ use crate::bucket::lifecycle::lifecycle::{ }; use crate::bucket::lifecycle::manual_transition_job::{ MANUAL_TRANSITION_JOB_RECORD_PREFIX, ManualTransitionJobRecord, ManualTransitionScopeAdmission, - ManualTransitionScopeAdmissionClaim, claim_manual_transition_scope_admission, + ManualTransitionScopeAdmissionClaim, ManualTransitionWorkerResult, claim_manual_transition_scope_admission, delete_manual_transition_scope_admission_if_current, load_manual_transition_job_record, load_manual_transition_job_record_with_etag, manual_transition_job_id_from_record_object_name, - manual_transition_job_lease_expired, persist_manual_transition_job_progress, save_manual_transition_job_record_if_current, + manual_transition_job_lease_expired, persist_manual_transition_job_progress, record_manual_transition_worker_result, + renew_manual_transition_job_lease, save_manual_transition_job_record_if_current, }; use crate::bucket::lifecycle::replication_sink; use crate::bucket::lifecycle::replication_sink::{ @@ -954,6 +955,7 @@ struct TransitionTask { obj_info: ObjectInfo, src: LcEventSrc, event: lifecycle::Event, + manual_job_id: Option, } impl ExpiryOp for TransitionTask { @@ -1259,6 +1261,7 @@ impl TransitionState { oi: &ObjectInfo, event: &lifecycle::Event, src: &LcEventSrc, + manual_job_id: Option, ) -> TransitionEnqueueOutcome { if is_immediate_transition_source(src) && should_force_immediate_transition_enqueue_timeout() { self.handle_immediate_enqueue_failure(oi, src, ImmediateEnqueueFailure::ForcedTimeout); @@ -1280,6 +1283,7 @@ impl TransitionState { obj_info: oi.clone(), src: src.clone(), event: event.clone(), + manual_job_id, }; if is_immediate_transition_source(src) { let outcome = match self.transition_tx.try_send(Some(task)) { @@ -1361,7 +1365,7 @@ impl TransitionState { } pub async fn queue_transition_task(self: &Arc, oi: &ObjectInfo, event: &lifecycle::Event, src: &LcEventSrc) -> bool { - self.queue_transition_task_outcome(oi, event, src).await.is_handled() + self.queue_transition_task_outcome(oi, event, src, None).await.is_handled() } pub async fn init(api: Arc) { @@ -1454,24 +1458,43 @@ impl TransitionState { ..Default::default() }; - if let Err(err) = transition_object(api.clone(), &task.obj_info, LcAuditEvent::new(task.event.clone(), task.src.clone())).await { - global_metrics().record_scanner_transition_failed(1); - if !is_err_version_not_found(&err) && !is_err_object_not_found(&err) && !is_network_or_host_down(&err.to_string(), false) && !err.to_string().contains("use of closed network connection") { - error!( - event = EVENT_LIFECYCLE_TIER_OPERATION_FAILED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_LIFECYCLE, - bucket = %task.obj_info.bucket, - object = %task.obj_info.name, - version_id = %task.obj_info.version_id.map(|v| v.to_string()).unwrap_or_default(), - tier = %task.event.storage_class, - operation = "transition_object", - error = %err, - "Lifecycle tier operation failed" - ); - } + if let Err(err) = + transition_object(api.clone(), &task.obj_info, LcAuditEvent::new(task.event.clone(), task.src.clone())) + .await + { + if let Some(job_id) = task.manual_job_id { + record_manual_transition_worker_result_for_task( + api.clone(), + job_id, + ManualTransitionWorkerResult::TierFailure, + ) + .await; + } + global_metrics().record_scanner_transition_failed(1); + if !is_err_version_not_found(&err) && !is_err_object_not_found(&err) && !is_network_or_host_down(&err.to_string(), false) && !err.to_string().contains("use of closed network connection") { + error!( + event = EVENT_LIFECYCLE_TIER_OPERATION_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + bucket = %task.obj_info.bucket, + object = %task.obj_info.name, + version_id = %task.obj_info.version_id.map(|v| v.to_string()).unwrap_or_default(), + tier = %task.event.storage_class, + operation = "transition_object", + error = %err, + "Lifecycle tier operation failed" + ); + } emit_transition_failed_event(obj_info_for_event); } else { + if let Some(job_id) = task.manual_job_id { + record_manual_transition_worker_result_for_task( + api.clone(), + job_id, + ManualTransitionWorkerResult::Completed, + ) + .await; + } global_metrics().record_scanner_transition_completed(1); let mut ts = TierStats { total_size: task.obj_info.size as u64, @@ -1577,6 +1600,20 @@ impl TransitionState { } } +async fn record_manual_transition_worker_result_for_task(api: Arc, job_id: Uuid, result: ManualTransitionWorkerResult) { + if let Err(err) = record_manual_transition_worker_result(api, job_id, result, manual_transition_queue_snapshot()).await { + warn!( + event = EVENT_LIFECYCLE_WORKER_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + job_id = %job_id, + error = %err, + state = "manual_transition_worker_result_failed", + "Manual transition worker failed to persist job result" + ); + } +} + pub async fn init_background_expiry(api: Arc) { let mut workers = get_env_usize("RUSTFS_MAX_EXPIRY_WORKERS", std::cmp::min(num_cpus::get(), 16)); //globalILMConfig.getExpirationWorkers() @@ -1606,7 +1643,7 @@ fn spawn_manual_transition_job_recovery_once(api: Arc) -> Option {} - result = recover_manual_transition_jobs_once(api, DEFAULT_MANUAL_TRANSITION_JOB_RECOVERY_LIMIT, None) => { + result = recover_manual_transition_jobs(api, DEFAULT_MANUAL_TRANSITION_JOB_RECOVERY_LIMIT) => { match result { Ok(stats) => { debug!( @@ -1658,6 +1695,32 @@ enum ManualTransitionJobRecoveryOutcome { Skipped, } +async fn recover_manual_transition_jobs(api: Arc, limit: usize) -> Result { + let mut marker = None; + let mut total = ManualTransitionJobRecoveryStats::default(); + + loop { + let stats = recover_manual_transition_jobs_once(api.clone(), limit, marker).await?; + total.scanned = total.scanned.saturating_add(stats.scanned); + total.resumed = total.resumed.saturating_add(stats.resumed); + total.cancelled = total.cancelled.saturating_add(stats.cancelled); + total.skipped = total.skipped.saturating_add(stats.skipped); + total.failed = total.failed.saturating_add(stats.failed); + + if !stats.truncated { + total.truncated = false; + total.next_marker = None; + return Ok(total); + } + let Some(next_marker) = stats.next_marker else { + return Err(Error::other("manual transition job recovery page is truncated without a next marker")); + }; + total.truncated = true; + total.next_marker = Some(next_marker.clone()); + marker = Some(next_marker); + } +} + pub async fn recover_manual_transition_jobs_once( api: Arc, limit: usize, @@ -1771,11 +1834,16 @@ async fn recover_manual_transition_job(api: Arc, job_id: Uuid) -> Resul } let mut options = record.resume_options(); + options.job_id = Some(job_id); options.cancel_check = Some(manual_transition_recovery_cancel_check(api.clone(), job_id)); options.progress_sink = Some(manual_transition_recovery_progress_sink(api.clone(), job_id)); let result = enqueue_transition_for_existing_objects_scoped(api.clone(), &record.bucket, options).await; let final_record = finalize_recovered_manual_transition_job(api.clone(), job_id, result).await?; - release_manual_transition_recovery_admission(api, &final_record).await; + if final_record.is_terminal() { + release_manual_transition_recovery_admission(api, &final_record).await; + } else { + spawn_manual_transition_recovery_heartbeat(api, job_id); + } Ok(ManualTransitionJobRecoveryOutcome::Resumed) } @@ -1845,6 +1913,34 @@ async fn release_manual_transition_recovery_admission(api: Arc, record: } } +fn spawn_manual_transition_recovery_heartbeat(api: Arc, job_id: Uuid) { + tokio::spawn(async move { + let mut interval = tokio::time::interval(std::time::Duration::from_secs(5)); + loop { + interval.tick().await; + match renew_manual_transition_job_lease(api.clone(), job_id, manual_transition_queue_snapshot()).await { + Ok(record) if record.is_terminal() => { + release_manual_transition_recovery_admission(api, &record).await; + return; + } + Ok(_) => {} + Err(Error::ConfigNotFound) => return, + Err(err) => { + warn!( + event = EVENT_LIFECYCLE_WORKER_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + job_id = %job_id, + error = %err, + state = "manual_transition_recovery_heartbeat_failed", + "Manual transition recovery failed to renew job lease" + ); + } + } + } + }); +} + async fn abandon_manual_transition_recovery_lease(api: Arc, job_id: Uuid, lease_id: Uuid) -> Result<(), Error> { for _ in 0..4 { let (mut record, etag) = match load_manual_transition_job_record_with_etag(api.clone(), job_id).await { @@ -2770,6 +2866,8 @@ pub struct ManualTransitionRunOptions { pub max_objects: Option, pub max_duration: Option, #[serde(skip)] + pub job_id: Option, + #[serde(skip)] pub cancel_token: Option, #[serde(skip)] pub cancel_check: Option, @@ -2788,6 +2886,7 @@ impl std::fmt::Debug for ManualTransitionRunOptions { .field("dry_run", &self.dry_run) .field("max_objects", &self.max_objects) .field("max_duration", &self.max_duration) + .field("job_id", &self.job_id) .field("cancel_token", &self.cancel_token.is_some()) .field("cancel_check", &self.cancel_check.is_some()) .field("progress_sink", &self.progress_sink.is_some()) @@ -2810,6 +2909,10 @@ impl PartialEq for ManualTransitionRunOptions { impl Eq for ManualTransitionRunOptions {} +fn is_zero_u64(value: &u64) -> bool { + *value == 0 +} + #[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)] #[serde(deny_unknown_fields)] pub struct ManualTransitionRunReport { @@ -2832,6 +2935,10 @@ pub struct ManualTransitionRunReport { pub skipped_queue_full: u64, pub skipped_queue_closed: u64, pub skipped_queue_timeout: u64, + #[serde(default, skip_serializing_if = "is_zero_u64")] + pub transition_completed: u64, + #[serde(default, skip_serializing_if = "is_zero_u64")] + pub transition_failed: u64, pub tier_failure: u64, pub truncated_by_limit: bool, pub truncated_by_duration: bool, @@ -2898,6 +3005,19 @@ impl ManualTransitionRunReport { } } } + + pub fn merge_scan_report_preserving_worker(&mut self, scan_report: &ManualTransitionRunReport) { + let transition_completed = self.transition_completed; + let transition_failed = self.transition_failed; + *self = scan_report.clone(); + self.transition_completed = transition_completed; + self.transition_failed = transition_failed; + self.tier_failure = scan_report.tier_failure.saturating_add(transition_failed); + } + + pub fn worker_transition_pending(&self) -> bool { + self.transition_completed.saturating_add(self.transition_failed) < self.enqueued + } } #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] @@ -3334,13 +3454,22 @@ async fn enqueue_transition_with_lifecycle_report( report.skipped_tier = report.skipped_tier.saturating_add(1); return false; } + if !options.dry_run + && !runtime_sources::tier_config_mgr_handle() + .read() + .await + .is_tier_valid(&event.storage_class) + { + report.tier_failure = report.tier_failure.saturating_add(1); + return false; + } report.eligible = report.eligible.saturating_add(1); if options.dry_run { report.dry_run_eligible = report.dry_run_eligible.saturating_add(1); return true; } let outcome = runtime_sources::transition_state_handle() - .queue_transition_task_outcome(oi, &event, src) + .queue_transition_task_outcome(oi, &event, src, options.job_id) .await; report.record_enqueue_outcome(outcome); return outcome.is_handled(); @@ -4223,12 +4352,12 @@ mod tests { lifecycle_rule_has_date_expiration, lifecycle_version_purge_state_from_completed_targets, 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, - persist_manual_transition_page_checkpoint, recover_manual_transition_jobs_once, 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, - set_recovered_free_version_enqueue_observer, should_defer_date_expiry_for_recent_config_update, - should_reuse_lifecycle_delete_replication_state, transitioned_cleanup_tuple, transitioned_object_delete_opts, - wait_for_tier_free_version_recovery, + persist_manual_transition_page_checkpoint, recover_manual_transition_jobs, recover_manual_transition_jobs_once, + 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, set_recovered_free_version_enqueue_observer, + should_defer_date_expiry_for_recent_config_update, should_reuse_lifecycle_delete_replication_state, + transitioned_cleanup_tuple, transitioned_object_delete_opts, wait_for_tier_free_version_recovery, }; #[cfg(feature = "test-util")] use super::{delete_free_version_remote_object_then, encode_dir_object, get_transitioned_object_reader_with_tier_manager}; @@ -6375,10 +6504,10 @@ mod tests { }; let first = state - .queue_transition_task_outcome(&object, &event, &LcEventSrc::Scanner) + .queue_transition_task_outcome(&object, &event, &LcEventSrc::Scanner, None) .await; let second = state - .queue_transition_task_outcome(&object, &event, &LcEventSrc::Scanner) + .queue_transition_task_outcome(&object, &event, &LcEventSrc::Scanner, None) .await; assert_eq!(first, TransitionEnqueueOutcome::Queued); @@ -6406,10 +6535,10 @@ mod tests { }; let first = state - .queue_transition_task_outcome(&first_object, &event, &LcEventSrc::Scanner) + .queue_transition_task_outcome(&first_object, &event, &LcEventSrc::Scanner, None) .await; let second = state - .queue_transition_task_outcome(&second_object, &event, &LcEventSrc::Scanner) + .queue_transition_task_outcome(&second_object, &event, &LcEventSrc::Scanner, None) .await; assert_eq!(first, TransitionEnqueueOutcome::Queued); @@ -6910,6 +7039,22 @@ mod tests { assert_eq!(report.skipped_tier, 1); } + #[tokio::test] + async fn manual_transition_reports_runtime_tier_failure_before_enqueue() { + let lc = latest_transition_lifecycle(); + let object = current_object(ReplicationStatusType::Completed); + let options = ManualTransitionRunOptions::default(); + let mut report = ManualTransitionRunReport::new(&object.bucket, &options); + + let handled = enqueue_transition_with_lifecycle_report(&object, &lc, &LcEventSrc::Scanner, &options, &mut report).await; + + assert!(!handled); + assert_eq!(report.eligible, 0); + assert_eq!(report.enqueued, 0); + assert_eq!(report.tier_failure, 1); + assert!(!report.has_partial_enqueue()); + } + #[tokio::test] async fn manual_transition_counts_already_transitioned_object() { let lc = latest_transition_lifecycle(); @@ -7108,6 +7253,93 @@ mod tests { ); } + #[tokio::test] + #[serial] + async fn manual_transition_recovery_drains_multiple_record_pages() { + let (_paths, ecstore) = setup_test_env().await; + let mut job_ids = Vec::new(); + + for bucket in ["manual-recovery-page-a", "manual-recovery-page-b"] { + let job_id = Uuid::new_v4(); + let options = ManualTransitionRunOptions { + prefix: "logs/".to_string(), + ..Default::default() + }; + let mut record = ManualTransitionJobRecord::new(job_id, bucket, &options, "old-owner"); + record.lease_expires_at_unix_nanos = 0; + save_manual_transition_job_record(ecstore.clone(), &record) + .await + .expect("expired job record should save"); + save_manual_transition_scope_admission_if_absent(ecstore.clone(), &ManualTransitionScopeAdmission::from_job(&record)) + .await + .expect("expired scope admission should save"); + job_ids.push((job_id, record.scope_key.clone())); + } + + let stats = recover_manual_transition_jobs(ecstore.clone(), 1) + .await + .expect("manual transition recovery should drain all pages"); + + assert_eq!(stats.resumed, 2); + assert_eq!(stats.failed, 0); + assert!(!stats.truncated); + assert!(stats.next_marker.is_none()); + for (job_id, scope_key) in job_ids { + let recovered = load_manual_transition_job_record(ecstore.clone(), job_id) + .await + .expect("recovered job should load"); + assert_eq!(recovered.state, ManualTransitionJobState::Completed); + assert!( + matches!( + load_manual_transition_scope_admission(ecstore.clone(), &scope_key).await, + Err(Error::ConfigNotFound) + ), + "completed recovery must release every scope admission" + ); + } + } + + #[tokio::test] + #[serial] + async fn manual_transition_recovery_completes_expired_cancelled_record() { + let (_paths, ecstore) = setup_test_env().await; + let job_id = Uuid::new_v4(); + let options = ManualTransitionRunOptions { + prefix: "logs/".to_string(), + ..Default::default() + }; + let mut record = ManualTransitionJobRecord::new(job_id, "manual-recovery-cancel-bucket", &options, "old-owner"); + record.lease_expires_at_unix_nanos = 0; + record.mark_cancel_requested(); + save_manual_transition_job_record(ecstore.clone(), &record) + .await + .expect("expired cancelled job record should save"); + save_manual_transition_scope_admission_if_absent(ecstore.clone(), &ManualTransitionScopeAdmission::from_job(&record)) + .await + .expect("expired cancelled scope admission should save"); + + let stats = recover_manual_transition_jobs(ecstore.clone(), 10) + .await + .expect("manual transition recovery should process cancelled jobs"); + + assert_eq!(stats.cancelled, 1); + assert_eq!(stats.resumed, 0); + assert_eq!(stats.failed, 0); + let recovered = load_manual_transition_job_record(ecstore.clone(), job_id) + .await + .expect("cancelled job should load"); + assert_eq!(recovered.state, ManualTransitionJobState::Cancelled); + assert!(recovered.cancel_requested); + assert!(recovered.report.cancelled); + assert!( + matches!( + load_manual_transition_scope_admission(ecstore, &record.scope_key).await, + Err(Error::ConfigNotFound) + ), + "cancelled recovery must release the scope admission" + ); + } + #[test] fn manual_transition_cancelled_report_is_partial_and_resumable() { let report = ManualTransitionRunReport { diff --git a/crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs b/crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs index 2c35e2594..626b6ff3c 100644 --- a/crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs +++ b/crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs @@ -37,6 +37,10 @@ pub const MAX_MANUAL_TRANSITION_JOB_RECORD_SIZE: usize = 64 * 1024; const MANUAL_TRANSITION_JOB_LEASE_SECONDS: i128 = 60; const MANUAL_TRANSITION_LEGACY_SCOPE_SCAN_LIMIT: i32 = 1000; +fn is_false(value: &bool) -> bool { + !*value +} + #[derive(Debug, thiserror::Error)] pub enum ManualTransitionJobError { #[error("manual transition job is corrupt: {0}")] @@ -75,6 +79,8 @@ pub struct ManualTransitionJobRecord { pub lease_id: Uuid, pub lease_expires_at_unix_nanos: i128, pub state: ManualTransitionJobState, + #[serde(default, skip_serializing_if = "is_false")] + pub scan_completed: bool, pub cancel_requested: bool, pub created_at_unix_nanos: i128, pub updated_at_unix_nanos: i128, @@ -103,6 +109,7 @@ impl ManualTransitionJobRecord { lease_id, lease_expires_at_unix_nanos: manual_transition_job_lease_expires_at(now), state: ManualTransitionJobState::Running, + scan_completed: false, cancel_requested: false, created_at_unix_nanos: now, updated_at_unix_nanos: now, @@ -120,17 +127,11 @@ impl ManualTransitionJobRecord { } pub fn complete(&mut self, report: ManualTransitionRunReport, queue_snapshot: ManualTransitionQueueSnapshot) { - self.state = if report.cancelled { - ManualTransitionJobState::Cancelled - } else if report.was_truncated() || report.has_partial_enqueue() || report.tier_failure > 0 { - ManualTransitionJobState::Partial - } else { - ManualTransitionJobState::Completed - }; - self.report = report; + self.scan_completed = true; + self.report.merge_scan_report_preserving_worker(&report); self.queue_snapshot = queue_snapshot; self.error = None; - self.mark_updated_terminal(); + self.mark_terminal_if_worker_drained(); } pub fn fail(&mut self, error: impl Into) { @@ -140,6 +141,24 @@ impl ManualTransitionJobRecord { self.mark_updated_terminal(); } + pub fn record_worker_result(&mut self, result: ManualTransitionWorkerResult, queue_snapshot: ManualTransitionQueueSnapshot) { + if self.is_terminal() { + return; + } + match result { + ManualTransitionWorkerResult::Completed => { + self.report.transition_completed = self.report.transition_completed.saturating_add(1); + } + ManualTransitionWorkerResult::TierFailure => { + self.report.transition_failed = self.report.transition_failed.saturating_add(1); + self.report.tier_failure = self.report.tier_failure.saturating_add(1); + } + } + self.queue_snapshot = queue_snapshot; + self.updated_at_unix_nanos = OffsetDateTime::now_utc().unix_timestamp_nanos(); + self.mark_terminal_if_worker_drained(); + } + pub fn mark_cancel_requested(&mut self) { self.cancel_requested = true; self.updated_at_unix_nanos = OffsetDateTime::now_utc().unix_timestamp_nanos(); @@ -178,9 +197,25 @@ impl ManualTransitionJobRecord { self.queue_snapshot = queue_snapshot; } + pub fn mark_unknown_if_worker_results_lost(&mut self, queue_snapshot: ManualTransitionQueueSnapshot) -> bool { + if self.state != ManualTransitionJobState::Running + || !self.scan_completed + || !self.report.worker_transition_pending() + || queue_snapshot.queued > 0 + || queue_snapshot.active > 0 + { + return false; + } + self.queue_snapshot = queue_snapshot; + self.state = ManualTransitionJobState::Unknown; + self.error = Some("manual transition worker result was not persisted before the transition queue drained".to_string()); + self.mark_updated_terminal(); + true + } + pub fn update_running_progress(&mut self, report: ManualTransitionRunReport, queue_snapshot: ManualTransitionQueueSnapshot) { if self.state == ManualTransitionJobState::Running { - self.report = report; + self.report.merge_scan_report_preserving_worker(&report); self.renew_lease(queue_snapshot); } } @@ -210,6 +245,24 @@ impl ManualTransitionJobRecord { self.completed_at_unix_nanos = Some(now); } + fn mark_terminal_if_worker_drained(&mut self) { + if !self.scan_completed || self.report.worker_transition_pending() { + return; + } + self.state = if self.cancel_requested || self.report.cancelled { + ManualTransitionJobState::Cancelled + } else if self.report.was_truncated() + || self.report.has_partial_enqueue() + || self.report.tier_failure > 0 + || self.report.transition_failed > 0 + { + ManualTransitionJobState::Partial + } else { + ManualTransitionJobState::Completed + }; + self.mark_updated_terminal(); + } + pub fn encode(&self) -> Result, ManualTransitionJobError> { self.validate()?; let job_bytes = serde_json::to_vec(self)?; @@ -275,6 +328,12 @@ impl ManualTransitionJobRecord { } } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ManualTransitionWorkerResult { + Completed, + TierFailure, +} + fn manual_transition_job_lease_expires_at(now_unix_nanos: i128) -> i128 { now_unix_nanos.saturating_add(MANUAL_TRANSITION_JOB_LEASE_SECONDS.saturating_mul(1_000_000_000)) } @@ -696,6 +755,40 @@ pub async fn persist_manual_transition_job_progress( Ok(record) } +pub async fn record_manual_transition_worker_result( + api: Arc, + job_id: Uuid, + result: ManualTransitionWorkerResult, + queue_snapshot: ManualTransitionQueueSnapshot, +) -> EcstoreResult { + for _ in 0..4 { + let (mut record, etag) = load_manual_transition_job_record_with_etag(api.clone(), job_id).await?; + if record.is_terminal() { + return Ok(record); + } + record.record_worker_result(result, queue_snapshot); + match save_manual_transition_job_record_if_current(api.clone(), &record, &etag).await { + Ok(()) => { + if record.is_terminal() { + delete_manual_transition_scope_admission_if_current( + api.clone(), + &record.scope_key, + record.job_id, + record.lease_id, + ) + .await?; + } else { + renew_manual_transition_scope_admission_from_job(api, &record).await?; + } + return Ok(record); + } + Err(Error::PreconditionFailed) => continue, + Err(err) => return Err(err), + } + } + Err(Error::PreconditionFailed) +} + pub async fn renew_manual_transition_job_lease( api: Arc, job_id: Uuid, @@ -703,9 +796,16 @@ pub async fn renew_manual_transition_job_lease( ) -> EcstoreResult { let (mut record, etag) = load_manual_transition_job_record_with_etag(api.clone(), job_id).await?; if record.state == ManualTransitionJobState::Running { - record.renew_lease(queue_snapshot); + let became_terminal = record.mark_unknown_if_worker_results_lost(queue_snapshot); + if !became_terminal { + record.renew_lease(queue_snapshot); + } save_manual_transition_job_record_if_current(api.clone(), &record, &etag).await?; - renew_manual_transition_scope_admission_from_job(api, &record).await?; + if became_terminal { + delete_manual_transition_scope_admission_if_current(api, &record.scope_key, record.job_id, record.lease_id).await?; + } else { + renew_manual_transition_scope_admission_from_job(api, &record).await?; + } } Ok(record) } @@ -801,6 +901,88 @@ mod tests { assert_eq!(decoded.max_objects, Some(17)); } + #[test] + fn manual_transition_job_record_waits_for_worker_results() { + let options = ManualTransitionRunOptions::default(); + let mut record = ManualTransitionJobRecord::new(Uuid::new_v4(), "bucket", &options, TEST_OWNER); + + record.complete( + ManualTransitionRunReport { + bucket: "bucket".to_string(), + enqueued: 2, + ..Default::default() + }, + ManualTransitionQueueSnapshot::default(), + ); + + assert!(record.scan_completed); + assert_eq!(record.state, ManualTransitionJobState::Running); + assert!(record.completed_at_unix_nanos.is_none()); + + record.record_worker_result(ManualTransitionWorkerResult::Completed, ManualTransitionQueueSnapshot::default()); + assert_eq!(record.state, ManualTransitionJobState::Running); + assert_eq!(record.report.transition_completed, 1); + + record.record_worker_result(ManualTransitionWorkerResult::TierFailure, ManualTransitionQueueSnapshot::default()); + assert_eq!(record.state, ManualTransitionJobState::Partial); + assert_eq!(record.report.transition_completed, 1); + assert_eq!(record.report.transition_failed, 1); + assert_eq!(record.report.tier_failure, 1); + assert!(record.completed_at_unix_nanos.is_some()); + } + + #[test] + fn manual_transition_job_scan_progress_preserves_worker_counters() { + let options = ManualTransitionRunOptions::default(); + let mut record = ManualTransitionJobRecord::new(Uuid::new_v4(), "bucket", &options, TEST_OWNER); + record.record_worker_result(ManualTransitionWorkerResult::TierFailure, ManualTransitionQueueSnapshot::default()); + + record.update_running_progress( + ManualTransitionRunReport { + bucket: "bucket".to_string(), + scanned: 3, + enqueued: 1, + tier_failure: 2, + ..Default::default() + }, + ManualTransitionQueueSnapshot::default(), + ); + + assert_eq!(record.report.scanned, 3); + assert_eq!(record.report.enqueued, 1); + assert_eq!(record.report.transition_failed, 1); + assert_eq!(record.report.tier_failure, 3); + } + + #[test] + fn manual_transition_job_marks_unknown_when_worker_results_are_lost() { + let options = ManualTransitionRunOptions::default(); + let mut record = ManualTransitionJobRecord::new(Uuid::new_v4(), "bucket", &options, TEST_OWNER); + + record.complete( + ManualTransitionRunReport { + bucket: "bucket".to_string(), + enqueued: 1, + ..Default::default() + }, + ManualTransitionQueueSnapshot { + queued: 1, + ..Default::default() + }, + ); + + assert_eq!(record.state, ManualTransitionJobState::Running); + assert!(!record.mark_unknown_if_worker_results_lost(ManualTransitionQueueSnapshot { + queued: 1, + ..Default::default() + })); + + assert!(record.mark_unknown_if_worker_results_lost(ManualTransitionQueueSnapshot::default())); + assert_eq!(record.state, ManualTransitionJobState::Unknown); + assert!(record.error.as_deref().is_some_and(|err| err.contains("worker result"))); + assert!(record.completed_at_unix_nanos.is_some()); + } + #[test] fn manual_transition_job_record_builds_resume_options() { let options = ManualTransitionRunOptions { diff --git a/rustfs/src/admin/handlers/ilm_transition.rs b/rustfs/src/admin/handlers/ilm_transition.rs index bbf7e5708..7c2bdd1e0 100644 --- a/rustfs/src/admin/handlers/ilm_transition.rs +++ b/rustfs/src/admin/handlers/ilm_transition.rs @@ -286,6 +286,7 @@ fn parse_manual_transition_query(query: Option<&str>) -> S3Result<(String, Manua 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), + job_id: None, cancel_token: None, cancel_check: None, progress_sink: None, @@ -412,7 +413,7 @@ async fn authorize_manual_transition_request(req: &S3Request) -> S3Result< } fn response_state(report: &ManualTransitionRunReport) -> &'static str { - if report.was_truncated() || report.has_partial_enqueue() { + if report.was_truncated() || report.has_partial_enqueue() || report.tier_failure > 0 || report.transition_failed > 0 { "partial" } else { "completed" @@ -659,15 +660,25 @@ async fn finalize_manual_transition_job( } } -fn spawn_manual_transition_job_heartbeat(store: Arc, job_id: Uuid, cancel_token: CancellationToken) { +fn spawn_manual_transition_job_heartbeat( + store: Arc, + job_id: Uuid, + scan_cancel_token: CancellationToken, + shutdown_token: CancellationToken, +) { tokio::spawn(async move { let mut interval = tokio::time::interval(StdDuration::from_secs(5)); loop { tokio::select! { - _ = cancel_token.cancelled() => return, + _ = shutdown_token.cancelled() => return, _ = interval.tick() => { match renew_manual_transition_job_lease(store.clone(), job_id, manual_transition_queue_snapshot()).await { - Ok(record) if record.cancel_requested => cancel_token.cancel(), + Ok(record) if record.is_terminal() => { + remove_active_manual_transition_job(job_id); + scan_cancel_token.cancel(); + return; + } + Ok(record) if record.cancel_requested => scan_cancel_token.cancel(), Ok(_) => {} Err(err) => { warn!( @@ -724,20 +735,28 @@ async fn start_manual_transition_job( } } - let cancel_token = CancellationToken::new(); - insert_active_manual_transition_job(job_id, cancel_token.clone()); + let scan_cancel_token = CancellationToken::new(); + let heartbeat_shutdown_token = CancellationToken::new(); + insert_active_manual_transition_job(job_id, scan_cancel_token.clone()); let mut run_options = options; - run_options.cancel_token = Some(cancel_token.clone()); + run_options.job_id = Some(job_id); + run_options.cancel_token = Some(scan_cancel_token.clone()); run_options.cancel_check = Some(manual_transition_durable_cancel_check(store.clone(), job_id)); run_options.progress_sink = Some(manual_transition_progress_sink(store.clone(), job_id)); let run_store = store.clone(); - spawn_manual_transition_job_heartbeat(store, job_id, cancel_token); + let job_scan_cancel_token = scan_cancel_token.clone(); + let job_heartbeat_shutdown_token = heartbeat_shutdown_token.clone(); + spawn_manual_transition_job_heartbeat(store, job_id, scan_cancel_token, heartbeat_shutdown_token); tokio::spawn(async move { let result = enqueue_transition_for_existing_objects_scoped(run_store.clone(), &bucket, run_options).await; - if let Some(final_record) = finalize_manual_transition_job(run_store.clone(), job_id, result).await { + if let Some(final_record) = finalize_manual_transition_job(run_store.clone(), job_id, result).await + && final_record.is_terminal() + { release_manual_transition_admission(run_store, &final_record); + job_scan_cancel_token.cancel(); + job_heartbeat_shutdown_token.cancel(); + remove_active_manual_transition_job(job_id); } - remove_active_manual_transition_job(job_id); }); Ok(StartManualTransitionJobResult::Started(Box::new(record))) @@ -1139,6 +1158,26 @@ mod tests { assert_eq!(response_state(&report), "partial"); } + #[test] + fn manual_transition_response_reports_partial_for_tier_failure() { + let report = ManualTransitionRunReport { + tier_failure: 1, + ..Default::default() + }; + + assert_eq!(response_state(&report), "partial"); + } + + #[test] + fn manual_transition_response_reports_partial_for_worker_failure() { + let report = ManualTransitionRunReport { + transition_failed: 1, + ..Default::default() + }; + + assert_eq!(response_state(&report), "partial"); + } + #[test] fn manual_transition_response_omits_raw_resume_markers() { let report = ManualTransitionRunReport { @@ -1212,6 +1251,17 @@ mod tests { assert!(active_manual_transition_cancel_token(job_id).is_none()); } + #[test] + fn manual_transition_heartbeat_keeps_running_after_scan_cancel() { + let src = include_str!("ilm_transition.rs"); + let heartbeat_block = + extract_block_between_markers(src, "fn spawn_manual_transition_job_heartbeat", "enum StartManualTransitionJobResult"); + + assert!(heartbeat_block.contains("scan_cancel_token.cancel()")); + assert!(heartbeat_block.contains("shutdown_token.cancelled()")); + assert!(!heartbeat_block.contains("scan_cancel_token.cancelled()")); + } + #[tokio::test] async fn manual_transition_job_handlers_reject_missing_credentials_before_status_contract() { let status_err = ManualTransitionJobStatusHandler {}