test(ilm): cover transition budget partial records (#5305)

Co-authored-by: heihutu <heihutu@gmail.com>

crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs

@@ -2076,6 +2076,60 @@ mod tests {
         assert!(record.error.is_none());
     }
 
+    #[test]
+    fn manual_transition_job_record_budget_reports_are_partial_and_resumable() {
+        let options = ManualTransitionRunOptions {
+            prefix: "logs/".to_string(),
+            tier: Some("warm".to_string()),
+            max_objects: Some(1),
+            ..Default::default()
+        };
+
+        for (bucket, truncated_by_limit, truncated_by_duration) in [
+            ("manual-budget-limit-bucket", true, false),
+            ("manual-budget-duration-bucket", false, true),
+        ] {
+            let mut record = ManualTransitionJobRecord::new(Uuid::new_v4(), bucket, &options, TEST_OWNER);
+            let queue_snapshot = ManualTransitionQueueSnapshot {
+                queue_capacity: 8,
+                queued: 1,
+                active: 1,
+                workers: 2,
+                ..Default::default()
+            };
+            record.complete(
+                ManualTransitionRunReport {
+                    bucket: bucket.to_string(),
+                    prefix: "logs/".to_string(),
+                    tier: Some("warm".to_string()),
+                    scanned: 1,
+                    eligible: 1,
+                    dry_run_eligible: 1,
+                    truncated_by_limit,
+                    truncated_by_duration,
+                    continuation_token: Some("opaque-budget-token".to_string()),
+                    ..Default::default()
+                },
+                queue_snapshot,
+            );
+
+            assert_eq!(record.state, ManualTransitionJobState::Partial);
+            assert_eq!(record.report.continuation_token.as_deref(), Some("opaque-budget-token"));
+            assert!(record.report.was_truncated());
+            assert_eq!(record.queue_snapshot, queue_snapshot);
+            assert!(record.completed_at_unix_nanos.is_some());
+            assert!(record.error.is_none());
+
+            let encoded = record.encode().expect("budget partial job should encode");
+            let decoded = ManualTransitionJobRecord::decode(record.job_id, &encoded).expect("budget partial job should decode");
+            assert_eq!(decoded.state, ManualTransitionJobState::Partial);
+            assert_eq!(decoded.report.truncated_by_limit, truncated_by_limit);
+            assert_eq!(decoded.report.truncated_by_duration, truncated_by_duration);
+            assert_eq!(decoded.report.continuation_token.as_deref(), Some("opaque-budget-token"));
+            assert_eq!(decoded.queue_snapshot, queue_snapshot);
+        }
+    }
+
     #[test]
     fn manual_transition_scope_key_is_stable_and_sanitized() {
         let first = manual_transition_scope_key(