gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-26 08:18:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(ilm): normalize cancelled manual job reports (#5255)

Browse Source 
Ensure cancelled durable manual transition jobs expose a cancelled report both when worker results drain after a cancel request and when legacy persisted records are decoded.

Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
2026-07-26 11:18:53 +08:00
committed by GitHub
parent 6fa2d06731
commit 92f72c3912
1 changed files with 53 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs
+53 -5
View File
@@ -249,7 +249,10 @@ impl ManualTransitionJobRecord {
if !self.scan_completed || self.report.worker_transition_pending() {
return;
}
self.state = if self.cancel_requested || self.report.cancelled {
if self.cancel_requested {
self.report.cancelled = true;
}
self.state = if self.report.cancelled {
ManualTransitionJobState::Cancelled
} else if self.report.was_truncated()
|| self.report.has_partial_enqueue()
@@ -290,16 +293,20 @@ impl ManualTransitionJobRecord {
if !is_sha256_checksum(&persisted.content_sha256) {
return Err(ManualTransitionJobError::Corrupt("content checksum is not a sha256 checksum"));
}
let job_bytes = serde_json::to_vec(&persisted.job)?;
let mut job = persisted.job;
let job_bytes = serde_json::to_vec(&job)?;
let actual_checksum = hex_sha256(&job_bytes, ToOwned::to_owned);
if persisted.content_sha256 != actual_checksum {
return Err(ManualTransitionJobError::ChecksumMismatch);
}
if persisted.job.job_id != expected_job_id {
if job.job_id != expected_job_id {
return Err(ManualTransitionJobError::Corrupt("job_id does not match record key"));
}
persisted.job.validate()?;
Ok(persisted.job)
if job.state == ManualTransitionJobState::Cancelled && job.cancel_requested {
job.report.cancelled = true;
}
job.validate()?;
Ok(job)
}
fn validate(&self) -> Result<(), ManualTransitionJobError> {
@@ -931,6 +938,47 @@ mod tests {
assert!(record.completed_at_unix_nanos.is_some());
}
#[test]
fn manual_transition_job_record_marks_report_cancelled_after_worker_drain() {
let options = ManualTransitionRunOptions::default();
let mut record = ManualTransitionJobRecord::new(Uuid::new_v4(), "bucket", &options, TEST_OWNER);
record.complete(
ManualTransitionRunReport {
bucket: "bucket".to_string(),
enqueued: 1,
..Default::default()
},
ManualTransitionQueueSnapshot::default(),
);
record.mark_cancel_requested();
record.record_worker_result(ManualTransitionWorkerResult::TierFailure, ManualTransitionQueueSnapshot::default());
assert_eq!(record.state, ManualTransitionJobState::Cancelled);
assert!(record.cancel_requested);
assert!(record.report.cancelled);
assert_eq!(record.report.transition_failed, 1);
assert_eq!(record.report.tier_failure, 1);
}
#[test]
fn manual_transition_job_record_decode_normalizes_legacy_cancelled_report() {
let options = ManualTransitionRunOptions::default();
let mut record = ManualTransitionJobRecord::new(Uuid::new_v4(), "bucket", &options, TEST_OWNER);
record.mark_cancel_requested();
record.state = ManualTransitionJobState::Cancelled;
record.completed_at_unix_nanos = Some(OffsetDateTime::now_utc().unix_timestamp_nanos());
record.report.cancelled = false;
let encoded = record.encode().expect("legacy-shaped cancelled job should encode");
let decoded =
ManualTransitionJobRecord::decode(record.job_id, &encoded).expect("legacy-shaped cancelled job should decode");
assert_eq!(decoded.state, ManualTransitionJobState::Cancelled);
assert!(decoded.cancel_requested);
assert!(decoded.report.cancelled);
}
#[test]
fn manual_transition_job_scan_progress_preserves_worker_counters() {
let options = ManualTransitionRunOptions::default();