diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 144336b40..5229a61ec 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -25,7 +25,7 @@ use crate::bucket::lifecycle::manual_transition_job::{ MANUAL_TRANSITION_JOB_RECORD_PREFIX, ManualTransitionJobRecord, ManualTransitionJobState, ManualTransitionScopeAdmission, ManualTransitionScopeAdmissionClaim, ManualTransitionTaskRecord, ManualTransitionWorkerFailureReason, 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, + load_manual_transition_job_record, load_manual_transition_job_record_with_etag, load_manual_transition_pending_task_records, 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, record_manual_transition_worker_result_with_reason, @@ -1167,6 +1167,13 @@ impl TransitionEnqueueOutcome { } } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ManualTransitionPendingTaskReplay { + Empty, + Queued, + Deferred, +} + impl TransitionState { #[allow(clippy::new_ret_no_self)] pub fn new() -> Arc { @@ -1453,7 +1460,8 @@ impl TransitionState { oi.name.clone(), oi.version_id, event.storage_class.clone(), - ); + ) + .with_object_metadata(oi.etag.clone(), oi.mod_time, oi.size, oi.is_latest); if let Err(err) = save_manual_transition_task_if_absent(api, &task_record).await { self.release_transition(oi); warn!( @@ -2103,19 +2111,6 @@ async fn recover_manual_transition_job( } } - if record.mark_unknown_if_worker_results_lost(recovery_unknown_snapshot) - || record.mark_unknown_if_recovery_would_skip_pending_page(recovery_unknown_snapshot) - { - return match save_manual_transition_job_record_if_current(api.clone(), &record, &etag).await { - Ok(()) => { - release_manual_transition_recovery_admission(api, &record).await; - Ok(ManualTransitionJobRecoveryOutcome::Unknown) - } - Err(Error::PreconditionFailed) => Ok(ManualTransitionJobRecoveryOutcome::Skipped), - Err(err) => Err(err), - }; - } - if record.cancel_requested { record.cancel_after_recovery(queue_snapshot); return match save_manual_transition_job_record_if_current(api.clone(), &record, &etag).await { @@ -2128,6 +2123,7 @@ async fn recover_manual_transition_job( }; } + let previous_lease_id = record.lease_id; record.claim_recovery_lease(manual_transition_recovery_owner_id(), queue_snapshot); let recovery_lease_id = record.lease_id; match save_manual_transition_job_record_if_current(api.clone(), &record, &etag).await { @@ -2135,6 +2131,7 @@ async fn recover_manual_transition_job( Err(Error::PreconditionFailed) => return Ok(ManualTransitionJobRecoveryOutcome::Skipped), Err(err) => return Err(err), } + delete_manual_transition_scope_admission_if_current(api.clone(), &record.scope_key, record.job_id, previous_lease_id).await?; match claim_manual_transition_scope_admission(api.clone(), &ManualTransitionScopeAdmission::from_job(&record)).await { Ok(ManualTransitionScopeAdmissionClaim::Claimed) => {} @@ -2148,6 +2145,29 @@ async fn recover_manual_transition_job( } } + let replay = replay_manual_transition_pending_tasks(api.clone(), job_id).await?; + if matches!( + replay, + ManualTransitionPendingTaskReplay::Queued | ManualTransitionPendingTaskReplay::Deferred + ) { + spawn_manual_transition_recovery_heartbeat(api, job_id); + return Ok(ManualTransitionJobRecoveryOutcome::Resumed); + } + + let (mut record, etag) = load_manual_transition_job_record_with_etag(api.clone(), job_id).await?; + if record.mark_unknown_if_worker_results_lost(recovery_unknown_snapshot) + || record.mark_unknown_if_recovery_would_skip_pending_page(recovery_unknown_snapshot) + { + return match save_manual_transition_job_record_if_current(api.clone(), &record, &etag).await { + Ok(()) => { + release_manual_transition_recovery_admission(api, &record).await; + Ok(ManualTransitionJobRecoveryOutcome::Unknown) + } + Err(Error::PreconditionFailed) => Ok(ManualTransitionJobRecoveryOutcome::Skipped), + Err(err) => Err(err), + }; + } + 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)); @@ -2162,6 +2182,69 @@ async fn recover_manual_transition_job( Ok(ManualTransitionJobRecoveryOutcome::Resumed) } +async fn replay_manual_transition_pending_tasks( + api: Arc, + job_id: Uuid, +) -> Result { + let transition_state = runtime_sources::transition_state_handle(); + let limit = transition_state.transition_queue_capacity.max(1); + let pending = load_manual_transition_pending_task_records(api.clone(), job_id, limit).await?; + if pending.is_empty() { + return Ok(ManualTransitionPendingTaskReplay::Empty); + } + + let mut queued = 0usize; + for task in pending { + let mod_time = match task.mod_time_unix_nanos { + Some(nanos) => Some( + OffsetDateTime::from_unix_timestamp_nanos(nanos) + .map_err(|_| Error::other("manual transition task journal mod_time is invalid"))?, + ), + None => None, + }; + let object = ObjectInfo { + bucket: task.bucket, + name: task.object, + version_id: task.version_id, + etag: task.etag, + mod_time, + size: task.size.map_or(0, |size| size), + is_latest: task.is_latest.unwrap_or(false), + ..Default::default() + }; + let event = lifecycle::Event { + action: if object.version_id.is_some() { + IlmAction::TransitionVersionAction + } else { + IlmAction::TransitionAction + }, + storage_class: task.storage_class, + ..Default::default() + }; + + match transition_state + .queue_transition_task_outcome(Some(api.clone()), &object, &event, &LcEventSrc::Scanner, Some(job_id)) + .await + { + TransitionEnqueueOutcome::Queued | TransitionEnqueueOutcome::AlreadyInFlight => { + queued = queued.saturating_add(1); + } + TransitionEnqueueOutcome::QueueFull + | TransitionEnqueueOutcome::QueueClosed + | TransitionEnqueueOutcome::QueueSendTimedOut + | TransitionEnqueueOutcome::TaskJournalFailed => { + break; + } + } + } + + if queued > 0 { + Ok(ManualTransitionPendingTaskReplay::Queued) + } else { + Ok(ManualTransitionPendingTaskReplay::Deferred) + } +} + fn manual_transition_recovery_owner_id() -> &'static str { "ecstore-manual-transition-recovery" } @@ -7015,6 +7098,10 @@ mod tests { bucket: "manual-task-journal-bucket".to_string(), name: "logs/object".to_string(), version_id: Some(version_id), + etag: Some("task-etag".to_string()), + mod_time: Some(OffsetDateTime::now_utc()), + size: 42, + is_latest: true, ..Default::default() }; let event = crate::bucket::lifecycle::lifecycle::Event { @@ -7039,6 +7126,10 @@ mod tests { assert_eq!(task_record.object, object.name); assert_eq!(task_record.version_id, Some(version_id)); assert_eq!(task_record.storage_class, "WARM"); + assert_eq!(task_record.etag.as_deref(), Some("task-etag")); + assert_eq!(task_record.mod_time_unix_nanos, object.mod_time.map(|time| time.unix_timestamp_nanos())); + assert_eq!(task_record.size, Some(42)); + assert_eq!(task_record.is_latest, Some(true)); } #[tokio::test] @@ -8327,7 +8418,7 @@ mod tests { #[tokio::test] #[serial] - async fn manual_transition_recovery_marks_unknown_before_cancel_for_cursor_pending_work() { + async fn manual_transition_recovery_cancels_cursor_pending_work_when_requested() { let (_paths, ecstore) = setup_test_env().await; let job_id = Uuid::new_v4(); let continuation_token = @@ -8353,20 +8444,15 @@ mod tests { let outcome = recover_manual_transition_job(ecstore.clone(), job_id, ManualTransitionQueueSnapshot::default()) .await - .expect("manual transition recovery should fail closed before cancelled pending cursor jobs"); + .expect("manual transition recovery should cancel requested pending cursor jobs"); - assert_eq!(outcome, ManualTransitionJobRecoveryOutcome::Unknown); + assert_eq!(outcome, ManualTransitionJobRecoveryOutcome::Cancelled); let recovered = load_manual_transition_job_record(ecstore.clone(), job_id) .await - .expect("unknown cancelled job should load"); - assert_eq!(recovered.state, ManualTransitionJobState::Unknown); + .expect("cancelled job should load"); + assert_eq!(recovered.state, ManualTransitionJobState::Cancelled); assert!(recovered.cancel_requested); - assert!( - recovered - .error - .as_deref() - .is_some_and(|error| error.contains("page/task journal")) - ); + assert!(recovered.error.is_none()); } #[tokio::test] @@ -8892,8 +8978,13 @@ mod tests { #[tokio::test] #[serial] - async fn manual_transition_recovery_marks_unknown_for_task_journal_without_result() { + async fn manual_transition_recovery_replays_task_journal_without_result() { let (_paths, ecstore) = setup_test_env().await; + let transition_state = runtime_sources::transition_state_handle(); + let original_workers = transition_state.num_workers.load(Ordering::SeqCst); + let absolute_max = resolve_transition_workers_absolute_max(); + TransitionState::resize_workers_to(ecstore.clone(), 0, 0, absolute_max); + let pending_before = transition_state.pending_tasks(); 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"); @@ -8916,29 +9007,27 @@ mod tests { 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"); + .expect("recovery should replay a task journal marker with no worker result"); - assert_eq!(outcome, ManualTransitionJobRecoveryOutcome::Unknown); + assert_eq!(outcome, ManualTransitionJobRecoveryOutcome::Resumed); 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); + .expect("replayed task journal job should load"); + assert_eq!(recovered.state, ManualTransitionJobState::Running); 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) + load_manual_transition_scope_admission(ecstore.clone(), &record.scope_key).await, + Ok(admission) if admission.job_id == job_id ), - "unknown task journal recovery must release the scope admission" + "replayed task journal recovery must keep the scope admission until worker results arrive" ); + if transition_state.pending_tasks() > pending_before { + let _ = transition_state.transition_rx.try_recv(); + } + TransitionState::resize_workers_to(ecstore, original_workers, original_workers, absolute_max); } #[tokio::test] diff --git a/crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs b/crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs index 006ddd7c2..d15d79e67 100644 --- a/crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs +++ b/crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::collections::BTreeMap; +use std::collections::{BTreeMap, BTreeSet}; use std::sync::Arc; use rustfs_utils::crypto::{hex_sha256, is_sha256_checksum}; @@ -505,6 +505,14 @@ pub struct ManualTransitionTaskRecord { pub object: String, pub version_id: Option, pub storage_class: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub etag: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub mod_time_unix_nanos: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub size: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub is_latest: Option, pub queued_at_unix_nanos: i128, } @@ -525,10 +533,28 @@ impl ManualTransitionTaskRecord { object: object.into(), version_id, storage_class: storage_class.into(), + etag: None, + mod_time_unix_nanos: None, + size: None, + is_latest: None, queued_at_unix_nanos: OffsetDateTime::now_utc().unix_timestamp_nanos(), } } + pub fn with_object_metadata( + mut self, + etag: Option, + mod_time: Option, + size: i64, + is_latest: bool, + ) -> Self { + self.etag = etag; + self.mod_time_unix_nanos = mod_time.map(|time| time.unix_timestamp_nanos()); + self.size = Some(size); + self.is_latest = Some(is_latest); + self + } + pub fn encode(&self) -> Result, ManualTransitionJobError> { self.validate()?; let record_bytes = serde_json::to_vec(self)?; @@ -588,6 +614,9 @@ impl ManualTransitionTaskRecord { if self.storage_class.trim().is_empty() { return Err(ManualTransitionJobError::Corrupt("task record storage_class is empty")); } + if self.size.is_some_and(|size| size < 0) { + return Err(ManualTransitionJobError::Corrupt("task record size is negative")); + } Ok(()) } } @@ -1151,23 +1180,95 @@ pub async fn load_manual_transition_worker_result_stats( job_id: Uuid, ) -> EcstoreResult { match scan_manual_transition_worker_result_journal(api, job_id).await? { - ManualTransitionWorkerResultJournal::Stats(stats) => Ok(stats), + ManualTransitionWorkerResultJournal::Stats(stats) => Ok(stats.stats), ManualTransitionWorkerResultJournal::Corrupt(error) => Err(Error::other(error)), } } enum ManualTransitionWorkerResultJournal { - Stats(ManualTransitionWorkerResultStats), + Stats(ManualTransitionWorkerResultJournalStats), Corrupt(String), } +#[derive(Debug, Clone, Default, PartialEq, Eq)] +struct ManualTransitionWorkerResultJournalStats { + stats: ManualTransitionWorkerResultStats, + task_keys: BTreeSet, +} + +impl ManualTransitionWorkerResultJournalStats { + fn record(&mut self, result: ManualTransitionWorkerResultRecord) { + self.task_keys.insert(result.task_key.clone()); + self.stats.record(result.result, result.failure_reason); + } +} + +pub async fn load_manual_transition_pending_task_records( + api: Arc, + job_id: Uuid, + limit: usize, +) -> EcstoreResult> { + if limit == 0 { + return Ok(Vec::new()); + } + + let result_keys = match scan_manual_transition_worker_result_journal(api.clone(), job_id).await? { + ManualTransitionWorkerResultJournal::Stats(stats) => stats.task_keys, + ManualTransitionWorkerResultJournal::Corrupt(error) => return Err(Error::other(error)), + }; + let prefix = manual_transition_task_object_prefix(job_id).map_err(manual_transition_job_store_error)?; + let mut marker = None; + let scan_limit = usize::try_from(MANUAL_TRANSITION_TASK_SCAN_LIMIT) + .map_err(|_| Error::other("manual transition task scan limit is invalid"))?; + let mut pending = Vec::with_capacity(limit.min(scan_limit)); + + 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 = + manual_transition_task_key_from_object_name(job_id, &object.name).map_err(manual_transition_job_store_error)?; + if result_keys.contains(&task_key) { + continue; + } + let object_name = manual_transition_task_object_name(job_id, &task_key).map_err(manual_transition_job_store_error)?; + let data = config_boundary::read_config(api.clone(), &object_name).await?; + let task = ManualTransitionTaskRecord::decode(job_id, &task_key, &data).map_err(manual_transition_job_store_error)?; + pending.push(task); + if pending.len() == limit { + return Ok(pending); + } + } + + if !page.is_truncated { + return Ok(pending); + } + 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); + } +} + async fn scan_manual_transition_worker_result_journal( api: Arc, job_id: Uuid, ) -> EcstoreResult { let prefix = manual_transition_worker_result_object_prefix(job_id).map_err(manual_transition_job_store_error)?; let mut marker = None; - let mut stats = ManualTransitionWorkerResultStats::default(); + let mut stats = ManualTransitionWorkerResultJournalStats::default(); loop { let page = api .clone() @@ -1197,7 +1298,7 @@ async fn scan_manual_transition_worker_result_journal( Ok(result) => result, Err(err) => return Ok(ManualTransitionWorkerResultJournal::Corrupt(err.to_string())), }; - stats.record(result.result, result.failure_reason); + stats.record(result); } if !page.is_truncated { return Ok(ManualTransitionWorkerResultJournal::Stats(stats)); @@ -1229,9 +1330,9 @@ 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, - &stats.tier_failure_by_reason, + stats.stats.completed, + stats.stats.failed, + &stats.stats.tier_failure_by_reason, task_stats.queued, queue_snapshot, );