From 12fcac61d2a816a70a8b6b2123691b3ee1b2a3d8 Mon Sep 17 00:00:00 2001 From: houseme Date: Sun, 26 Jul 2026 12:29:18 +0800 Subject: [PATCH] fix(ilm): deduplicate manual transition worker results Persist per-task manual transition worker result markers before applying job counters so duplicate worker completion reports are no-ops. Reconcile persisted markers during drained-queue lease renewal and recovery to restore marker-before-record crash windows. Fail closed when persisted worker result markers are corrupt. Co-Authored-By: heihutu --- .../bucket/lifecycle/bucket_lifecycle_ops.rs | 348 ++++++++++++++- .../bucket/lifecycle/manual_transition_job.rs | 420 +++++++++++++++++- 2 files changed, 756 insertions(+), 12 deletions(-) diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index dba337333..6c711c5e6 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -22,12 +22,13 @@ use crate::bucket::lifecycle::lifecycle::{ self, Lifecycle, ObjectOpts, TransitionOptions, abort_incomplete_multipart_upload_due, }; use crate::bucket::lifecycle::manual_transition_job::{ - MANUAL_TRANSITION_JOB_RECORD_PREFIX, ManualTransitionJobRecord, ManualTransitionScopeAdmission, + 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, persist_manual_transition_job_progress, record_manual_transition_worker_result, - renew_manual_transition_job_lease, save_manual_transition_job_record_if_current, + 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, }; use crate::bucket::lifecycle::replication_sink; use crate::bucket::lifecycle::replication_sink::{ @@ -1097,6 +1098,7 @@ struct TransitionTask { src: LcEventSrc, event: lifecycle::Event, manual_job_id: Option, + manual_result_key: Option, } impl ExpiryOp for TransitionTask { @@ -1425,6 +1427,8 @@ impl TransitionState { 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)), }; if is_immediate_transition_source(src) { let outcome = match self.transition_tx.try_send(Some(task)) { @@ -1603,10 +1607,11 @@ impl TransitionState { 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 { + if let (Some(job_id), Some(result_key)) = (task.manual_job_id, task.manual_result_key.as_deref()) { record_manual_transition_worker_result_for_task( api.clone(), job_id, + result_key, ManualTransitionWorkerResult::TierFailure, ) .await; @@ -1628,10 +1633,11 @@ impl TransitionState { } emit_transition_failed_event(obj_info_for_event); } else { - if let Some(job_id) = task.manual_job_id { + if let (Some(job_id), Some(result_key)) = (task.manual_job_id, task.manual_result_key.as_deref()) { record_manual_transition_worker_result_for_task( api.clone(), job_id, + result_key, ManualTransitionWorkerResult::Completed, ) .await; @@ -1741,8 +1747,15 @@ 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 { +async fn record_manual_transition_worker_result_for_task( + api: Arc, + job_id: Uuid, + result_key: &str, + result: ManualTransitionWorkerResult, +) { + if let Err(err) = + record_manual_transition_worker_result(api, job_id, result_key, result, manual_transition_queue_snapshot()).await + { warn!( event = EVENT_LIFECYCLE_WORKER_STATE, component = LOG_COMPONENT_ECSTORE, @@ -1940,7 +1953,7 @@ async fn recover_manual_transition_job( job_id: Uuid, queue_snapshot: ManualTransitionQueueSnapshot, ) -> Result { - let (mut record, etag) = match load_manual_transition_job_record_with_etag(api.clone(), job_id).await { + let (mut record, mut etag) = match load_manual_transition_job_record_with_etag(api.clone(), job_id).await { Ok(record) => record, Err(Error::ConfigNotFound) => return Ok(ManualTransitionJobRecoveryOutcome::Skipped), Err(err) => return Err(err), @@ -1950,6 +1963,21 @@ async fn recover_manual_transition_job( } let recovery_unknown_snapshot = ManualTransitionQueueSnapshot::default(); + if record.scan_completed && record.report.worker_transition_pending() { + 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; + return match reconciled.state { + ManualTransitionJobState::Cancelled => Ok(ManualTransitionJobRecoveryOutcome::Cancelled), + ManualTransitionJobState::Unknown => Ok(ManualTransitionJobRecoveryOutcome::Unknown), + _ => Ok(ManualTransitionJobRecoveryOutcome::Resumed), + }; + } + if reconciled != record { + (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) { @@ -4594,12 +4622,15 @@ mod tests { use crate::bucket::lifecycle::config_boundary; use crate::bucket::lifecycle::manual_transition_job::{ ManualTransitionJobRecord, ManualTransitionJobState, ManualTransitionScopeAdmission, ManualTransitionScopeAdmissionClaim, - ManualTransitionWorkerResult, claim_manual_transition_scope_admission, + 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, - renew_manual_transition_job_lease, request_manual_transition_job_cancel, save_manual_transition_job_record, + 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_worker_result_if_absent, }; use crate::bucket::lifecycle::replication_sink::{ ReplicateDecision, ReplicateTargetDecision, ReplicationStatusType, VersionPurgeStatusType, @@ -8002,6 +8033,303 @@ mod tests { ); } + #[tokio::test] + #[serial] + async fn manual_transition_worker_result_duplicate_marker_is_noop() { + let (_paths, ecstore) = setup_test_env().await; + let job_id = Uuid::new_v4(); + let bucket = format!("manual-worker-result-{}", job_id.simple()); + let options = ManualTransitionRunOptions::default(); + let mut record = ManualTransitionJobRecord::new(job_id, &bucket, &options, "worker-owner"); + record.complete( + ManualTransitionRunReport { + bucket: bucket.clone(), + enqueued: 2, + ..Default::default() + }, + ManualTransitionQueueSnapshot::default(), + ); + save_manual_transition_job_record(ecstore.clone(), &record) + .await + .expect("worker result job record should save"); + save_manual_transition_scope_admission_if_absent(ecstore.clone(), &ManualTransitionScopeAdmission::from_job(&record)) + .await + .expect("worker result admission should save"); + + let first_key = manual_transition_worker_result_task_key(&bucket, "logs/a", None); + let first = record_manual_transition_worker_result( + ecstore.clone(), + job_id, + &first_key, + ManualTransitionWorkerResult::Completed, + ManualTransitionQueueSnapshot::default(), + ) + .await + .expect("first worker result should persist"); + assert_eq!(first.state, ManualTransitionJobState::Running); + assert_eq!(first.report.transition_completed, 1); + assert_eq!(first.report.transition_failed, 0); + + let duplicate = record_manual_transition_worker_result( + ecstore.clone(), + job_id, + &first_key, + ManualTransitionWorkerResult::Completed, + ManualTransitionQueueSnapshot::default(), + ) + .await + .expect("duplicate worker result should be idempotent"); + assert_eq!(duplicate.state, ManualTransitionJobState::Running); + assert_eq!(duplicate.report.transition_completed, 1); + assert_eq!(duplicate.report.transition_failed, 0); + + let second_key = manual_transition_worker_result_task_key(&bucket, "logs/b", None); + let final_record = record_manual_transition_worker_result( + ecstore.clone(), + job_id, + &second_key, + ManualTransitionWorkerResult::TierFailure, + ManualTransitionQueueSnapshot::default(), + ) + .await + .expect("second distinct worker result should persist"); + + assert_eq!(final_record.state, ManualTransitionJobState::Partial); + assert_eq!(final_record.report.transition_completed, 1); + assert_eq!(final_record.report.transition_failed, 1); + assert_eq!(final_record.report.tier_failure, 1); + assert!( + matches!( + load_manual_transition_scope_admission(ecstore, &record.scope_key).await, + Err(Error::ConfigNotFound) + ), + "terminal worker result must release the scope admission" + ); + } + + #[tokio::test] + #[serial] + async fn manual_transition_worker_result_reconcile_applies_marker_and_releases_admission() { + let (_paths, ecstore) = setup_test_env().await; + let job_id = Uuid::new_v4(); + let bucket = format!("manual-worker-reconcile-{}", job_id.simple()); + let options = ManualTransitionRunOptions::default(); + let mut record = ManualTransitionJobRecord::new(job_id, &bucket, &options, "worker-owner"); + record.complete( + ManualTransitionRunReport { + bucket: bucket.clone(), + enqueued: 1, + ..Default::default() + }, + ManualTransitionQueueSnapshot::default(), + ); + save_manual_transition_job_record(ecstore.clone(), &record) + .await + .expect("worker reconcile job record should save"); + save_manual_transition_scope_admission_if_absent(ecstore.clone(), &ManualTransitionScopeAdmission::from_job(&record)) + .await + .expect("worker reconcile admission should save"); + + let task_key = manual_transition_worker_result_task_key(&bucket, "logs/a", None); + let marker = ManualTransitionWorkerResultRecord::new(job_id, &task_key, ManualTransitionWorkerResult::Completed); + assert!( + save_manual_transition_worker_result_if_absent(ecstore.clone(), &marker) + .await + .expect("worker result marker should save"), + "new worker result marker must be created once" + ); + assert!( + !save_manual_transition_worker_result_if_absent(ecstore.clone(), &marker) + .await + .expect("duplicate worker result marker should not fail"), + "duplicate worker result marker must be reported as existing" + ); + let duplicate_noop = record_manual_transition_worker_result( + ecstore.clone(), + job_id, + &task_key, + ManualTransitionWorkerResult::Completed, + ManualTransitionQueueSnapshot::default(), + ) + .await + .expect("duplicate worker result should not apply journal counts"); + assert_eq!(duplicate_noop.state, ManualTransitionJobState::Running); + assert_eq!(duplicate_noop.report.transition_completed, 0); + assert_eq!(duplicate_noop.report.transition_failed, 0); + + let reconciled = + reconcile_manual_transition_worker_results(ecstore.clone(), job_id, ManualTransitionQueueSnapshot::default()) + .await + .expect("worker result marker should reconcile into job record"); + + assert_eq!(reconciled.state, ManualTransitionJobState::Completed); + assert_eq!(reconciled.report.transition_completed, 1); + assert_eq!(reconciled.report.transition_failed, 0); + assert_eq!(reconciled.report.tier_failure, 0); + assert!( + matches!( + load_manual_transition_scope_admission(ecstore, &record.scope_key).await, + Err(Error::ConfigNotFound) + ), + "terminal reconcile must release the scope admission" + ); + } + + #[tokio::test] + #[serial] + async fn manual_transition_worker_result_heartbeat_applies_marker_before_record_update() { + let (_paths, ecstore) = setup_test_env().await; + let job_id = Uuid::new_v4(); + let bucket = format!("manual-worker-heartbeat-{}", job_id.simple()); + let mut record = ManualTransitionJobRecord::new(job_id, &bucket, &ManualTransitionRunOptions::default(), "worker-owner"); + record.complete( + ManualTransitionRunReport { + bucket: bucket.clone(), + enqueued: 1, + ..Default::default() + }, + ManualTransitionQueueSnapshot::default(), + ); + save_manual_transition_job_record(ecstore.clone(), &record) + .await + .expect("worker heartbeat job record should save"); + save_manual_transition_scope_admission_if_absent(ecstore.clone(), &ManualTransitionScopeAdmission::from_job(&record)) + .await + .expect("worker heartbeat admission should save"); + let task_key = manual_transition_worker_result_task_key(&bucket, "logs/a", None); + let marker = ManualTransitionWorkerResultRecord::new(job_id, &task_key, ManualTransitionWorkerResult::Completed); + assert!( + save_manual_transition_worker_result_if_absent(ecstore.clone(), &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 marker before unknown fallback"); + + assert_eq!(renewed.state, ManualTransitionJobState::Completed); + assert_eq!(renewed.report.transition_completed, 1); + assert!( + matches!( + load_manual_transition_scope_admission(ecstore, &record.scope_key).await, + Err(Error::ConfigNotFound) + ), + "terminal heartbeat reconcile must release the scope admission" + ); + } + + #[tokio::test] + #[serial] + async fn manual_transition_worker_result_recovery_applies_marker_before_record_update() { + let (_paths, ecstore) = setup_test_env().await; + let job_id = Uuid::new_v4(); + let bucket = format!("manual-worker-recovery-{}", job_id.simple()); + let mut record = ManualTransitionJobRecord::new(job_id, &bucket, &ManualTransitionRunOptions::default(), "old-owner"); + record.complete( + ManualTransitionRunReport { + bucket: bucket.clone(), + enqueued: 1, + ..Default::default() + }, + ManualTransitionQueueSnapshot::default(), + ); + record.lease_expires_at_unix_nanos = 0; + save_manual_transition_job_record(ecstore.clone(), &record) + .await + .expect("worker recovery job record should save"); + save_manual_transition_scope_admission_if_absent(ecstore.clone(), &ManualTransitionScopeAdmission::from_job(&record)) + .await + .expect("worker recovery admission should save"); + let task_key = manual_transition_worker_result_task_key(&bucket, "logs/a", None); + let marker = ManualTransitionWorkerResultRecord::new(job_id, &task_key, ManualTransitionWorkerResult::Completed); + assert!( + save_manual_transition_worker_result_if_absent(ecstore.clone(), &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 marker before unknown fallback"); + + assert_eq!(outcome, ManualTransitionJobRecoveryOutcome::Resumed); + let recovered = load_manual_transition_job_record(ecstore.clone(), job_id) + .await + .expect("reconciled job should load"); + assert_eq!(recovered.state, ManualTransitionJobState::Completed); + assert_eq!(recovered.report.transition_completed, 1); + assert!( + matches!( + load_manual_transition_scope_admission(ecstore, &record.scope_key).await, + Err(Error::ConfigNotFound) + ), + "terminal recovery reconcile must release the scope admission" + ); + } + + #[tokio::test] + #[serial] + async fn manual_transition_worker_result_recovery_marks_unknown_for_corrupt_marker() { + let (_paths, ecstore) = setup_test_env().await; + let job_id = Uuid::new_v4(); + let bucket = format!("manual-worker-corrupt-{}", job_id.simple()); + let mut record = ManualTransitionJobRecord::new(job_id, &bucket, &ManualTransitionRunOptions::default(), "old-owner"); + record.complete( + ManualTransitionRunReport { + bucket: bucket.clone(), + enqueued: 1, + ..Default::default() + }, + ManualTransitionQueueSnapshot::default(), + ); + record.lease_expires_at_unix_nanos = 0; + save_manual_transition_job_record(ecstore.clone(), &record) + .await + .expect("corrupt marker job record should save"); + save_manual_transition_scope_admission_if_absent(ecstore.clone(), &ManualTransitionScopeAdmission::from_job(&record)) + .await + .expect("corrupt marker admission should save"); + let task_key = manual_transition_worker_result_task_key(&bucket, "logs/a", None); + let marker = ManualTransitionWorkerResultRecord::new(job_id, &task_key, ManualTransitionWorkerResult::Completed); + let encoded = marker.encode().expect("worker result marker should encode"); + let mut value: serde_json::Value = serde_json::from_slice(&encoded).expect("worker result marker should be json"); + value["record"]["result"] = serde_json::Value::String("tier_failure".to_string()); + let object = manual_transition_worker_result_object_name(job_id, &task_key).expect("worker result object should encode"); + config_boundary::save_config( + ecstore.clone(), + &object, + serde_json::to_vec(&value).expect("corrupt marker should encode"), + ) + .await + .expect("corrupt worker result marker should save"); + + let outcome = recover_manual_transition_job(ecstore.clone(), job_id, ManualTransitionQueueSnapshot::default()) + .await + .expect("recovery should fail closed on corrupt marker"); + + assert_eq!(outcome, ManualTransitionJobRecoveryOutcome::Unknown); + let recovered = load_manual_transition_job_record(ecstore.clone(), job_id) + .await + .expect("unknown job should load"); + assert_eq!(recovered.state, ManualTransitionJobState::Unknown); + assert!( + recovered + .error + .as_deref() + .is_some_and(|error| error.contains("worker result journal is corrupt")) + ); + assert!( + matches!( + load_manual_transition_scope_admission(ecstore, &record.scope_key).await, + Err(Error::ConfigNotFound) + ), + "corrupt marker unknown 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 2e5487087..57879b3ef 100644 --- a/crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs +++ b/crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs @@ -31,11 +31,15 @@ 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_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_WORKER_RESULT_PREFIX: &str = "ilm/manual-transition/results"; pub const MAX_MANUAL_TRANSITION_JOB_RECORD_SIZE: usize = 64 * 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_WORKER_RESULT_SCAN_LIMIT: i32 = 1000; fn is_false(value: &bool) -> bool { !*value @@ -141,6 +145,21 @@ impl ManualTransitionJobRecord { self.mark_updated_terminal(); } + pub fn mark_unknown_for_worker_result_journal_error( + &mut self, + error: impl Into, + queue_snapshot: ManualTransitionQueueSnapshot, + ) -> bool { + if self.state != ManualTransitionJobState::Running || !self.scan_completed || !self.report.worker_transition_pending() { + return false; + } + self.queue_snapshot = queue_snapshot; + self.state = ManualTransitionJobState::Unknown; + self.error = Some(format!("manual transition worker result 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; @@ -170,6 +189,32 @@ impl ManualTransitionJobRecord { self.mark_terminal_if_worker_drained(); } + fn apply_worker_result_counts(&mut self, completed: u64, failed: u64, queue_snapshot: ManualTransitionQueueSnapshot) -> bool { + if self.is_terminal() { + return false; + } + if completed.saturating_add(failed) > self.report.enqueued { + self.queue_snapshot = queue_snapshot; + self.state = ManualTransitionJobState::Unknown; + self.error = Some("manual transition worker result journal exceeds enqueued count".to_string()); + self.mark_updated_terminal(); + return true; + } + 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 { + return false; + } + let scan_tier_failure = self.report.tier_failure.saturating_sub(self.report.transition_failed); + self.report.transition_completed = transition_completed; + self.report.transition_failed = transition_failed; + self.report.tier_failure = scan_tier_failure.saturating_add(transition_failed); + self.queue_snapshot = queue_snapshot; + self.updated_at_unix_nanos = OffsetDateTime::now_utc().unix_timestamp_nanos(); + self.mark_terminal_if_worker_drained(); + true + } + pub fn mark_cancel_requested(&mut self) { self.cancel_requested = true; self.updated_at_unix_nanos = OffsetDateTime::now_utc().unix_timestamp_nanos(); @@ -363,12 +408,116 @@ impl ManualTransitionJobRecord { } } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] pub enum ManualTransitionWorkerResult { Completed, TierFailure, } +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct ManualTransitionWorkerResultStats { + pub completed: u64, + pub failed: u64, +} + +impl ManualTransitionWorkerResultStats { + fn record(&mut self, result: ManualTransitionWorkerResult) { + match result { + ManualTransitionWorkerResult::Completed => self.completed = self.completed.saturating_add(1), + ManualTransitionWorkerResult::TierFailure => self.failed = self.failed.saturating_add(1), + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ManualTransitionWorkerResultRecord { + pub schema: String, + pub job_id: Uuid, + pub task_key: String, + pub result: ManualTransitionWorkerResult, + pub completed_at_unix_nanos: i128, +} + +impl ManualTransitionWorkerResultRecord { + pub fn new(job_id: Uuid, task_key: impl Into, result: ManualTransitionWorkerResult) -> Self { + Self { + schema: MANUAL_TRANSITION_WORKER_RESULT_SCHEMA.to_string(), + job_id, + task_key: task_key.into(), + result, + completed_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 = PersistedManualTransitionWorkerResultRecord { + schema: MANUAL_TRANSITION_WORKER_RESULT_SCHEMA.to_string(), + content_sha256, + record: self.clone(), + }; + let encoded = serde_json::to_vec(&persisted)?; + if encoded.len() > MAX_MANUAL_TRANSITION_WORKER_RESULT_RECORD_SIZE { + return Err(ManualTransitionJobError::Corrupt("encoded worker result 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_WORKER_RESULT_RECORD_SIZE { + return Err(ManualTransitionJobError::Corrupt("encoded worker result exceeds maximum size")); + } + let persisted: PersistedManualTransitionWorkerResultRecord = serde_json::from_slice(data)?; + if persisted.schema != MANUAL_TRANSITION_WORKER_RESULT_SCHEMA { + return Err(ManualTransitionJobError::UnsupportedSchema(persisted.schema)); + } + if !is_sha256_checksum(&persisted.content_sha256) { + return Err(ManualTransitionJobError::Corrupt( + "worker result 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("worker result job_id does not match record key")); + } + if record.task_key != expected_task_key { + return Err(ManualTransitionJobError::Corrupt("worker result task_key does not match record key")); + } + record.validate()?; + Ok(record) + } + + fn validate(&self) -> Result<(), ManualTransitionJobError> { + if self.schema != MANUAL_TRANSITION_WORKER_RESULT_SCHEMA { + return Err(ManualTransitionJobError::UnsupportedSchema(self.schema.clone())); + } + if self.job_id.is_nil() { + return Err(ManualTransitionJobError::Corrupt("worker result job_id is nil")); + } + if !is_sha256_checksum(&self.task_key) { + return Err(ManualTransitionJobError::Corrupt("worker result task_key is not a sha256 checksum")); + } + Ok(()) + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct PersistedManualTransitionWorkerResultRecord { + schema: String, + content_sha256: String, + record: ManualTransitionWorkerResultRecord, +} + 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)) } @@ -485,6 +634,59 @@ pub fn manual_transition_job_record_object_name(job_id: Uuid) -> Result Result { + if job_id.is_nil() { + return Err(ManualTransitionJobError::Corrupt("job_id is nil")); + } + let job_key = job_id.simple().to_string(); + Ok(format!("{}/{}/{}/{}", prefix, &job_key[..2], &job_key[2..4], job_key)) +} + +pub fn manual_transition_worker_result_task_key(bucket: &str, object: &str, version_id: Option) -> String { + let version = version_id.map(|version| version.to_string()).unwrap_or_default(); + let mut material = Vec::with_capacity(bucket.len() + object.len() + version.len() + 32); + push_len_prefixed(&mut material, bucket.as_bytes()); + push_len_prefixed(&mut material, object.as_bytes()); + push_len_prefixed(&mut material, version.as_bytes()); + hex_sha256(&material, ToOwned::to_owned) +} + +fn push_len_prefixed(out: &mut Vec, value: &[u8]) { + out.extend_from_slice(value.len().to_string().as_bytes()); + out.push(b':'); + out.extend_from_slice(value); +} + +pub fn manual_transition_worker_result_object_prefix(job_id: Uuid) -> Result { + manual_transition_job_sharded_prefix(MANUAL_TRANSITION_WORKER_RESULT_PREFIX, job_id) +} + +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")); + } + Ok(format!("{}/{}.json", manual_transition_worker_result_object_prefix(job_id)?, task_key)) +} + +fn manual_transition_worker_result_task_key_from_object_name( + job_id: Uuid, + object_name: &str, +) -> Result { + let prefix = manual_transition_worker_result_object_prefix(job_id)?; + let rest = object_name + .strip_prefix(&prefix) + .ok_or(ManualTransitionJobError::Corrupt("worker result object prefix is invalid"))? + .strip_prefix('/') + .ok_or(ManualTransitionJobError::Corrupt("worker result object path is invalid"))?; + let task_key = rest + .strip_suffix(".json") + .ok_or(ManualTransitionJobError::Corrupt("worker result suffix is invalid"))?; + if task_key.contains('/') || !is_sha256_checksum(task_key) { + return Err(ManualTransitionJobError::Corrupt("worker result 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")); @@ -582,6 +784,160 @@ pub async fn save_manual_transition_job_record_if_current( .await } +pub(crate) async fn save_manual_transition_worker_result_if_absent( + api: Arc, + record: &ManualTransitionWorkerResultRecord, +) -> EcstoreResult { + let object = manual_transition_worker_result_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_worker_result_stats( + api: Arc, + job_id: Uuid, +) -> EcstoreResult { + match scan_manual_transition_worker_result_journal(api, job_id).await? { + ManualTransitionWorkerResultJournal::Stats(stats) => Ok(stats), + ManualTransitionWorkerResultJournal::Corrupt(error) => Err(Error::other(error)), + } +} + +enum ManualTransitionWorkerResultJournal { + Stats(ManualTransitionWorkerResultStats), + Corrupt(String), +} + +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(); + loop { + let page = api + .clone() + .list_objects_v2( + RUSTFS_META_BUCKET, + &prefix, + marker, + None, + MANUAL_TRANSITION_WORKER_RESULT_SCAN_LIMIT, + false, + None, + false, + ) + .await?; + for object in page.objects { + let task_key = match manual_transition_worker_result_task_key_from_object_name(job_id, &object.name) { + Ok(task_key) => task_key, + Err(err) => return Ok(ManualTransitionWorkerResultJournal::Corrupt(err.to_string())), + }; + let object_name = + manual_transition_worker_result_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), + }; + let result = match ManualTransitionWorkerResultRecord::decode(job_id, &task_key, &data) { + Ok(result) => result, + Err(err) => return Ok(ManualTransitionWorkerResultJournal::Corrupt(err.to_string())), + }; + stats.record(result.result); + } + if !page.is_truncated { + return Ok(ManualTransitionWorkerResultJournal::Stats(stats)); + } + let Some(next_marker) = page.next_continuation_token else { + return Err(Error::other("manual transition worker result page is truncated without a next marker")); + }; + marker = Some(next_marker); + } +} + +pub async fn reconcile_manual_transition_worker_results( + api: Arc, + job_id: Uuid, + queue_snapshot: ManualTransitionQueueSnapshot, +) -> EcstoreResult { + let stats = match scan_manual_transition_worker_result_journal(api.clone(), job_id).await? { + ManualTransitionWorkerResultJournal::Stats(stats) => stats, + ManualTransitionWorkerResultJournal::Corrupt(error) => { + return mark_manual_transition_job_unknown_for_worker_result_journal_error(api, job_id, error, queue_snapshot).await; + } + }; + 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); + if !changed { + return Ok(record); + } + 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) +} + +async fn mark_manual_transition_job_unknown_for_worker_result_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_worker_result_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) +} + pub async fn save_manual_transition_scope_admission_if_absent( api: Arc, admission: &ManualTransitionScopeAdmission, @@ -793,9 +1149,15 @@ pub async fn persist_manual_transition_job_progress( pub async fn record_manual_transition_worker_result( api: Arc, job_id: Uuid, + task_key: &str, result: ManualTransitionWorkerResult, queue_snapshot: ManualTransitionQueueSnapshot, ) -> EcstoreResult { + let result_record = ManualTransitionWorkerResultRecord::new(job_id, task_key, result); + if !save_manual_transition_worker_result_if_absent(api.clone(), &result_record).await? { + return load_manual_transition_job_record(api, job_id).await; + } + for _ in 0..4 { let (mut record, etag) = load_manual_transition_job_record_with_etag(api.clone(), job_id).await?; if record.is_terminal() { @@ -829,8 +1191,19 @@ pub async fn renew_manual_transition_job_lease( job_id: Uuid, queue_snapshot: ManualTransitionQueueSnapshot, ) -> EcstoreResult { - let (mut record, etag) = load_manual_transition_job_record_with_etag(api.clone(), job_id).await?; + 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 + { + 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); + } + (record, etag) = load_manual_transition_job_record_with_etag(api.clone(), job_id).await?; + } let became_terminal = record.mark_unknown_if_worker_results_lost(queue_snapshot); if !became_terminal { record.renew_lease(queue_snapshot); @@ -1100,6 +1473,33 @@ mod tests { assert!(record.completed_at_unix_nanos.is_some()); } + #[test] + fn manual_transition_job_marks_unknown_for_corrupt_worker_result_journal() { + let options = ManualTransitionRunOptions::default(); + let mut record = ManualTransitionJobRecord::new(Uuid::new_v4(), "bucket", &options, TEST_OWNER); + + assert!(!record.mark_unknown_for_worker_result_journal_error("bad marker", ManualTransitionQueueSnapshot::default())); + + record.complete( + ManualTransitionRunReport { + bucket: "bucket".to_string(), + enqueued: 1, + ..Default::default() + }, + ManualTransitionQueueSnapshot::default(), + ); + assert!(record.mark_unknown_for_worker_result_journal_error("bad marker", ManualTransitionQueueSnapshot::default())); + + assert_eq!(record.state, ManualTransitionJobState::Unknown); + assert!( + record + .error + .as_deref() + .is_some_and(|error| error.contains("worker result journal is corrupt")) + ); + assert!(record.completed_at_unix_nanos.is_some()); + } + #[test] fn manual_transition_job_record_builds_resume_options() { let options = ManualTransitionRunOptions { @@ -1166,6 +1566,22 @@ mod tests { assert!(matches!(err, ManualTransitionJobError::ChecksumMismatch)); } + #[test] + fn manual_transition_worker_result_record_rejects_checksum_drift() { + let job_id = Uuid::new_v4(); + let task_key = manual_transition_worker_result_task_key("bucket", "logs/a", None); + let record = ManualTransitionWorkerResultRecord::new(job_id, &task_key, ManualTransitionWorkerResult::Completed); + let encoded = record.encode().expect("worker result record should encode"); + let mut value: serde_json::Value = serde_json::from_slice(&encoded).expect("encoded worker result should be json"); + value["record"]["result"] = serde_json::Value::String("tier_failure".to_string()); + let mutated = serde_json::to_vec(&value).expect("mutated worker result should encode"); + + let err = ManualTransitionWorkerResultRecord::decode(job_id, &task_key, &mutated) + .expect_err("worker result checksum drift must fail closed"); + + assert!(matches!(err, ManualTransitionJobError::ChecksumMismatch)); + } + #[test] fn manual_transition_job_record_rejects_unknown_report_fields() { let options = ManualTransitionRunOptions::default();