oxide_batch_repository/
flow.rs

1//! Durable flow-decision records exchanged with a metadata repository.
2//!
3//! A flow decision is repository-authoritative and append-only: the runtime
4//! validates and proposes one transition, and the adapter allocates its
5//! identity and commits it. The records below carry no engine, runtime, or plan
6//! type, so a metadata adapter can persist and replay them without depending on
7//! the flow engine that produced them.
8
9use std::fmt;
10use std::num::NonZeroU64;
11use std::time::SystemTime;
12
13use oxide_batch_core::{
14    DomainError, ExecutionContext, ExitCode, FlowTarget, IdentifierKind, JobExecutionId, NodeId,
15    StepExecution, StepExecutionId,
16};
17
18/// Opaque durable identifier of one selected transition.
19#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
20pub struct FlowDecisionId(NonZeroU64);
21
22impl FlowDecisionId {
23    /// Constructs a positive flow-decision identifier.
24    ///
25    /// # Errors
26    ///
27    /// Returns [`DomainError::ZeroIdentifier`] for zero.
28    pub fn new(value: u64) -> Result<Self, DomainError> {
29        NonZeroU64::new(value)
30            .map(Self)
31            .ok_or(DomainError::ZeroIdentifier {
32                kind: IdentifierKind::FlowDecision,
33            })
34    }
35
36    /// Returns the numeric identifier.
37    #[must_use]
38    pub const fn get(self) -> u64 {
39        self.0.get()
40    }
41}
42
43impl fmt::Display for FlowDecisionId {
44    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
45        self.get().fmt(formatter)
46    }
47}
48
49/// Positive, execution-local ordering of selected transitions.
50#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
51pub struct FlowDecisionSequence(NonZeroU64);
52
53impl FlowDecisionSequence {
54    /// Constructs a positive sequence.
55    ///
56    /// # Errors
57    ///
58    /// Returns [`DomainError::ZeroIdentifier`] for zero.
59    ///
60    /// [`DomainError::ZeroIdentifier`]: DomainError::ZeroIdentifier
61    pub fn new(value: u64) -> Result<Self, DomainError> {
62        NonZeroU64::new(value)
63            .map(Self)
64            .ok_or(DomainError::ZeroIdentifier {
65                kind: IdentifierKind::FlowDecisionSequence,
66            })
67    }
68
69    /// Returns the numeric sequence.
70    #[must_use]
71    pub const fn get(self) -> u64 {
72        self.0.get()
73    }
74}
75
76/// Why one transition was selected.
77#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
78#[non_exhaustive]
79pub enum FlowTransitionKind {
80    /// A newly executed step produced the observed outcome.
81    StepExit,
82    /// A deterministic decider produced the observed outcome.
83    Decider,
84    /// Restart reused a completed step without invoking it.
85    CompletedStepReuse,
86    /// A bounded local split joined durable branch results in declared order.
87    SplitAggregate,
88}
89
90impl FlowTransitionKind {
91    /// Returns the stable durable code of this transition kind.
92    #[doc(hidden)]
93    #[must_use]
94    pub const fn durable_code(self) -> &'static str {
95        match self {
96            Self::StepExit => "STEP_EXIT",
97            Self::Decider => "DECIDER",
98            Self::CompletedStepReuse => "COMPLETED_STEP_REUSE",
99            Self::SplitAggregate => "SPLIT_AGGREGATE",
100        }
101    }
102
103    /// Reads one stable durable transition-kind code.
104    #[doc(hidden)]
105    #[must_use]
106    pub fn from_durable_code(value: &str) -> Option<Self> {
107        match value {
108            "STEP_EXIT" => Some(Self::StepExit),
109            "DECIDER" => Some(Self::Decider),
110            "COMPLETED_STEP_REUSE" => Some(Self::CompletedStepReuse),
111            "SPLIT_AGGREGATE" => Some(Self::SplitAggregate),
112            _ => None,
113        }
114    }
115}
116
117/// One append-only, repository-authoritative selected transition.
118#[derive(Clone, Debug, Eq, PartialEq)]
119pub struct FlowDecision {
120    id: FlowDecisionId,
121    job_execution_id: JobExecutionId,
122    sequence: FlowDecisionSequence,
123    source_node_id: NodeId,
124    source_step_execution_id: Option<StepExecutionId>,
125    kind: FlowTransitionKind,
126    observed_outcome: ExitCode,
127    target: FlowTarget,
128    plan_fingerprint: [u8; 32],
129    input_digest: [u8; 32],
130    reused_decision_id: Option<FlowDecisionId>,
131    decided_at: SystemTime,
132}
133
134impl FlowDecision {
135    /// Reconstructs one durable flow decision an adapter allocated.
136    #[allow(clippy::too_many_arguments)]
137    #[doc(hidden)]
138    #[must_use]
139    pub const fn new(
140        id: FlowDecisionId,
141        job_execution_id: JobExecutionId,
142        sequence: FlowDecisionSequence,
143        source_node_id: NodeId,
144        source_step_execution_id: Option<StepExecutionId>,
145        kind: FlowTransitionKind,
146        observed_outcome: ExitCode,
147        target: FlowTarget,
148        plan_fingerprint: [u8; 32],
149        input_digest: [u8; 32],
150        reused_decision_id: Option<FlowDecisionId>,
151        decided_at: SystemTime,
152    ) -> Self {
153        Self {
154            id,
155            job_execution_id,
156            sequence,
157            source_node_id,
158            source_step_execution_id,
159            kind,
160            observed_outcome,
161            target,
162            plan_fingerprint,
163            input_digest,
164            reused_decision_id,
165            decided_at,
166        }
167    }
168
169    /// Returns the repository-owned identifier.
170    #[must_use]
171    pub const fn id(&self) -> FlowDecisionId {
172        self.id
173    }
174
175    /// Returns the execution that recorded this traversal.
176    #[must_use]
177    pub const fn job_execution_id(&self) -> JobExecutionId {
178        self.job_execution_id
179    }
180
181    /// Returns the execution-local ordering.
182    #[must_use]
183    pub const fn sequence(&self) -> FlowDecisionSequence {
184        self.sequence
185    }
186
187    /// Borrows the source logical node.
188    #[must_use]
189    pub const fn source_node_id(&self) -> &NodeId {
190        &self.source_node_id
191    }
192
193    /// Returns the step attempt whose durable result was observed, if any.
194    #[must_use]
195    pub const fn source_step_execution_id(&self) -> Option<StepExecutionId> {
196        self.source_step_execution_id
197    }
198
199    /// Returns why this transition was selected.
200    #[must_use]
201    pub const fn kind(&self) -> FlowTransitionKind {
202        self.kind
203    }
204
205    /// Borrows the bounded observed outcome.
206    #[must_use]
207    pub const fn observed_outcome(&self) -> &ExitCode {
208        &self.observed_outcome
209    }
210
211    /// Borrows the selected node or terminal.
212    #[must_use]
213    pub const fn target(&self) -> &FlowTarget {
214        &self.target
215    }
216
217    /// Returns the exact plan fingerprint under which the choice was made.
218    #[must_use]
219    pub const fn plan_fingerprint(&self) -> &[u8; 32] {
220        &self.plan_fingerprint
221    }
222
223    /// Returns the value-redacted durable-input digest.
224    #[must_use]
225    pub const fn input_digest(&self) -> &[u8; 32] {
226        &self.input_digest
227    }
228
229    /// Returns the prior committed decision reused by restart, if any.
230    #[must_use]
231    pub const fn reused_decision_id(&self) -> Option<FlowDecisionId> {
232        self.reused_decision_id
233    }
234
235    /// Returns the injected facade-clock timestamp.
236    #[must_use]
237    pub const fn decided_at(&self) -> SystemTime {
238        self.decided_at
239    }
240}
241
242/// A validated transition awaiting repository allocation and commit.
243#[derive(Clone, Debug, Eq, PartialEq)]
244pub struct FlowDecisionRequest {
245    job_execution_id: JobExecutionId,
246    sequence: FlowDecisionSequence,
247    source_node_id: NodeId,
248    source_step_execution_id: Option<StepExecutionId>,
249    kind: FlowTransitionKind,
250    observed_outcome: ExitCode,
251    target: FlowTarget,
252    plan_fingerprint: [u8; 32],
253    input_digest: [u8; 32],
254    reused_decision_id: Option<FlowDecisionId>,
255    decided_at: SystemTime,
256}
257
258impl FlowDecisionRequest {
259    /// Builds one validated transition awaiting allocation and commit.
260    #[allow(clippy::too_many_arguments)]
261    #[doc(hidden)]
262    #[must_use]
263    pub const fn new(
264        job_execution_id: JobExecutionId,
265        sequence: FlowDecisionSequence,
266        source_node_id: NodeId,
267        source_step_execution_id: Option<StepExecutionId>,
268        kind: FlowTransitionKind,
269        observed_outcome: ExitCode,
270        target: FlowTarget,
271        plan_fingerprint: [u8; 32],
272        input_digest: [u8; 32],
273        reused_decision_id: Option<FlowDecisionId>,
274        decided_at: SystemTime,
275    ) -> Self {
276        Self {
277            job_execution_id,
278            sequence,
279            source_node_id,
280            source_step_execution_id,
281            kind,
282            observed_outcome,
283            target,
284            plan_fingerprint,
285            input_digest,
286            reused_decision_id,
287            decided_at,
288        }
289    }
290
291    /// Returns the execution that will own the append.
292    #[must_use]
293    pub const fn job_execution_id(&self) -> JobExecutionId {
294        self.job_execution_id
295    }
296    /// Returns the expected execution-local append sequence.
297    #[must_use]
298    pub const fn sequence(&self) -> FlowDecisionSequence {
299        self.sequence
300    }
301    /// Borrows the selected transition's source node.
302    #[must_use]
303    pub const fn source_node_id(&self) -> &NodeId {
304        &self.source_node_id
305    }
306    /// Returns the durable source step, when the source is a step.
307    #[must_use]
308    pub const fn source_step_execution_id(&self) -> Option<StepExecutionId> {
309        self.source_step_execution_id
310    }
311    /// Returns why the runtime selected this transition.
312    #[must_use]
313    pub const fn kind(&self) -> FlowTransitionKind {
314        self.kind
315    }
316    /// Borrows the bounded outcome used for selection.
317    #[must_use]
318    pub const fn observed_outcome(&self) -> &ExitCode {
319        &self.observed_outcome
320    }
321    /// Borrows the selected node or terminal.
322    #[must_use]
323    pub const fn target(&self) -> &FlowTarget {
324        &self.target
325    }
326    /// Returns the exact persisted plan fingerprint.
327    #[must_use]
328    pub const fn plan_fingerprint(&self) -> &[u8; 32] {
329        &self.plan_fingerprint
330    }
331    /// Returns the value-redacted durable-input digest.
332    #[must_use]
333    pub const fn input_digest(&self) -> &[u8; 32] {
334        &self.input_digest
335    }
336    /// Returns the exact prior decision reused by restart, when present.
337    #[must_use]
338    pub const fn reused_decision_id(&self) -> Option<FlowDecisionId> {
339        self.reused_decision_id
340    }
341    /// Returns the injected facade-clock decision time.
342    #[must_use]
343    pub const fn decided_at(&self) -> SystemTime {
344        self.decided_at
345    }
346
347    /// Materializes the immutable record after an adapter allocates its ID.
348    ///
349    /// Repository implementations should call this only after validating the
350    /// request against the persisted plan and committing its append rules.
351    #[must_use]
352    pub fn materialize(&self, id: FlowDecisionId) -> FlowDecision {
353        FlowDecision::new(
354            id,
355            self.job_execution_id,
356            self.sequence,
357            self.source_node_id.clone(),
358            self.source_step_execution_id,
359            self.kind,
360            self.observed_outcome.clone(),
361            self.target.clone(),
362            self.plan_fingerprint,
363            self.input_digest,
364            self.reused_decision_id,
365            self.decided_at,
366        )
367    }
368}
369
370/// Latest durable attempt for one logical step, used to reconstruct restart.
371#[derive(Clone, Debug, Eq, PartialEq)]
372pub struct FlowStepState {
373    node_id: NodeId,
374    execution: StepExecution,
375    context: Option<ExecutionContext>,
376}
377
378impl FlowStepState {
379    /// Constructs adapter-supplied latest logical-step state.
380    ///
381    /// Repository implementations must verify that `execution` belongs to the
382    /// requested job instance and `node_id` before returning this value.
383    #[must_use]
384    pub const fn new(
385        node_id: NodeId,
386        execution: StepExecution,
387        context: Option<ExecutionContext>,
388    ) -> Self {
389        Self {
390            node_id,
391            execution,
392            context,
393        }
394    }
395
396    /// Borrows the stable step logical identifier.
397    #[must_use]
398    pub const fn node_id(&self) -> &NodeId {
399        &self.node_id
400    }
401
402    /// Borrows the latest step attempt.
403    #[must_use]
404    pub const fn execution(&self) -> &StepExecution {
405        &self.execution
406    }
407
408    /// Borrows committed step context when the adapter exposes it.
409    #[must_use]
410    pub const fn context(&self) -> Option<&ExecutionContext> {
411        self.context.as_ref()
412    }
413}
oxide_batch_repository/flow.rs

oxide_batch_repository/
flow.rs
