oxide_batch_core/domain/
execution.rs

1use std::fmt;
2use std::time::SystemTime;
3
4use super::lifecycle::{validate_expected_version, validate_restart, validate_transition};
5use super::{
6    DomainError, ExecutionVersion, ExitCode, FailureId, JobExecutionId, JobInstanceId,
7    JobInstanceKey, LifecycleError, LifecycleTransition, StepExecutionId, StepName,
8};
9
10/// The framework lifecycle status of a job or step execution.
11#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
12#[non_exhaustive]
13pub enum BatchStatus {
14    /// Metadata exists and user work has not started.
15    Starting,
16    /// User work is running.
17    Started,
18    /// A cooperative stop is in progress.
19    Stopping,
20    /// The attempt stopped cooperatively and may be restarted.
21    Stopped,
22    /// The attempt failed and may be restartable.
23    Failed,
24    /// The attempt completed successfully.
25    Completed,
26    /// The instance is intentionally terminal and cannot restart.
27    Abandoned,
28    /// The durable outcome is ambiguous and requires recovery.
29    Unknown,
30}
31
32impl BatchStatus {
33    /// Returns whether the attempt is actively running or stopping.
34    #[must_use]
35    pub const fn is_active(self) -> bool {
36        matches!(self, Self::Starting | Self::Started | Self::Stopping)
37    }
38
39    /// Returns whether the attempt has a known finished outcome.
40    #[must_use]
41    pub const fn is_finished(self) -> bool {
42        matches!(
43            self,
44            Self::Stopped | Self::Failed | Self::Completed | Self::Abandoned
45        )
46    }
47
48    /// Returns whether the logical instance is terminal and not restartable.
49    #[must_use]
50    pub const fn is_terminal(self) -> bool {
51        matches!(self, Self::Completed | Self::Abandoned)
52    }
53
54    /// Returns the stable durable code for this status.
55    ///
56    /// The code is the same value durable adapters store and audit records
57    /// carry, so a projection never renames a status.
58    #[must_use]
59    pub const fn as_str(self) -> &'static str {
60        match self {
61            Self::Starting => "STARTING",
62            Self::Started => "STARTED",
63            Self::Stopping => "STOPPING",
64            Self::Stopped => "STOPPED",
65            Self::Failed => "FAILED",
66            Self::Completed => "COMPLETED",
67            Self::Abandoned => "ABANDONED",
68            Self::Unknown => "UNKNOWN",
69        }
70    }
71}
72
73impl fmt::Display for BatchStatus {
74    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
75        formatter.write_str(self.as_str())
76    }
77}
78
79/// A flow- and operator-facing result kept separate from [`BatchStatus`].
80#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
81pub struct ExitStatus {
82    code: ExitCode,
83}
84
85impl ExitStatus {
86    /// Constructs an exit status from a validated code.
87    #[must_use]
88    pub const fn new(code: ExitCode) -> Self {
89        Self { code }
90    }
91
92    /// Constructs the conventional `UNKNOWN` exit status.
93    ///
94    /// This cannot fail because the framework-owned code is valid.
95    #[must_use]
96    pub fn unknown() -> Self {
97        Self {
98            code: ExitCode::framework_owned("UNKNOWN"),
99        }
100    }
101
102    /// Constructs the conventional `COMPLETED` exit status.
103    #[must_use]
104    pub fn completed() -> Self {
105        Self {
106            code: ExitCode::framework_owned("COMPLETED"),
107        }
108    }
109
110    /// Constructs the conventional `FAILED` exit status.
111    #[must_use]
112    pub fn failed() -> Self {
113        Self {
114            code: ExitCode::framework_owned("FAILED"),
115        }
116    }
117
118    /// Constructs the conventional `STOPPED` exit status.
119    #[must_use]
120    pub fn stopped() -> Self {
121        Self {
122            code: ExitCode::framework_owned("STOPPED"),
123        }
124    }
125
126    /// Borrows the exit code.
127    #[must_use]
128    pub const fn code(&self) -> &ExitCode {
129        &self.code
130    }
131}
132
133impl fmt::Display for ExitStatus {
134    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
135        self.code.fmt(formatter)
136    }
137}
138
139/// Durable item and transaction counters for an execution.
140#[derive(Clone, Copy, Debug, Default, Eq, Hash, Ord, PartialEq, PartialOrd)]
141#[non_exhaustive]
142pub struct ExecutionCounts {
143    read: u64,
144    processed: u64,
145    written: u64,
146    filtered: u64,
147    committed: u64,
148    rolled_back: u64,
149}
150
151impl ExecutionCounts {
152    /// Constructs a complete counter snapshot.
153    #[must_use]
154    pub const fn new(
155        read: u64,
156        processed: u64,
157        written: u64,
158        filtered: u64,
159        committed: u64,
160        rolled_back: u64,
161    ) -> Self {
162        Self {
163            read,
164            processed,
165            written,
166            filtered,
167            committed,
168            rolled_back,
169        }
170    }
171
172    /// Returns the durable read count.
173    #[must_use]
174    pub const fn read(self) -> u64 {
175        self.read
176    }
177
178    /// Returns the durable processed count.
179    #[must_use]
180    pub const fn processed(self) -> u64 {
181        self.processed
182    }
183
184    /// Returns the durable written count.
185    #[must_use]
186    pub const fn written(self) -> u64 {
187        self.written
188    }
189
190    /// Returns the durable filtered count.
191    #[must_use]
192    pub const fn filtered(self) -> u64 {
193        self.filtered
194    }
195
196    /// Returns the committed chunk/transaction count.
197    #[must_use]
198    pub const fn committed(self) -> u64 {
199        self.committed
200    }
201
202    /// Returns the rolled-back chunk/transaction count.
203    #[must_use]
204    pub const fn rolled_back(self) -> u64 {
205        self.rolled_back
206    }
207
208    fn with_terminal_rollback(self) -> Result<Self, LifecycleError> {
209        let rolled_back = self
210            .rolled_back
211            .checked_add(1)
212            .ok_or(LifecycleError::CountExhausted)?;
213        Ok(Self {
214            rolled_back,
215            ..self
216        })
217    }
218}
219
220/// Validated creation, start, and end instants for an execution attempt.
221#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
222#[allow(clippy::struct_field_names)]
223pub struct ExecutionTimestamps {
224    created_at: SystemTime,
225    started_at: Option<SystemTime>,
226    ended_at: Option<SystemTime>,
227}
228
229impl ExecutionTimestamps {
230    /// Validates and constructs an execution timestamp snapshot.
231    ///
232    /// # Errors
233    ///
234    /// Returns [`DomainError::InvalidTimestampOrder`] when start precedes
235    /// creation or end precedes creation/start.
236    pub fn new(
237        created_at: SystemTime,
238        started_at: Option<SystemTime>,
239        ended_at: Option<SystemTime>,
240    ) -> Result<Self, DomainError> {
241        if started_at.is_some_and(|started| started < created_at)
242            || ended_at.is_some_and(|ended| ended < started_at.unwrap_or(created_at))
243        {
244            return Err(DomainError::InvalidTimestampOrder);
245        }
246        Ok(Self {
247            created_at,
248            started_at,
249            ended_at,
250        })
251    }
252
253    /// Returns the metadata creation instant.
254    #[must_use]
255    pub const fn created_at(self) -> SystemTime {
256        self.created_at
257    }
258
259    /// Returns when user work began, when known.
260    #[must_use]
261    pub const fn started_at(self) -> Option<SystemTime> {
262        self.started_at
263    }
264
265    /// Returns when the attempt ended, when known.
266    #[must_use]
267    pub const fn ended_at(self) -> Option<SystemTime> {
268        self.ended_at
269    }
270}
271
272/// A stable framework category for a redacted failure.
273#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
274#[non_exhaustive]
275pub enum FailureCategory {
276    /// Invalid job definition or launch configuration.
277    InvalidDefinition,
278    /// Duplicate, completed, or non-restartable execution.
279    DuplicateExecution,
280    /// Illegal or conflicting lifecycle transition.
281    IllegalTransition,
282    /// A transient repository or infrastructure failure.
283    TransientInfrastructure,
284    /// A permanent repository or infrastructure failure.
285    PermanentInfrastructure,
286    /// A user reader, processor, writer, tasklet, or listener failure.
287    UserComponent,
288    /// Cancellation, stop, or deadline expiry.
289    Cancelled,
290    /// Serialization or version incompatibility.
291    Serialization,
292    /// A framework invariant was violated.
293    Invariant,
294    /// An optimistic version check lost to a concurrent writer.
295    OptimisticConflict,
296    /// A bounded operation exceeded its deadline.
297    Timeout,
298    /// The selected resource cannot provide a required capability.
299    UnsupportedCapability,
300    /// A commit outcome is unknown and must never be guessed.
301    UnknownCommit,
302    /// A bounded shutdown could not join or resolve every owned child.
303    ShutdownIncomplete,
304    /// An execution was resolved from stale-ownership evidence.
305    StaleRecovered,
306}
307
308impl FailureCategory {
309    /// Returns the stable, low-cardinality manifest and telemetry name.
310    #[must_use]
311    pub const fn as_str(self) -> &'static str {
312        match self {
313            Self::InvalidDefinition => "invalid_definition",
314            Self::DuplicateExecution => "duplicate_execution",
315            Self::IllegalTransition => "illegal_transition",
316            Self::TransientInfrastructure => "transient_infrastructure",
317            Self::PermanentInfrastructure => "permanent_infrastructure",
318            Self::UserComponent => "user_component",
319            Self::Cancelled => "cancelled",
320            Self::Serialization => "serialization",
321            Self::Invariant => "invariant",
322            Self::OptimisticConflict => "optimistic_conflict",
323            Self::Timeout => "timeout",
324            Self::UnsupportedCapability => "unsupported_capability",
325            Self::UnknownCommit => "unknown_commit",
326            Self::ShutdownIncomplete => "shutdown_incomplete",
327            Self::StaleRecovered => "stale_recovered",
328        }
329    }
330
331    /// Returns whether a fault policy may retry or skip this category.
332    ///
333    /// Definition, lifecycle, cancellation, serialization, invariant,
334    /// capability, and unknown-commit failures always fail closed.
335    #[must_use]
336    pub const fn is_policy_eligible(self) -> bool {
337        matches!(
338            self,
339            Self::TransientInfrastructure
340                | Self::PermanentInfrastructure
341                | Self::UserComponent
342                | Self::OptimisticConflict
343                | Self::Timeout
344        )
345    }
346
347    /// Returns the stable code durable adapters persist for this category.
348    ///
349    /// The spelling is durable data: an existing code is never renamed, and a
350    /// new variant only ever adds a code.
351    #[must_use]
352    pub const fn durable_code(self) -> &'static str {
353        match self {
354            Self::InvalidDefinition => "INVALID_DEFINITION",
355            Self::DuplicateExecution => "DUPLICATE_EXECUTION",
356            Self::IllegalTransition => "ILLEGAL_TRANSITION",
357            Self::TransientInfrastructure => "TRANSIENT_INFRASTRUCTURE",
358            Self::PermanentInfrastructure => "PERMANENT_INFRASTRUCTURE",
359            Self::UserComponent => "USER_COMPONENT",
360            Self::Cancelled => "CANCELLED",
361            Self::Serialization => "SERIALIZATION",
362            Self::Invariant => "INVARIANT",
363            Self::OptimisticConflict => "OPTIMISTIC_CONFLICT",
364            Self::Timeout => "TIMEOUT",
365            Self::UnsupportedCapability => "UNSUPPORTED_CAPABILITY",
366            Self::UnknownCommit => "UNKNOWN_COMMIT",
367            Self::ShutdownIncomplete => "SHUTDOWN_INCOMPLETE",
368            Self::StaleRecovered => "STALE_RECOVERED",
369        }
370    }
371
372    /// Returns the category for one durable code, rejecting unknown values.
373    #[must_use]
374    pub fn from_durable_code(value: &str) -> Option<Self> {
375        Some(match value {
376            "INVALID_DEFINITION" => Self::InvalidDefinition,
377            "DUPLICATE_EXECUTION" => Self::DuplicateExecution,
378            "ILLEGAL_TRANSITION" => Self::IllegalTransition,
379            "TRANSIENT_INFRASTRUCTURE" => Self::TransientInfrastructure,
380            "PERMANENT_INFRASTRUCTURE" => Self::PermanentInfrastructure,
381            "USER_COMPONENT" => Self::UserComponent,
382            "CANCELLED" => Self::Cancelled,
383            "SERIALIZATION" => Self::Serialization,
384            "INVARIANT" => Self::Invariant,
385            "OPTIMISTIC_CONFLICT" => Self::OptimisticConflict,
386            "TIMEOUT" => Self::Timeout,
387            "UNSUPPORTED_CAPABILITY" => Self::UnsupportedCapability,
388            "UNKNOWN_COMMIT" => Self::UnknownCommit,
389            "SHUTDOWN_INCOMPLETE" => Self::ShutdownIncomplete,
390            "STALE_RECOVERED" => Self::StaleRecovered,
391            _ => return None,
392        })
393    }
394}
395
396/// A value-redacted failure summary suitable for execution inspection.
397#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
398pub struct FailureSummary {
399    category: FailureCategory,
400    failure_id: FailureId,
401}
402
403impl FailureSummary {
404    /// Constructs a failure summary from a stable category and opaque ID.
405    #[must_use]
406    pub const fn new(category: FailureCategory, failure_id: FailureId) -> Self {
407        Self {
408            category,
409            failure_id,
410        }
411    }
412
413    /// Returns the stable failure category.
414    #[must_use]
415    pub const fn category(self) -> FailureCategory {
416        self.category
417    }
418
419    /// Returns the opaque diagnostic correlation ID.
420    #[must_use]
421    pub const fn failure_id(self) -> FailureId {
422        self.failure_id
423    }
424}
425
426/// Validated lifecycle, outcome, timestamps, counters, and failure metadata.
427#[derive(Clone, Debug, Eq, PartialEq)]
428pub struct ExecutionMetadata {
429    status: BatchStatus,
430    exit_status: ExitStatus,
431    timestamps: ExecutionTimestamps,
432    counts: ExecutionCounts,
433    failure: Option<FailureSummary>,
434}
435
436impl ExecutionMetadata {
437    /// Validates and constructs an execution metadata snapshot.
438    ///
439    /// # Errors
440    ///
441    /// Active executions cannot have an end time, known finished executions
442    /// require one, and failed executions require a redacted failure summary.
443    pub fn new(
444        status: BatchStatus,
445        exit_status: ExitStatus,
446        timestamps: ExecutionTimestamps,
447        counts: ExecutionCounts,
448        failure: Option<FailureSummary>,
449    ) -> Result<Self, DomainError> {
450        if status.is_active() && timestamps.ended_at().is_some() {
451            return Err(DomainError::ActiveExecutionHasEndTime);
452        }
453        if status.is_finished() && timestamps.ended_at().is_none() {
454            return Err(DomainError::FinishedExecutionMissingEndTime);
455        }
456        if matches!(status, BatchStatus::Failed) && failure.is_none() {
457            return Err(DomainError::FailedExecutionMissingFailure);
458        }
459        Ok(Self {
460            status,
461            exit_status,
462            timestamps,
463            counts,
464            failure,
465        })
466    }
467
468    /// Returns the framework lifecycle status.
469    #[must_use]
470    pub const fn status(&self) -> BatchStatus {
471        self.status
472    }
473
474    /// Borrows the flow/operator exit status.
475    #[must_use]
476    pub const fn exit_status(&self) -> &ExitStatus {
477        &self.exit_status
478    }
479
480    /// Returns the validated timestamps.
481    #[must_use]
482    pub const fn timestamps(&self) -> ExecutionTimestamps {
483        self.timestamps
484    }
485
486    /// Returns the durable counters.
487    #[must_use]
488    pub const fn counts(&self) -> ExecutionCounts {
489        self.counts
490    }
491
492    /// Returns the redacted failure summary, when present.
493    #[must_use]
494    pub const fn failure(&self) -> Option<FailureSummary> {
495        self.failure
496    }
497
498    fn transition(&self, transition: LifecycleTransition) -> Result<Self, LifecycleError> {
499        validate_transition(self.status, transition)?;
500
501        let target = transition.target();
502        let transitioned_at = transition.transitioned_at();
503        let current_timestamps = self.timestamps;
504        if transitioned_at
505            < current_timestamps
506                .started_at()
507                .unwrap_or(current_timestamps.created_at())
508            || current_timestamps
509                .ended_at()
510                .is_some_and(|ended_at| transitioned_at < ended_at)
511        {
512            return Err(LifecycleError::InvalidTransitionTime {
513                source: DomainError::InvalidTimestampOrder,
514            });
515        }
516        let started_at = if matches!(target, BatchStatus::Started) {
517            Some(transitioned_at)
518        } else {
519            current_timestamps.started_at()
520        };
521        let ended_at = if target.is_finished() {
522            current_timestamps.ended_at().or(Some(transitioned_at))
523        } else {
524            None
525        };
526        let timestamps =
527            ExecutionTimestamps::new(current_timestamps.created_at(), started_at, ended_at)
528                .map_err(|source| LifecycleError::InvalidTransitionTime { source })?;
529        let failure = transition.failure().or(self.failure);
530
531        let counts = if transition.terminal_rollback() {
532            self.counts.with_terminal_rollback()?
533        } else {
534            self.counts
535        };
536
537        Self::new(
538            target,
539            self.exit_status.clone(),
540            timestamps,
541            counts,
542            failure,
543        )
544        .map_err(|source| match source {
545            DomainError::FailedExecutionMissingFailure => {
546                LifecycleError::FailedTransitionMissingFailure
547            }
548            source => LifecycleError::InvalidTransitionTime { source },
549        })
550    }
551
552    fn with_exit_status(&self, exit_status: ExitStatus) -> Self {
553        Self {
554            status: self.status,
555            exit_status,
556            timestamps: self.timestamps,
557            counts: self.counts,
558            failure: self.failure,
559        }
560    }
561
562    fn starting(created_at: SystemTime) -> Self {
563        Self {
564            status: BatchStatus::Starting,
565            exit_status: ExitStatus::unknown(),
566            timestamps: ExecutionTimestamps {
567                created_at,
568                started_at: None,
569                ended_at: None,
570            },
571            counts: ExecutionCounts::default(),
572            failure: None,
573        }
574    }
575}
576
577/// One logical occurrence of a named job and its identifying parameters.
578#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
579pub struct JobInstance {
580    id: JobInstanceId,
581    key: JobInstanceKey,
582}
583
584impl JobInstance {
585    /// Constructs a logical job instance.
586    #[must_use]
587    pub const fn new(id: JobInstanceId, key: JobInstanceKey) -> Self {
588        Self { id, key }
589    }
590
591    /// Returns the opaque instance identifier.
592    #[must_use]
593    pub const fn id(&self) -> JobInstanceId {
594        self.id
595    }
596
597    /// Borrows the canonical logical key.
598    #[must_use]
599    pub const fn key(&self) -> &JobInstanceKey {
600        &self.key
601    }
602}
603
604/// One launch or restart attempt for a [`JobInstance`].
605#[derive(Clone, Debug, Eq, PartialEq)]
606pub struct JobExecution {
607    id: JobExecutionId,
608    job_instance_id: JobInstanceId,
609    metadata: ExecutionMetadata,
610    version: ExecutionVersion,
611}
612
613impl JobExecution {
614    /// Constructs a job execution record from validated metadata.
615    #[must_use]
616    pub const fn new(
617        id: JobExecutionId,
618        job_instance_id: JobInstanceId,
619        metadata: ExecutionMetadata,
620    ) -> Self {
621        Self {
622            id,
623            job_instance_id,
624            metadata,
625            version: ExecutionVersion::INITIAL,
626        }
627    }
628
629    /// Reconstructs a job execution snapshot with its optimistic version.
630    #[must_use]
631    pub const fn from_snapshot(
632        id: JobExecutionId,
633        job_instance_id: JobInstanceId,
634        metadata: ExecutionMetadata,
635        version: ExecutionVersion,
636    ) -> Self {
637        Self {
638            id,
639            job_instance_id,
640            metadata,
641            version,
642        }
643    }
644
645    /// Returns the attempt identifier.
646    #[must_use]
647    pub const fn id(&self) -> JobExecutionId {
648        self.id
649    }
650
651    /// Returns the logical instance identifier.
652    #[must_use]
653    pub const fn job_instance_id(&self) -> JobInstanceId {
654        self.job_instance_id
655    }
656
657    /// Borrows the execution metadata.
658    #[must_use]
659    pub const fn metadata(&self) -> &ExecutionMetadata {
660        &self.metadata
661    }
662
663    /// Returns the facade-owned optimistic-lock version.
664    #[must_use]
665    pub const fn version(&self) -> ExecutionVersion {
666        self.version
667    }
668
669    /// Applies one legal lifecycle transition using compare-and-swap semantics.
670    ///
671    /// The transition never changes exit status. Restart requests from
672    /// `STOPPED` or `FAILED` return
673    /// [`LifecycleError::RestartRequiresNewAttempt`].
674    ///
675    /// # Errors
676    ///
677    /// Returns a typed stale-version, illegal-transition, timestamp, failure,
678    /// or version-exhaustion error without mutating this snapshot.
679    pub fn transition(
680        &mut self,
681        expected_version: ExecutionVersion,
682        transition: LifecycleTransition,
683    ) -> Result<ExecutionVersion, LifecycleError> {
684        transition_execution(
685            &mut self.metadata,
686            &mut self.version,
687            expected_version,
688            transition,
689        )
690    }
691
692    /// Enriches flow/operator exit status without changing batch status.
693    ///
694    /// # Errors
695    ///
696    /// Returns [`LifecycleError::StaleVersion`] or
697    /// [`LifecycleError::VersionExhausted`] without mutating this snapshot.
698    pub fn enrich_exit_status(
699        &mut self,
700        expected_version: ExecutionVersion,
701        exit_status: ExitStatus,
702    ) -> Result<ExecutionVersion, LifecycleError> {
703        enrich_execution_exit_status(
704            &mut self.metadata,
705            &mut self.version,
706            expected_version,
707            exit_status,
708        )
709    }
710
711    /// Creates a fresh `STARTING` attempt for the same logical job instance.
712    ///
713    /// The prior execution remains unchanged. Definition-level restart
714    /// permission is a repository/launcher concern layered on top of this
715    /// status policy.
716    ///
717    /// # Errors
718    ///
719    /// Returns a typed stale-version, non-restartable, or reused-ID error.
720    pub fn new_restart_attempt(
721        &self,
722        expected_version: ExecutionVersion,
723        new_execution_id: JobExecutionId,
724        created_at: SystemTime,
725    ) -> Result<Self, LifecycleError> {
726        validate_expected_version(expected_version, self.version)?;
727        validate_restart(self.metadata.status())?;
728        if new_execution_id == self.id {
729            return Err(LifecycleError::AttemptIdentifierReused);
730        }
731        validate_restart_time(&self.metadata, created_at)?;
732        Ok(Self::new(
733            new_execution_id,
734            self.job_instance_id,
735            ExecutionMetadata::starting(created_at),
736        ))
737    }
738}
739
740/// One attempt to execute a named step within a job execution.
741#[derive(Clone, Debug, Eq, PartialEq)]
742pub struct StepExecution {
743    id: StepExecutionId,
744    job_execution_id: JobExecutionId,
745    step_name: StepName,
746    metadata: ExecutionMetadata,
747    version: ExecutionVersion,
748}
749
750impl StepExecution {
751    /// Constructs a step execution record from validated metadata.
752    #[must_use]
753    pub const fn new(
754        id: StepExecutionId,
755        job_execution_id: JobExecutionId,
756        step_name: StepName,
757        metadata: ExecutionMetadata,
758    ) -> Self {
759        Self {
760            id,
761            job_execution_id,
762            step_name,
763            metadata,
764            version: ExecutionVersion::INITIAL,
765        }
766    }
767
768    /// Reconstructs a step execution snapshot with its optimistic version.
769    #[must_use]
770    pub const fn from_snapshot(
771        id: StepExecutionId,
772        job_execution_id: JobExecutionId,
773        step_name: StepName,
774        metadata: ExecutionMetadata,
775        version: ExecutionVersion,
776    ) -> Self {
777        Self {
778            id,
779            job_execution_id,
780            step_name,
781            metadata,
782            version,
783        }
784    }
785
786    /// Returns the step-attempt identifier.
787    #[must_use]
788    pub const fn id(&self) -> StepExecutionId {
789        self.id
790    }
791
792    /// Returns the enclosing job-execution identifier.
793    #[must_use]
794    pub const fn job_execution_id(&self) -> JobExecutionId {
795        self.job_execution_id
796    }
797
798    /// Borrows the logical step name.
799    #[must_use]
800    pub const fn step_name(&self) -> &StepName {
801        &self.step_name
802    }
803
804    /// Borrows the execution metadata.
805    #[must_use]
806    pub const fn metadata(&self) -> &ExecutionMetadata {
807        &self.metadata
808    }
809
810    /// Returns the facade-owned optimistic-lock version.
811    #[must_use]
812    pub const fn version(&self) -> ExecutionVersion {
813        self.version
814    }
815
816    /// Applies one legal lifecycle transition using compare-and-swap semantics.
817    ///
818    /// # Errors
819    ///
820    /// Returns a typed lifecycle/conflict error without mutating this snapshot.
821    pub fn transition(
822        &mut self,
823        expected_version: ExecutionVersion,
824        transition: LifecycleTransition,
825    ) -> Result<ExecutionVersion, LifecycleError> {
826        transition_execution(
827            &mut self.metadata,
828            &mut self.version,
829            expected_version,
830            transition,
831        )
832    }
833
834    /// Enriches flow/operator exit status without changing batch status.
835    ///
836    /// # Errors
837    ///
838    /// Returns a typed conflict/version error without mutating this snapshot.
839    pub fn enrich_exit_status(
840        &mut self,
841        expected_version: ExecutionVersion,
842        exit_status: ExitStatus,
843    ) -> Result<ExecutionVersion, LifecycleError> {
844        enrich_execution_exit_status(
845            &mut self.metadata,
846            &mut self.version,
847            expected_version,
848            exit_status,
849        )
850    }
851
852    /// Creates a fresh `STARTING` step attempt under a new job execution.
853    ///
854    /// The prior step execution remains unchanged.
855    ///
856    /// # Errors
857    ///
858    /// Returns a typed stale-version, non-restartable, or reused-ID error.
859    pub fn new_restart_attempt(
860        &self,
861        expected_version: ExecutionVersion,
862        new_execution_id: StepExecutionId,
863        new_job_execution_id: JobExecutionId,
864        created_at: SystemTime,
865    ) -> Result<Self, LifecycleError> {
866        validate_expected_version(expected_version, self.version)?;
867        validate_restart(self.metadata.status())?;
868        if new_execution_id == self.id || new_job_execution_id == self.job_execution_id {
869            return Err(LifecycleError::AttemptIdentifierReused);
870        }
871        validate_restart_time(&self.metadata, created_at)?;
872        Ok(Self::new(
873            new_execution_id,
874            new_job_execution_id,
875            self.step_name.clone(),
876            ExecutionMetadata::starting(created_at),
877        ))
878    }
879}
880
881fn transition_execution(
882    metadata: &mut ExecutionMetadata,
883    version: &mut ExecutionVersion,
884    expected_version: ExecutionVersion,
885    transition: LifecycleTransition,
886) -> Result<ExecutionVersion, LifecycleError> {
887    validate_expected_version(expected_version, *version)?;
888    let updated_metadata = metadata.transition(transition)?;
889    let updated_version = version.next()?;
890    *metadata = updated_metadata;
891    *version = updated_version;
892    Ok(updated_version)
893}
894
895fn enrich_execution_exit_status(
896    metadata: &mut ExecutionMetadata,
897    version: &mut ExecutionVersion,
898    expected_version: ExecutionVersion,
899    exit_status: ExitStatus,
900) -> Result<ExecutionVersion, LifecycleError> {
901    validate_expected_version(expected_version, *version)?;
902    let updated_version = version.next()?;
903    *metadata = metadata.with_exit_status(exit_status);
904    *version = updated_version;
905    Ok(updated_version)
906}
907
908fn validate_restart_time(
909    metadata: &ExecutionMetadata,
910    created_at: SystemTime,
911) -> Result<(), LifecycleError> {
912    if metadata
913        .timestamps()
914        .ended_at()
915        .is_some_and(|ended_at| created_at < ended_at)
916    {
917        return Err(LifecycleError::InvalidTransitionTime {
918            source: DomainError::InvalidTimestampOrder,
919        });
920    }
921    Ok(())
922}
oxide_batch_core/domain/execution.rs

oxide_batch_core/domain/
execution.rs
