From 7cacd1f558413c03d6c57c8dacbca44cad0fab27 Mon Sep 17 00:00:00 2001 From: houseme Date: Sun, 26 Jul 2026 20:38:41 +0800 Subject: [PATCH] feat(ilm): persist manual transition task journals (#5296) * feat(ilm): persist manual transition task journal Co-Authored-By: heihutu * fix(ilm): reconcile manual task journals on recovery Co-Authored-By: heihutu --------- Co-authored-by: heihutu --- .../bucket/lifecycle/bucket_lifecycle_ops.rs | 359 +++++++++++++++--- .../bucket/lifecycle/manual_transition_job.rs | 313 ++++++++++++++- 2 files changed, 618 insertions(+), 54 deletions(-) diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 2e6650dc0..5537da57c 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -23,12 +23,13 @@ use crate::bucket::lifecycle::lifecycle::{ }; use crate::bucket::lifecycle::manual_transition_job::{ MANUAL_TRANSITION_JOB_RECORD_PREFIX, ManualTransitionJobRecord, ManualTransitionJobState, ManualTransitionScopeAdmission, - 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, manual_transition_worker_result_task_key, persist_manual_transition_job_progress, - reconcile_manual_transition_worker_results, record_manual_transition_worker_result, renew_manual_transition_job_lease, - save_manual_transition_job_record_if_current, + ManualTransitionScopeAdmissionClaim, ManualTransitionTaskRecord, 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, + manual_transition_worker_result_task_key, persist_manual_transition_job_progress, reconcile_manual_transition_worker_results, + record_manual_transition_worker_result, renew_manual_transition_job_lease, save_manual_transition_job_record_if_current, + save_manual_transition_task_if_absent, }; use crate::bucket::lifecycle::replication_sink; use crate::bucket::lifecycle::replication_sink::{ @@ -1155,6 +1156,7 @@ enum TransitionEnqueueOutcome { QueueFull, QueueClosed, QueueSendTimedOut, + TaskJournalFailed, } impl TransitionEnqueueOutcome { @@ -1401,6 +1403,7 @@ impl TransitionState { async fn queue_transition_task_outcome( self: &Arc, + api: Option>, oi: &ObjectInfo, event: &lifecycle::Event, src: &LcEventSrc, @@ -1422,13 +1425,58 @@ impl TransitionState { return TransitionEnqueueOutcome::AlreadyInFlight; } + let manual_result_key = + manual_job_id.map(|_| manual_transition_worker_result_task_key(&oi.bucket, &oi.name, oi.version_id)); + if let (Some(job_id), Some(result_key)) = (manual_job_id, manual_result_key.as_deref()) { + let Some(api) = api else { + self.release_transition(oi); + warn!( + event = EVENT_LIFECYCLE_WORKER_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + bucket = %oi.bucket, + object = %oi.name, + version_id = %oi.version_id.map(|v| v.to_string()).unwrap_or_default(), + job_id = %job_id, + state = "manual_transition_task_journal_missing_store", + "Manual transition task was not enqueued because no task journal store was available" + ); + self.record_scanner_transition_state(); + return TransitionEnqueueOutcome::TaskJournalFailed; + }; + let task_record = ManualTransitionTaskRecord::new( + job_id, + result_key, + oi.bucket.clone(), + oi.name.clone(), + oi.version_id, + event.storage_class.clone(), + ); + if let Err(err) = save_manual_transition_task_if_absent(api, &task_record).await { + self.release_transition(oi); + warn!( + event = EVENT_LIFECYCLE_WORKER_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + bucket = %oi.bucket, + object = %oi.name, + version_id = %oi.version_id.map(|v| v.to_string()).unwrap_or_default(), + job_id = %job_id, + error = %err, + state = "manual_transition_task_journal_failed", + "Manual transition task was not enqueued because its task journal marker could not be persisted" + ); + self.record_scanner_transition_state(); + return TransitionEnqueueOutcome::TaskJournalFailed; + } + } + let task = TransitionTask { obj_info: oi.clone(), src: src.clone(), event: event.clone(), manual_job_id, - manual_result_key: manual_job_id - .map(|_| manual_transition_worker_result_task_key(&oi.bucket, &oi.name, oi.version_id)), + manual_result_key, }; if is_immediate_transition_source(src) { let outcome = match self.transition_tx.try_send(Some(task)) { @@ -1510,7 +1558,9 @@ 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, None).await.is_handled() + self.queue_transition_task_outcome(None, oi, event, src, None) + .await + .is_handled() } pub async fn init(api: Arc) { @@ -1963,7 +2013,7 @@ async fn recover_manual_transition_job( } let recovery_unknown_snapshot = ManualTransitionQueueSnapshot::default(); - if record.scan_completed && record.report.worker_transition_pending() { + if record.scan_completed { let reconciled = reconcile_manual_transition_worker_results(api.clone(), job_id, recovery_unknown_snapshot).await?; if reconciled.is_terminal() { release_manual_transition_recovery_admission(api, &reconciled).await; @@ -3246,6 +3296,9 @@ impl ManualTransitionRunReport { TransitionEnqueueOutcome::QueueSendTimedOut => { self.skipped_queue_timeout = self.skipped_queue_timeout.saturating_add(1); } + TransitionEnqueueOutcome::TaskJournalFailed => { + self.skipped_queue_closed = self.skipped_queue_closed.saturating_add(1); + } } } @@ -3391,7 +3444,7 @@ pub async fn enqueue_transition_for_existing_objects_scoped( return Ok(report); } report.scanned = report.scanned.saturating_add(1); - enqueue_transition_with_lifecycle_report(object, &lc, &src, &options, &mut report).await; + enqueue_transition_with_lifecycle_report(Some(api.clone()), object, &lc, &src, &options, &mut report).await; if report.has_partial_enqueue() { report.next_marker.clone_from(&previous_marker); report.next_version_idmarker.clone_from(&previous_version_marker); @@ -3656,6 +3709,7 @@ fn manual_transition_version_marker(oi: &ObjectInfo) -> String { } async fn enqueue_transition_with_lifecycle_report( + api: Option>, oi: &ObjectInfo, lc: &BucketLifecycleConfiguration, src: &LcEventSrc, @@ -3713,7 +3767,7 @@ async fn enqueue_transition_with_lifecycle_report( return true; } let outcome = runtime_sources::transition_state_handle() - .queue_transition_task_outcome(oi, &event, src, options.job_id) + .queue_transition_task_outcome(api.clone(), oi, &event, src, options.job_id) .await; report.record_enqueue_outcome(outcome); return outcome.is_handled(); @@ -4606,13 +4660,12 @@ mod tests { manual_transition_duration_elapsed, manual_transition_has_more_after_limit, manual_transition_recovery_progress_sink, manual_transition_version_marker, mark_delete_opts_skip_decommissioned_on_remote_success, merge_stale_multipart_candidate, persist_manual_transition_job_progress, persist_manual_transition_page_checkpoint, - recover_manual_transition_job, recover_manual_transition_jobs, recover_manual_transition_jobs_once, - replication_state_for_delete, resolve_tier_free_version_recovery_enabled, 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_lifecycle_observability_observer, - 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, + recover_manual_transition_job, recover_manual_transition_jobs, replication_state_for_delete, + resolve_tier_free_version_recovery_enabled, 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_lifecycle_observability_observer, 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}; @@ -4623,14 +4676,15 @@ mod tests { use crate::bucket::lifecycle::config_boundary; use crate::bucket::lifecycle::manual_transition_job::{ ManualTransitionJobRecord, ManualTransitionJobState, ManualTransitionScopeAdmission, ManualTransitionScopeAdmissionClaim, - ManualTransitionWorkerResult, ManualTransitionWorkerResultRecord, claim_manual_transition_scope_admission, - delete_manual_transition_scope_admission_if_current, legacy_manual_transition_scope_key, - load_manual_transition_job_record, load_manual_transition_scope_admission, - load_manual_transition_scope_admission_with_etag, manual_transition_scope_record_object_name, - manual_transition_worker_result_object_name, manual_transition_worker_result_task_key, - reconcile_manual_transition_worker_results, record_manual_transition_worker_result, renew_manual_transition_job_lease, - request_manual_transition_job_cancel, save_manual_transition_job_record, - save_manual_transition_scope_admission_if_absent, save_manual_transition_scope_admission_if_current, + ManualTransitionTaskRecord, ManualTransitionWorkerResult, ManualTransitionWorkerResultRecord, + claim_manual_transition_scope_admission, delete_manual_transition_scope_admission_if_current, + legacy_manual_transition_scope_key, load_manual_transition_job_record, load_manual_transition_scope_admission, + load_manual_transition_scope_admission_with_etag, load_manual_transition_task_record, + manual_transition_scope_record_object_name, manual_transition_worker_result_object_name, + manual_transition_worker_result_task_key, reconcile_manual_transition_worker_results, + record_manual_transition_worker_result, renew_manual_transition_job_lease, request_manual_transition_job_cancel, + save_manual_transition_job_record, save_manual_transition_scope_admission_if_absent, + save_manual_transition_scope_admission_if_current, save_manual_transition_task_if_absent, save_manual_transition_worker_result_if_absent, }; use crate::bucket::lifecycle::replication_sink::{ @@ -6823,10 +6877,10 @@ mod tests { }; let first = state - .queue_transition_task_outcome(&object, &event, &LcEventSrc::Scanner, None) + .queue_transition_task_outcome(None, &object, &event, &LcEventSrc::Scanner, None) .await; let second = state - .queue_transition_task_outcome(&object, &event, &LcEventSrc::Scanner, None) + .queue_transition_task_outcome(None, &object, &event, &LcEventSrc::Scanner, None) .await; assert_eq!(first, TransitionEnqueueOutcome::Queued); @@ -6854,10 +6908,10 @@ mod tests { }; let first = state - .queue_transition_task_outcome(&first_object, &event, &LcEventSrc::Scanner, None) + .queue_transition_task_outcome(None, &first_object, &event, &LcEventSrc::Scanner, None) .await; let second = state - .queue_transition_task_outcome(&second_object, &event, &LcEventSrc::Scanner, None) + .queue_transition_task_outcome(None, &second_object, &event, &LcEventSrc::Scanner, None) .await; assert_eq!(first, TransitionEnqueueOutcome::Queued); @@ -6865,6 +6919,73 @@ mod tests { assert_eq!(state.transition_rx.len(), 1); } + #[tokio::test] + #[serial] + async fn queue_transition_task_outcome_persists_manual_task_journal() { + let (_paths, ecstore) = setup_test_env().await; + let state = TransitionState::new_with_capacity(4); + let job_id = Uuid::new_v4(); + let version_id = Uuid::new_v4(); + let object = ObjectInfo { + bucket: "manual-task-journal-bucket".to_string(), + name: "logs/object".to_string(), + version_id: Some(version_id), + ..Default::default() + }; + let event = crate::bucket::lifecycle::lifecycle::Event { + action: IlmAction::TransitionAction, + storage_class: "WARM".to_string(), + ..Default::default() + }; + let task_key = manual_transition_worker_result_task_key(&object.bucket, &object.name, object.version_id); + + let outcome = state + .queue_transition_task_outcome(Some(ecstore.clone()), &object, &event, &LcEventSrc::Scanner, Some(job_id)) + .await; + + assert_eq!(outcome, TransitionEnqueueOutcome::Queued); + assert_eq!(state.transition_rx.len(), 1); + let task_record = load_manual_transition_task_record(ecstore, job_id, &task_key) + .await + .expect("manual task journal marker should load"); + assert_eq!(task_record.job_id, job_id); + assert_eq!(task_record.task_key, task_key); + assert_eq!(task_record.bucket, object.bucket); + assert_eq!(task_record.object, object.name); + assert_eq!(task_record.version_id, Some(version_id)); + assert_eq!(task_record.storage_class, "WARM"); + } + + #[tokio::test] + #[serial] + async fn queue_transition_task_outcome_fails_closed_when_manual_task_journal_fails() { + let (_paths, ecstore) = setup_test_env().await; + let state = TransitionState::new_with_capacity(4); + let object = ObjectInfo { + bucket: "manual-task-journal-fail-bucket".to_string(), + name: "logs/object".to_string(), + ..Default::default() + }; + let event = crate::bucket::lifecycle::lifecycle::Event { + action: IlmAction::TransitionAction, + storage_class: "WARM".to_string(), + ..Default::default() + }; + + let failed = state + .queue_transition_task_outcome(Some(ecstore.clone()), &object, &event, &LcEventSrc::Scanner, Some(Uuid::nil())) + .await; + + assert_eq!(failed, TransitionEnqueueOutcome::TaskJournalFailed); + assert_eq!(state.transition_rx.len(), 0); + + let retried = state + .queue_transition_task_outcome(Some(ecstore), &object, &event, &LcEventSrc::Scanner, Some(Uuid::new_v4())) + .await; + assert_eq!(retried, TransitionEnqueueOutcome::Queued); + assert_eq!(state.transition_rx.len(), 1); + } + #[tokio::test] #[serial] async fn queue_transition_task_dedupes_immediate_and_scanner_sources_for_same_version() { @@ -7322,7 +7443,8 @@ mod tests { }; let mut report = ManualTransitionRunReport::new(&object.bucket, &options); - let handled = enqueue_transition_with_lifecycle_report(&object, &lc, &LcEventSrc::Scanner, &options, &mut report).await; + let handled = + enqueue_transition_with_lifecycle_report(None, &object, &lc, &LcEventSrc::Scanner, &options, &mut report).await; assert!(handled); assert_eq!(report.eligible, 1); @@ -7343,7 +7465,8 @@ mod tests { }; let mut report = ManualTransitionRunReport::new(&object.bucket, &options); - let handled = enqueue_transition_with_lifecycle_report(&object, &lc, &LcEventSrc::Scanner, &options, &mut report).await; + let handled = + enqueue_transition_with_lifecycle_report(None, &object, &lc, &LcEventSrc::Scanner, &options, &mut report).await; assert!(!handled); assert_eq!(report.eligible, 0); @@ -7361,7 +7484,8 @@ mod tests { }; let mut report = ManualTransitionRunReport::new(&object.bucket, &options); - let handled = enqueue_transition_with_lifecycle_report(&object, &lc, &LcEventSrc::Scanner, &options, &mut report).await; + let handled = + enqueue_transition_with_lifecycle_report(None, &object, &lc, &LcEventSrc::Scanner, &options, &mut report).await; assert!(!handled); assert_eq!(report.eligible, 0); @@ -7375,7 +7499,8 @@ mod tests { 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; + let handled = + enqueue_transition_with_lifecycle_report(None, &object, &lc, &LcEventSrc::Scanner, &options, &mut report).await; assert!(!handled); assert_eq!(report.eligible, 0); @@ -7426,7 +7551,8 @@ mod tests { }; let mut report = ManualTransitionRunReport::new(&object.bucket, &options); - let handled = enqueue_transition_with_lifecycle_report(&object, &lc, &LcEventSrc::Scanner, &options, &mut report).await; + let handled = + enqueue_transition_with_lifecycle_report(None, &object, &lc, &LcEventSrc::Scanner, &options, &mut report).await; assert!(!handled); assert_eq!(report.eligible, 0); @@ -7920,13 +8046,11 @@ mod tests { .await .expect("expired scope admission should save"); - let stats = recover_manual_transition_jobs_once(ecstore.clone(), 10, None) + let outcome = recover_manual_transition_job(ecstore.clone(), job_id, ManualTransitionQueueSnapshot::default()) .await .expect("manual transition recovery should run"); - assert!(stats.scanned >= 1); - assert_eq!(stats.resumed, 1); - assert_eq!(stats.failed, 0); + assert_eq!(outcome, ManualTransitionJobRecoveryOutcome::Resumed); let recovered = load_manual_transition_job_record(ecstore.clone(), job_id) .await .expect("recovered job should load"); @@ -8214,14 +8338,11 @@ mod tests { .await .expect("expired cancelled scope admission should save"); - let stats = recover_manual_transition_jobs(ecstore.clone(), 10) + let outcome = recover_manual_transition_job(ecstore.clone(), job_id, ManualTransitionQueueSnapshot::default()) .await .expect("manual transition recovery should process cancelled jobs"); - assert_eq!(stats.cancelled, 1); - assert_eq!(stats.unknown, 0); - assert_eq!(stats.resumed, 0); - assert_eq!(stats.failed, 0); + assert_eq!(outcome, ManualTransitionJobRecoveryOutcome::Cancelled); let recovered = load_manual_transition_job_record(ecstore.clone(), job_id) .await .expect("cancelled job should load"); @@ -8424,6 +8545,52 @@ mod tests { ); } + #[tokio::test] + #[serial] + async fn manual_transition_worker_result_heartbeat_uses_task_journal_enqueued_floor() { + let (_paths, ecstore) = setup_test_env().await; + let job_id = Uuid::new_v4(); + let bucket = format!("manual-task-heartbeat-{}", job_id.simple()); + let mut record = ManualTransitionJobRecord::new(job_id, &bucket, &ManualTransitionRunOptions::default(), "worker-owner"); + record.scan_completed = true; + save_manual_transition_job_record(ecstore.clone(), &record) + .await + .expect("worker heartbeat task journal job record should save"); + save_manual_transition_scope_admission_if_absent(ecstore.clone(), &ManualTransitionScopeAdmission::from_job(&record)) + .await + .expect("worker heartbeat task journal admission should save"); + let task_key = manual_transition_worker_result_task_key(&bucket, "logs/a", None); + let task_marker = ManualTransitionTaskRecord::new(job_id, &task_key, &bucket, "logs/a", None, "WARM"); + assert!( + save_manual_transition_task_if_absent(ecstore.clone(), &task_marker) + .await + .expect("task journal marker should save"), + "new task journal marker must be created" + ); + let result_marker = ManualTransitionWorkerResultRecord::new(job_id, &task_key, ManualTransitionWorkerResult::Completed); + assert!( + save_manual_transition_worker_result_if_absent(ecstore.clone(), &result_marker) + .await + .expect("worker result marker should save"), + "new worker result marker must be created" + ); + + let renewed = renew_manual_transition_job_lease(ecstore.clone(), job_id, ManualTransitionQueueSnapshot::default()) + .await + .expect("heartbeat should reconcile task and result journals"); + + assert_eq!(renewed.state, ManualTransitionJobState::Completed); + assert_eq!(renewed.report.enqueued, 1); + assert_eq!(renewed.report.transition_completed, 1); + assert!( + matches!( + load_manual_transition_scope_admission(ecstore, &record.scope_key).await, + Err(Error::ConfigNotFound) + ), + "terminal task journal heartbeat must release the scope admission" + ); + } + #[tokio::test] #[serial] async fn manual_transition_worker_result_recovery_applies_marker_before_record_update() { @@ -8474,6 +8641,108 @@ mod tests { ); } + #[tokio::test] + #[serial] + async fn manual_transition_recovery_uses_task_journal_as_enqueued_floor() { + let (_paths, ecstore) = setup_test_env().await; + let job_id = Uuid::new_v4(); + let bucket = format!("manual-task-floor-{}", job_id.simple()); + let mut record = ManualTransitionJobRecord::new(job_id, &bucket, &ManualTransitionRunOptions::default(), "old-owner"); + record.scan_completed = true; + record.lease_expires_at_unix_nanos = 0; + save_manual_transition_job_record(ecstore.clone(), &record) + .await + .expect("task journal recovery job record should save"); + save_manual_transition_scope_admission_if_absent(ecstore.clone(), &ManualTransitionScopeAdmission::from_job(&record)) + .await + .expect("task journal recovery admission should save"); + let task_key = manual_transition_worker_result_task_key(&bucket, "logs/a", None); + let task_marker = ManualTransitionTaskRecord::new(job_id, &task_key, &bucket, "logs/a", None, "WARM"); + assert!( + save_manual_transition_task_if_absent(ecstore.clone(), &task_marker) + .await + .expect("task journal marker should save"), + "new task journal marker must be created" + ); + let result_marker = ManualTransitionWorkerResultRecord::new(job_id, &task_key, ManualTransitionWorkerResult::Completed); + assert!( + save_manual_transition_worker_result_if_absent(ecstore.clone(), &result_marker) + .await + .expect("worker result marker should save"), + "new worker result marker must be created" + ); + + let outcome = recover_manual_transition_job(ecstore.clone(), job_id, ManualTransitionQueueSnapshot::default()) + .await + .expect("recovery should reconcile task and result journals"); + + assert_eq!(outcome, ManualTransitionJobRecoveryOutcome::Resumed); + let recovered = load_manual_transition_job_record(ecstore.clone(), job_id) + .await + .expect("reconciled task journal job should load"); + assert_eq!(recovered.state, ManualTransitionJobState::Completed); + assert_eq!(recovered.report.enqueued, 1); + assert_eq!(recovered.report.transition_completed, 1); + assert!( + matches!( + load_manual_transition_scope_admission(ecstore, &record.scope_key).await, + Err(Error::ConfigNotFound) + ), + "terminal task journal recovery must release the scope admission" + ); + } + + #[tokio::test] + #[serial] + async fn manual_transition_recovery_marks_unknown_for_task_journal_without_result() { + let (_paths, ecstore) = setup_test_env().await; + let job_id = Uuid::new_v4(); + let bucket = format!("manual-task-lost-result-{}", job_id.simple()); + let mut record = ManualTransitionJobRecord::new(job_id, &bucket, &ManualTransitionRunOptions::default(), "old-owner"); + record.scan_completed = true; + record.lease_expires_at_unix_nanos = 0; + save_manual_transition_job_record(ecstore.clone(), &record) + .await + .expect("lost-result job record should save"); + save_manual_transition_scope_admission_if_absent(ecstore.clone(), &ManualTransitionScopeAdmission::from_job(&record)) + .await + .expect("lost-result admission should save"); + let task_key = manual_transition_worker_result_task_key(&bucket, "logs/a", None); + let task_marker = ManualTransitionTaskRecord::new(job_id, &task_key, &bucket, "logs/a", None, "WARM"); + assert!( + save_manual_transition_task_if_absent(ecstore.clone(), &task_marker) + .await + .expect("task journal marker should save"), + "new task journal marker must be created" + ); + + let outcome = recover_manual_transition_job(ecstore.clone(), job_id, ManualTransitionQueueSnapshot::default()) + .await + .expect("recovery should fail closed when a task journal marker has no worker result"); + + assert_eq!(outcome, ManualTransitionJobRecoveryOutcome::Unknown); + let recovered = load_manual_transition_job_record(ecstore.clone(), job_id) + .await + .expect("unknown task journal job should load"); + assert_eq!(recovered.state, ManualTransitionJobState::Unknown); + assert_eq!(recovered.report.enqueued, 1); + assert_eq!(recovered.report.transition_completed, 0); + assert_eq!(recovered.report.transition_failed, 0); + assert!( + recovered + .error + .as_deref() + .is_some_and(|error| error.contains("worker result was not persisted")) + ); + assert!( + matches!( + load_manual_transition_scope_admission(ecstore, &record.scope_key).await, + Err(Error::ConfigNotFound) + ), + "unknown task journal recovery must release the scope admission" + ); + } + #[tokio::test] #[serial] async fn manual_transition_worker_result_recovery_marks_unknown_for_corrupt_marker() { diff --git a/crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs b/crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs index 5fbdcccb0..b45d55e1a 100644 --- a/crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs +++ b/crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs @@ -31,14 +31,18 @@ use crate::storage_api_contracts::object::HTTPPreconditions; use crate::store::ECStore; pub const MANUAL_TRANSITION_JOB_SCHEMA: &str = "rustfs-manual-transition-job-v1"; +pub const MANUAL_TRANSITION_TASK_SCHEMA: &str = "rustfs-manual-transition-task-v1"; pub const MANUAL_TRANSITION_WORKER_RESULT_SCHEMA: &str = "rustfs-manual-transition-worker-result-v1"; pub const MANUAL_TRANSITION_JOB_RECORD_PREFIX: &str = "ilm/manual-transition/jobs"; pub const MANUAL_TRANSITION_SCOPE_RECORD_PREFIX: &str = "ilm/manual-transition/scopes"; +pub const MANUAL_TRANSITION_TASK_PREFIX: &str = "ilm/manual-transition/tasks"; pub const MANUAL_TRANSITION_WORKER_RESULT_PREFIX: &str = "ilm/manual-transition/results"; pub const MAX_MANUAL_TRANSITION_JOB_RECORD_SIZE: usize = 64 * 1024; +pub const MAX_MANUAL_TRANSITION_TASK_RECORD_SIZE: usize = 16 * 1024; pub const MAX_MANUAL_TRANSITION_WORKER_RESULT_RECORD_SIZE: usize = 8 * 1024; const MANUAL_TRANSITION_JOB_LEASE_SECONDS: i128 = 60; const MANUAL_TRANSITION_LEGACY_SCOPE_SCAN_LIMIT: i32 = 1000; +const MANUAL_TRANSITION_TASK_SCAN_LIMIT: i32 = 1000; const MANUAL_TRANSITION_WORKER_RESULT_SCAN_LIMIT: i32 = 1000; fn is_false(value: &bool) -> bool { @@ -160,6 +164,21 @@ impl ManualTransitionJobRecord { true } + pub fn mark_unknown_for_task_journal_error( + &mut self, + error: impl Into, + queue_snapshot: ManualTransitionQueueSnapshot, + ) -> bool { + if self.state != ManualTransitionJobState::Running || !self.scan_completed { + return false; + } + self.queue_snapshot = queue_snapshot; + self.state = ManualTransitionJobState::Unknown; + self.error = Some(format!("manual transition task journal is corrupt: {}", error.into())); + self.mark_updated_terminal(); + true + } + pub fn cancel_after_recovery(&mut self, queue_snapshot: ManualTransitionQueueSnapshot) { if self.state == ManualTransitionJobState::Running && self.cancel_requested { self.scan_completed = true; @@ -189,11 +208,18 @@ impl ManualTransitionJobRecord { self.mark_terminal_if_worker_drained(); } - fn apply_worker_result_counts(&mut self, completed: u64, failed: u64, queue_snapshot: ManualTransitionQueueSnapshot) -> bool { + fn apply_worker_result_counts( + &mut self, + completed: u64, + failed: u64, + task_queued: u64, + queue_snapshot: ManualTransitionQueueSnapshot, + ) -> bool { if self.is_terminal() { return false; } - if completed.saturating_add(failed) > self.report.enqueued { + let enqueued = self.report.enqueued.max(task_queued); + if completed.saturating_add(failed) > enqueued { self.queue_snapshot = queue_snapshot; self.state = ManualTransitionJobState::Unknown; self.error = Some("manual transition worker result journal exceeds enqueued count".to_string()); @@ -202,10 +228,14 @@ impl ManualTransitionJobRecord { } let transition_completed = self.report.transition_completed.max(completed); let transition_failed = self.report.transition_failed.max(failed); - if transition_completed == self.report.transition_completed && transition_failed == self.report.transition_failed { + if enqueued == self.report.enqueued + && transition_completed == self.report.transition_completed + && transition_failed == self.report.transition_failed + { return false; } let scan_tier_failure = self.report.tier_failure.saturating_sub(self.report.transition_failed); + self.report.enqueued = enqueued; self.report.transition_completed = transition_completed; self.report.transition_failed = transition_failed; self.report.tier_failure = scan_tier_failure.saturating_add(transition_failed); @@ -415,6 +445,111 @@ pub enum ManualTransitionWorkerResult { TierFailure, } +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ManualTransitionTaskRecord { + pub schema: String, + pub job_id: Uuid, + pub task_key: String, + pub bucket: String, + pub object: String, + pub version_id: Option, + pub storage_class: String, + pub queued_at_unix_nanos: i128, +} + +impl ManualTransitionTaskRecord { + pub fn new( + job_id: Uuid, + task_key: impl Into, + bucket: impl Into, + object: impl Into, + version_id: Option, + storage_class: impl Into, + ) -> Self { + Self { + schema: MANUAL_TRANSITION_TASK_SCHEMA.to_string(), + job_id, + task_key: task_key.into(), + bucket: bucket.into(), + object: object.into(), + version_id, + storage_class: storage_class.into(), + queued_at_unix_nanos: OffsetDateTime::now_utc().unix_timestamp_nanos(), + } + } + + pub fn encode(&self) -> Result, ManualTransitionJobError> { + self.validate()?; + let record_bytes = serde_json::to_vec(self)?; + let content_sha256 = hex_sha256(&record_bytes, ToOwned::to_owned); + let persisted = PersistedManualTransitionTaskRecord { + schema: MANUAL_TRANSITION_TASK_SCHEMA.to_string(), + content_sha256, + record: self.clone(), + }; + let encoded = serde_json::to_vec(&persisted)?; + if encoded.len() > MAX_MANUAL_TRANSITION_TASK_RECORD_SIZE { + return Err(ManualTransitionJobError::Corrupt("encoded task record exceeds maximum size")); + } + Ok(encoded) + } + + pub fn decode(expected_job_id: Uuid, expected_task_key: &str, data: &[u8]) -> Result { + if data.len() > MAX_MANUAL_TRANSITION_TASK_RECORD_SIZE { + return Err(ManualTransitionJobError::Corrupt("encoded task record exceeds maximum size")); + } + let persisted: PersistedManualTransitionTaskRecord = serde_json::from_slice(data)?; + if persisted.schema != MANUAL_TRANSITION_TASK_SCHEMA { + return Err(ManualTransitionJobError::UnsupportedSchema(persisted.schema)); + } + if !is_sha256_checksum(&persisted.content_sha256) { + return Err(ManualTransitionJobError::Corrupt("task record content checksum is not a sha256 checksum")); + } + let record = persisted.record; + let record_bytes = serde_json::to_vec(&record)?; + let actual_checksum = hex_sha256(&record_bytes, ToOwned::to_owned); + if persisted.content_sha256 != actual_checksum { + return Err(ManualTransitionJobError::ChecksumMismatch); + } + if record.job_id != expected_job_id { + return Err(ManualTransitionJobError::Corrupt("task record job_id does not match record key")); + } + if record.task_key != expected_task_key { + return Err(ManualTransitionJobError::Corrupt("task record task_key does not match record key")); + } + record.validate()?; + Ok(record) + } + + fn validate(&self) -> Result<(), ManualTransitionJobError> { + if self.job_id.is_nil() { + return Err(ManualTransitionJobError::Corrupt("task record job_id is nil")); + } + if !is_sha256_checksum(&self.task_key) { + return Err(ManualTransitionJobError::Corrupt("task record task_key is not a sha256 checksum")); + } + if self.bucket.is_empty() { + return Err(ManualTransitionJobError::Corrupt("task record bucket is empty")); + } + if self.object.is_empty() { + return Err(ManualTransitionJobError::Corrupt("task record object is empty")); + } + if self.storage_class.trim().is_empty() { + return Err(ManualTransitionJobError::Corrupt("task record storage_class is empty")); + } + Ok(()) + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct PersistedManualTransitionTaskRecord { + schema: String, + content_sha256: String, + record: ManualTransitionTaskRecord, +} + #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub struct ManualTransitionWorkerResultStats { pub completed: u64, @@ -661,6 +796,17 @@ pub fn manual_transition_worker_result_object_prefix(job_id: Uuid) -> Result Result { + manual_transition_job_sharded_prefix(MANUAL_TRANSITION_TASK_PREFIX, job_id) +} + +pub fn manual_transition_task_object_name(job_id: Uuid, task_key: &str) -> Result { + if !is_sha256_checksum(task_key) { + return Err(ManualTransitionJobError::Corrupt("task record task_key is not a sha256 checksum")); + } + Ok(format!("{}/{}.json", manual_transition_task_object_prefix(job_id)?, task_key)) +} + pub fn manual_transition_worker_result_object_name(job_id: Uuid, task_key: &str) -> Result { if !is_sha256_checksum(task_key) { return Err(ManualTransitionJobError::Corrupt("worker result task_key is not a sha256 checksum")); @@ -687,6 +833,22 @@ fn manual_transition_worker_result_task_key_from_object_name( Ok(task_key.to_string()) } +fn manual_transition_task_key_from_object_name(job_id: Uuid, object_name: &str) -> Result { + let prefix = manual_transition_task_object_prefix(job_id)?; + let rest = object_name + .strip_prefix(&prefix) + .ok_or(ManualTransitionJobError::Corrupt("task record object prefix is invalid"))? + .strip_prefix('/') + .ok_or(ManualTransitionJobError::Corrupt("task record object path is invalid"))?; + let task_key = rest + .strip_suffix(".json") + .ok_or(ManualTransitionJobError::Corrupt("task record suffix is invalid"))?; + if task_key.contains('/') || !is_sha256_checksum(task_key) { + return Err(ManualTransitionJobError::Corrupt("task record task_key is invalid")); + } + Ok(task_key.to_string()) +} + pub fn manual_transition_job_id_from_record_object_name(object_name: &str) -> Result { let Some(rest) = object_name.strip_prefix(MANUAL_TRANSITION_JOB_RECORD_PREFIX) else { return Err(ManualTransitionJobError::Corrupt("job record object prefix is invalid")); @@ -812,6 +974,97 @@ pub(crate) async fn save_manual_transition_worker_result_if_absent( } } +pub(crate) async fn save_manual_transition_task_if_absent( + api: Arc, + record: &ManualTransitionTaskRecord, +) -> EcstoreResult { + let object = + manual_transition_task_object_name(record.job_id, &record.task_key).map_err(manual_transition_job_store_error)?; + let data = record.encode().map_err(manual_transition_job_store_error)?; + match config_boundary::save_config_with_opts( + api, + &object, + data, + &ObjectOptions { + max_parity: true, + http_preconditions: Some(HTTPPreconditions { + if_none_match: Some("*".to_string()), + ..Default::default() + }), + ..Default::default() + }, + ) + .await + { + Ok(()) => Ok(true), + Err(Error::PreconditionFailed) => Ok(false), + Err(err) => Err(err), + } +} + +pub async fn load_manual_transition_task_record( + api: Arc, + job_id: Uuid, + task_key: &str, +) -> EcstoreResult { + let object = manual_transition_task_object_name(job_id, task_key).map_err(manual_transition_job_store_error)?; + let data = config_boundary::read_config(api, &object).await?; + ManualTransitionTaskRecord::decode(job_id, task_key, &data).map_err(manual_transition_job_store_error) +} + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +struct ManualTransitionTaskJournalStats { + queued: u64, +} + +enum ManualTransitionTaskJournal { + Stats(ManualTransitionTaskJournalStats), + Corrupt(String), +} + +async fn scan_manual_transition_task_journal(api: Arc, job_id: Uuid) -> EcstoreResult { + let prefix = manual_transition_task_object_prefix(job_id).map_err(manual_transition_job_store_error)?; + let mut marker = None; + let mut stats = ManualTransitionTaskJournalStats::default(); + loop { + let page = api + .clone() + .list_objects_v2( + RUSTFS_META_BUCKET, + &prefix, + marker, + None, + MANUAL_TRANSITION_TASK_SCAN_LIMIT, + false, + None, + false, + ) + .await?; + for object in page.objects { + let task_key = match manual_transition_task_key_from_object_name(job_id, &object.name) { + Ok(task_key) => task_key, + Err(err) => return Ok(ManualTransitionTaskJournal::Corrupt(err.to_string())), + }; + let object_name = manual_transition_task_object_name(job_id, &task_key).map_err(manual_transition_job_store_error)?; + let data = match config_boundary::read_config(api.clone(), &object_name).await { + Ok(data) => data, + Err(err) => return Err(err), + }; + if let Err(err) = ManualTransitionTaskRecord::decode(job_id, &task_key, &data) { + return Ok(ManualTransitionTaskJournal::Corrupt(err.to_string())); + } + stats.queued = stats.queued.saturating_add(1); + } + if !page.is_truncated { + return Ok(ManualTransitionTaskJournal::Stats(stats)); + } + let Some(next_marker) = page.next_continuation_token else { + return Err(Error::other("manual transition task journal page is truncated without a next marker")); + }; + marker = Some(next_marker); + } +} + pub async fn load_manual_transition_worker_result_stats( api: Arc, job_id: Uuid, @@ -880,6 +1133,12 @@ pub async fn reconcile_manual_transition_worker_results( job_id: Uuid, queue_snapshot: ManualTransitionQueueSnapshot, ) -> EcstoreResult { + let task_stats = match scan_manual_transition_task_journal(api.clone(), job_id).await? { + ManualTransitionTaskJournal::Stats(stats) => stats, + ManualTransitionTaskJournal::Corrupt(error) => { + return mark_manual_transition_job_unknown_for_task_journal_error(api, job_id, error, queue_snapshot).await; + } + }; let stats = match scan_manual_transition_worker_result_journal(api.clone(), job_id).await? { ManualTransitionWorkerResultJournal::Stats(stats) => stats, ManualTransitionWorkerResultJournal::Corrupt(error) => { @@ -888,7 +1147,7 @@ pub async fn reconcile_manual_transition_worker_results( }; for _ in 0..4 { let (mut record, etag) = load_manual_transition_job_record_with_etag(api.clone(), job_id).await?; - let changed = record.apply_worker_result_counts(stats.completed, stats.failed, queue_snapshot); + let changed = record.apply_worker_result_counts(stats.completed, stats.failed, task_stats.queued, queue_snapshot); if !changed { return Ok(record); } @@ -914,6 +1173,30 @@ pub async fn reconcile_manual_transition_worker_results( Err(Error::PreconditionFailed) } +async fn mark_manual_transition_job_unknown_for_task_journal_error( + api: Arc, + job_id: Uuid, + error: String, + 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.mark_unknown_for_task_journal_error(error.clone(), queue_snapshot) { + return Ok(record); + } + match save_manual_transition_job_record_if_current(api.clone(), &record, &etag).await { + Ok(()) => { + delete_manual_transition_scope_admission_if_current(api, &record.scope_key, record.job_id, record.lease_id) + .await?; + return Ok(record); + } + Err(Error::PreconditionFailed) => continue, + Err(err) => return Err(err), + } + } + Err(Error::PreconditionFailed) +} + async fn mark_manual_transition_job_unknown_for_worker_result_journal_error( api: Arc, job_id: Uuid, @@ -1199,11 +1482,7 @@ pub async fn renew_manual_transition_job_lease( ) -> EcstoreResult { let (mut record, mut etag) = load_manual_transition_job_record_with_etag(api.clone(), job_id).await?; if record.state == ManualTransitionJobState::Running { - if record.scan_completed - && record.report.worker_transition_pending() - && queue_snapshot.queued == 0 - && queue_snapshot.active == 0 - { + if record.scan_completed && queue_snapshot.queued == 0 && queue_snapshot.active == 0 { record = reconcile_manual_transition_worker_results(api.clone(), job_id, queue_snapshot).await?; if record.is_terminal() || !record.report.worker_transition_pending() { return Ok(record); @@ -1588,6 +1867,22 @@ mod tests { assert!(matches!(err, ManualTransitionJobError::ChecksumMismatch)); } + #[test] + fn manual_transition_task_record_rejects_checksum_drift() { + let job_id = Uuid::new_v4(); + let task_key = manual_transition_worker_result_task_key("bucket", "logs/a", Some(Uuid::new_v4())); + let record = ManualTransitionTaskRecord::new(job_id, &task_key, "bucket", "logs/a", Some(Uuid::new_v4()), "WARM"); + let encoded = record.encode().expect("task record should encode"); + let mut value: serde_json::Value = serde_json::from_slice(&encoded).expect("encoded task record should be json"); + value["record"]["storage_class"] = serde_json::Value::String("COLD".to_string()); + let mutated = serde_json::to_vec(&value).expect("mutated task record should encode"); + + let err = + ManualTransitionTaskRecord::decode(job_id, &task_key, &mutated).expect_err("task checksum drift must fail closed"); + + assert!(matches!(err, ManualTransitionJobError::ChecksumMismatch)); + } + #[test] fn manual_transition_job_record_rejects_unknown_report_fields() { let options = ManualTransitionRunOptions::default();