1use std::collections::BTreeSet;
10use std::error::Error;
11use std::fmt;
12use std::time::{Duration, SystemTime};
13
14use oxide_batch_core::{
15 BatchStatus, ExecutionVersion, JobExecutionId, JobInstanceId, JobName, RetentionActionId,
16};
17
18use crate::{ActorRef, CanonicalWriter, OperationId, ReasonCode, RepositoryError, hex_digest};
19
20pub const MAX_PURGE_BATCH: u32 = 1000;
22pub const MIN_PURGE_AGE: Duration = Duration::from_hours(1);
24pub const DEFAULT_PURGE_AGE: Duration = Duration::from_hours(30 * 24);
26
27#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
29pub struct PurgeBatchBound(u32);
30
31impl PurgeBatchBound {
32 pub const fn new(value: u32) -> Result<Self, RetentionError> {
38 if value == 0 || value > MAX_PURGE_BATCH {
39 return Err(RetentionError::BatchBoundOutOfRange { requested: value });
40 }
41 Ok(Self(value))
42 }
43
44 #[must_use]
46 pub const fn get(self) -> u32 {
47 self.0
48 }
49}
50
51impl Default for PurgeBatchBound {
52 fn default() -> Self {
53 Self(MAX_PURGE_BATCH)
54 }
55}
56
57#[derive(Clone, Debug, Eq, PartialEq)]
59pub struct TerminalStatusSet(BTreeSet<BatchStatus>);
60
61impl TerminalStatusSet {
62 pub fn new(statuses: impl IntoIterator<Item = BatchStatus>) -> Result<Self, RetentionError> {
69 let mut set = BTreeSet::new();
70 for status in statuses {
71 if !status.is_finished() {
72 return Err(RetentionError::NonTerminalStatus { status });
73 }
74 set.insert(status);
75 }
76 if set.is_empty() {
77 return Err(RetentionError::EmptyStatusSet);
78 }
79 Ok(Self(set))
80 }
81
82 #[must_use]
84 pub fn all() -> Self {
85 Self(
86 [
87 BatchStatus::Completed,
88 BatchStatus::Failed,
89 BatchStatus::Stopped,
90 BatchStatus::Abandoned,
91 ]
92 .into_iter()
93 .collect(),
94 )
95 }
96
97 #[must_use]
99 pub fn contains(&self, status: BatchStatus) -> bool {
100 self.0.contains(&status)
101 }
102
103 #[must_use]
105 pub fn iter(&self) -> impl ExactSizeIterator<Item = BatchStatus> + '_ {
106 self.0.iter().copied()
107 }
108}
109
110#[derive(Clone, Debug, Eq, PartialEq)]
112pub struct PurgePlanRequest {
113 job_name: JobName,
114 statuses: TerminalStatusSet,
115 minimum_age: Duration,
116 batch: PurgeBatchBound,
117}
118
119impl PurgePlanRequest {
120 pub fn new(
126 job_name: JobName,
127 statuses: TerminalStatusSet,
128 minimum_age: Duration,
129 batch: PurgeBatchBound,
130 ) -> Result<Self, RetentionError> {
131 if minimum_age.as_secs() < MIN_PURGE_AGE.as_secs() {
132 return Err(RetentionError::AgeBoundTooSmall {
133 minimum: MIN_PURGE_AGE,
134 });
135 }
136 Ok(Self {
137 job_name,
138 statuses,
139 minimum_age,
140 batch,
141 })
142 }
143
144 #[must_use]
146 pub const fn job_name(&self) -> &JobName {
147 &self.job_name
148 }
149
150 #[must_use]
152 pub const fn statuses(&self) -> &TerminalStatusSet {
153 &self.statuses
154 }
155
156 #[must_use]
158 pub const fn minimum_age(&self) -> Duration {
159 self.minimum_age
160 }
161
162 #[must_use]
164 pub const fn batch(&self) -> PurgeBatchBound {
165 self.batch
166 }
167}
168
169#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
171pub struct PurgeCandidate {
172 job_instance_id: JobInstanceId,
173 job_execution_id: JobExecutionId,
174 version: ExecutionVersion,
175}
176
177impl PurgeCandidate {
178 #[must_use]
180 pub const fn new(
181 job_instance_id: JobInstanceId,
182 job_execution_id: JobExecutionId,
183 version: ExecutionVersion,
184 ) -> Self {
185 Self {
186 job_instance_id,
187 job_execution_id,
188 version,
189 }
190 }
191
192 #[must_use]
194 pub const fn job_instance_id(&self) -> JobInstanceId {
195 self.job_instance_id
196 }
197
198 #[must_use]
200 pub const fn job_execution_id(&self) -> JobExecutionId {
201 self.job_execution_id
202 }
203
204 #[must_use]
206 pub const fn version(&self) -> ExecutionVersion {
207 self.version
208 }
209}
210
211#[derive(Clone, Copy, Debug, Default, Eq, Hash, Ord, PartialEq, PartialOrd)]
213pub struct PurgeCounts {
214 flow_decisions: u64,
215 recovery_decisions: u64,
216 operator_requests: u64,
217 step_partitions: u64,
218 step_executions: u64,
219 job_executions: u64,
220 job_instances: u64,
221}
222
223impl PurgeCounts {
224 #[must_use]
226 pub const fn new(
227 flow_decisions: u64,
228 recovery_decisions: u64,
229 operator_requests: u64,
230 step_partitions: u64,
231 step_executions: u64,
232 job_executions: u64,
233 job_instances: u64,
234 ) -> Self {
235 Self {
236 flow_decisions,
237 recovery_decisions,
238 operator_requests,
239 step_partitions,
240 step_executions,
241 job_executions,
242 job_instances,
243 }
244 }
245
246 #[must_use]
248 pub const fn flow_decisions(self) -> u64 {
249 self.flow_decisions
250 }
251
252 #[must_use]
254 pub const fn recovery_decisions(self) -> u64 {
255 self.recovery_decisions
256 }
257
258 #[must_use]
260 pub const fn operator_requests(self) -> u64 {
261 self.operator_requests
262 }
263
264 #[must_use]
266 pub const fn step_partitions(self) -> u64 {
267 self.step_partitions
268 }
269
270 #[must_use]
272 pub const fn step_executions(self) -> u64 {
273 self.step_executions
274 }
275
276 #[must_use]
278 pub const fn job_executions(self) -> u64 {
279 self.job_executions
280 }
281
282 #[must_use]
284 pub const fn job_instances(self) -> u64 {
285 self.job_instances
286 }
287}
288
289#[derive(Clone, Debug, Default, Eq, PartialEq)]
291pub struct PurgeSurvey {
292 candidates: Vec<PurgeCandidate>,
293 counts: PurgeCounts,
294}
295
296impl PurgeSurvey {
297 #[must_use]
299 pub const fn new(candidates: Vec<PurgeCandidate>, counts: PurgeCounts) -> Self {
300 Self { candidates, counts }
301 }
302
303 #[must_use]
305 pub fn candidates(&self) -> &[PurgeCandidate] {
306 &self.candidates
307 }
308
309 #[must_use]
311 pub const fn counts(&self) -> PurgeCounts {
312 self.counts
313 }
314}
315
316#[derive(Clone, Debug, Eq, PartialEq)]
318pub struct PurgePlan {
319 request: PurgePlanRequest,
320 candidates: Vec<PurgeCandidate>,
321 counts: PurgeCounts,
322 digest: [u8; 32],
323}
324
325impl PurgePlan {
326 #[doc(hidden)]
328 #[must_use]
329 pub fn new(request: PurgePlanRequest, survey: PurgeSurvey) -> Self {
330 let digest = plan_digest(&request, survey.candidates());
331 Self {
332 request,
333 candidates: survey.candidates,
334 counts: survey.counts,
335 digest,
336 }
337 }
338
339 #[must_use]
341 pub const fn request(&self) -> &PurgePlanRequest {
342 &self.request
343 }
344
345 #[must_use]
347 pub fn candidates(&self) -> &[PurgeCandidate] {
348 &self.candidates
349 }
350
351 #[must_use]
353 pub const fn counts(&self) -> PurgeCounts {
354 self.counts
355 }
356
357 #[must_use]
359 pub const fn digest(&self) -> &[u8; 32] {
360 &self.digest
361 }
362
363 #[must_use]
365 pub fn digest_hex(&self) -> String {
366 hex_digest(&self.digest)
367 }
368
369 #[must_use]
371 pub fn is_empty(&self) -> bool {
372 self.candidates.is_empty()
373 }
374}
375
376fn plan_digest(request: &PurgePlanRequest, candidates: &[PurgeCandidate]) -> [u8; 32] {
377 let mut writer = CanonicalWriter::new("oxide-batch.retention-plan.v1");
378 writer.push_str(request.job_name().as_str());
379 for status in request.statuses().iter() {
380 writer.push_str(status.as_str());
381 }
382 writer.push_u64(request.minimum_age().as_secs());
383 writer.push_u64(u64::from(request.batch().get()));
384 for candidate in candidates {
385 writer.push_u64(candidate.job_instance_id().get());
386 writer.push_u64(candidate.job_execution_id().get());
387 writer.push_u64(candidate.version().get());
388 }
389 writer.digest()
390}
391
392#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
394#[non_exhaustive]
395pub enum RetentionAction {
396 Hold,
398 ReleaseHold,
400 ApplyPurge,
402}
403
404impl RetentionAction {
405 #[must_use]
407 pub const fn as_str(self) -> &'static str {
408 match self {
409 Self::Hold => "HOLD",
410 Self::ReleaseHold => "RELEASE_HOLD",
411 Self::ApplyPurge => "APPLY_PURGE",
412 }
413 }
414}
415
416impl fmt::Display for RetentionAction {
417 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
418 formatter.write_str(self.as_str())
419 }
420}
421
422#[derive(Clone, Debug, Eq, PartialEq)]
427pub struct RetentionHold {
428 job_instance_id: JobInstanceId,
429 actor: ActorRef,
430 reason: ReasonCode,
431 placed_at: SystemTime,
432}
433
434impl RetentionHold {
435 #[must_use]
437 pub const fn new(
438 job_instance_id: JobInstanceId,
439 actor: ActorRef,
440 reason: ReasonCode,
441 placed_at: SystemTime,
442 ) -> Self {
443 Self {
444 job_instance_id,
445 actor,
446 reason,
447 placed_at,
448 }
449 }
450
451 #[must_use]
453 pub const fn job_instance_id(&self) -> JobInstanceId {
454 self.job_instance_id
455 }
456
457 #[must_use]
459 pub const fn actor(&self) -> &ActorRef {
460 &self.actor
461 }
462
463 #[must_use]
465 pub const fn reason(&self) -> &ReasonCode {
466 &self.reason
467 }
468
469 #[must_use]
471 pub const fn placed_at(&self) -> SystemTime {
472 self.placed_at
473 }
474}
475
476#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
478#[non_exhaustive]
479pub enum RetentionOutcome {
480 Applied,
482 Replayed,
484 Rejected,
486}
487
488impl RetentionOutcome {
489 #[must_use]
491 pub const fn as_str(self) -> &'static str {
492 match self {
493 Self::Applied => "APPLIED",
494 Self::Replayed => "REPLAYED",
495 Self::Rejected => "REJECTED",
496 }
497 }
498}
499
500#[derive(Clone, Debug, Eq, PartialEq)]
502pub struct RetentionRecord {
503 id: RetentionActionId,
504 action: RetentionAction,
505 operation_id: OperationId,
506 actor: ActorRef,
507 reason: ReasonCode,
508 job_instance_id: Option<JobInstanceId>,
509 plan_digest: Option<[u8; 32]>,
510 counts: PurgeCounts,
511 batch_bound: Option<PurgeBatchBound>,
512 outcome: RetentionOutcome,
513 applied_at: SystemTime,
514}
515
516impl RetentionRecord {
517 #[must_use]
519 pub fn from_parts(id: RetentionActionId, draft: RetentionRecordDraft) -> Self {
520 Self {
521 id,
522 action: draft.action,
523 operation_id: draft.operation_id,
524 actor: draft.actor,
525 reason: draft.reason,
526 job_instance_id: draft.job_instance_id,
527 plan_digest: draft.plan_digest,
528 counts: draft.counts,
529 batch_bound: draft.batch_bound,
530 outcome: draft.outcome,
531 applied_at: draft.applied_at,
532 }
533 }
534
535 #[must_use]
537 pub const fn id(&self) -> RetentionActionId {
538 self.id
539 }
540
541 #[must_use]
543 pub const fn action(&self) -> RetentionAction {
544 self.action
545 }
546
547 #[must_use]
549 pub const fn operation_id(&self) -> &OperationId {
550 &self.operation_id
551 }
552
553 #[must_use]
555 pub const fn actor(&self) -> &ActorRef {
556 &self.actor
557 }
558
559 #[must_use]
561 pub const fn reason(&self) -> &ReasonCode {
562 &self.reason
563 }
564
565 #[must_use]
567 pub const fn job_instance_id(&self) -> Option<JobInstanceId> {
568 self.job_instance_id
569 }
570
571 #[must_use]
573 pub const fn plan_digest(&self) -> Option<&[u8; 32]> {
574 self.plan_digest.as_ref()
575 }
576
577 #[must_use]
579 pub const fn counts(&self) -> PurgeCounts {
580 self.counts
581 }
582
583 #[must_use]
585 pub const fn batch_bound(&self) -> Option<PurgeBatchBound> {
586 self.batch_bound
587 }
588
589 #[must_use]
591 pub const fn outcome(&self) -> RetentionOutcome {
592 self.outcome
593 }
594
595 #[must_use]
597 pub const fn applied_at(&self) -> SystemTime {
598 self.applied_at
599 }
600}
601
602#[derive(Clone, Debug, Eq, PartialEq)]
604pub struct RetentionRecordDraft {
605 action: RetentionAction,
606 operation_id: OperationId,
607 actor: ActorRef,
608 reason: ReasonCode,
609 job_instance_id: Option<JobInstanceId>,
610 plan_digest: Option<[u8; 32]>,
611 counts: PurgeCounts,
612 batch_bound: Option<PurgeBatchBound>,
613 outcome: RetentionOutcome,
614 applied_at: SystemTime,
615}
616
617impl RetentionRecordDraft {
618 #[must_use]
623 pub fn instance_action(
624 action: RetentionAction,
625 operation_id: OperationId,
626 actor: ActorRef,
627 reason: ReasonCode,
628 job_instance_id: JobInstanceId,
629 applied_at: SystemTime,
630 ) -> Self {
631 Self {
632 action,
633 operation_id,
634 actor,
635 reason,
636 job_instance_id: Some(job_instance_id),
637 plan_digest: None,
638 counts: PurgeCounts::default(),
639 batch_bound: None,
640 outcome: RetentionOutcome::Applied,
641 applied_at,
642 }
643 }
644
645 #[must_use]
651 pub const fn purge(
652 operation_id: OperationId,
653 actor: ActorRef,
654 reason: ReasonCode,
655 plan_digest: [u8; 32],
656 counts: PurgeCounts,
657 batch_bound: PurgeBatchBound,
658 applied_at: SystemTime,
659 ) -> Self {
660 Self {
661 action: RetentionAction::ApplyPurge,
662 operation_id,
663 actor,
664 reason,
665 job_instance_id: None,
666 plan_digest: Some(plan_digest),
667 counts,
668 batch_bound: Some(batch_bound),
669 outcome: RetentionOutcome::Applied,
670 applied_at,
671 }
672 }
673
674 #[must_use]
676 #[allow(clippy::too_many_arguments)]
677 pub const fn from_durable(
678 action: RetentionAction,
679 operation_id: OperationId,
680 actor: ActorRef,
681 reason: ReasonCode,
682 job_instance_id: Option<JobInstanceId>,
683 plan_digest: Option<[u8; 32]>,
684 counts: PurgeCounts,
685 batch_bound: Option<PurgeBatchBound>,
686 outcome: RetentionOutcome,
687 applied_at: SystemTime,
688 ) -> Self {
689 Self {
690 action,
691 operation_id,
692 actor,
693 reason,
694 job_instance_id,
695 plan_digest,
696 counts,
697 batch_bound,
698 outcome,
699 applied_at,
700 }
701 }
702
703 #[must_use]
705 pub const fn action(&self) -> RetentionAction {
706 self.action
707 }
708
709 #[must_use]
711 pub const fn operation_id(&self) -> &OperationId {
712 &self.operation_id
713 }
714
715 #[must_use]
717 pub const fn actor(&self) -> &ActorRef {
718 &self.actor
719 }
720
721 #[must_use]
723 pub const fn reason(&self) -> &ReasonCode {
724 &self.reason
725 }
726
727 #[must_use]
729 pub const fn job_instance_id(&self) -> Option<JobInstanceId> {
730 self.job_instance_id
731 }
732
733 #[must_use]
735 pub const fn plan_digest(&self) -> Option<&[u8; 32]> {
736 self.plan_digest.as_ref()
737 }
738
739 #[must_use]
741 pub const fn counts(&self) -> PurgeCounts {
742 self.counts
743 }
744
745 #[must_use]
747 pub const fn batch_bound(&self) -> Option<PurgeBatchBound> {
748 self.batch_bound
749 }
750
751 #[must_use]
753 pub const fn outcome(&self) -> RetentionOutcome {
754 self.outcome
755 }
756
757 #[must_use]
759 pub const fn applied_at(&self) -> SystemTime {
760 self.applied_at
761 }
762}
763
764#[derive(Clone, Debug, Eq, PartialEq)]
766#[non_exhaustive]
767pub enum RetentionError {
768 BatchBoundOutOfRange {
770 requested: u32,
772 },
773 AgeBoundTooSmall {
775 minimum: Duration,
777 },
778 NonTerminalStatus {
780 status: BatchStatus,
782 },
783 EmptyStatusSet,
785 RetentionPlanStale,
787 InstanceHeld {
789 job_instance_id: JobInstanceId,
791 },
792 OperationIdConflict {
794 action: RetentionAction,
796 operation_id: OperationId,
798 },
799 OperationOutcomeUnknown,
801 Repository(RepositoryError),
803}
804
805impl fmt::Display for RetentionError {
806 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
807 match self {
808 Self::BatchBoundOutOfRange { requested } => write!(
809 formatter,
810 "purge batch bound {requested} is outside 1..={MAX_PURGE_BATCH}"
811 ),
812 Self::AgeBoundTooSmall { minimum } => write!(
813 formatter,
814 "the minimum age must be at least {} seconds",
815 minimum.as_secs()
816 ),
817 Self::NonTerminalStatus { status } => {
818 write!(formatter, "{status} is not a finished status")
819 }
820 Self::EmptyStatusSet => formatter.write_str("a purge must target at least one status"),
821 Self::RetentionPlanStale => {
822 formatter.write_str("the purge plan is stale and nothing was deleted")
823 }
824 Self::InstanceHeld { job_instance_id } => {
825 write!(formatter, "job instance {job_instance_id} is held")
826 }
827 Self::OperationIdConflict {
828 action,
829 operation_id,
830 } => write!(
831 formatter,
832 "operation identifier {operation_id} was already recorded for {action} with a different request"
833 ),
834 Self::OperationOutcomeUnknown => {
835 formatter.write_str("the retention commit outcome is unknown")
836 }
837 Self::Repository(error) => error.fmt(formatter),
838 }
839 }
840}
841
842impl Error for RetentionError {
843 fn source(&self) -> Option<&(dyn Error + 'static)> {
844 match self {
845 Self::Repository(error) => Some(error),
846 _ => None,
847 }
848 }
849}
850
851impl From<RepositoryError> for RetentionError {
852 fn from(value: RepositoryError) -> Self {
853 match value {
854 RepositoryError::CommitOutcomeUnknown => Self::OperationOutcomeUnknown,
855 RepositoryError::RetentionPlanStale => Self::RetentionPlanStale,
856 other => Self::Repository(other),
857 }
858 }
859}