oxide_batch_core/
flow.rs

1//! Durable flow identities, transition targets, and start controls.
2//!
3//! Every value here is written to metadata, read back on restart, or
4//! hashed into a definition fingerprint. The graph that arranges them and
5//! the compiler that validates it live above this crate.
6
7use std::num::NonZeroU32;
8
9use crate::{DefinitionError, DefinitionTokenKind, definition_token, validate_token};
10
11/// The maximum number of durable local partitions in one partitioned step.
12pub const MAX_PARTITIONS: u16 = 1_024;
13definition_token!(
14    NodeId,
15    DefinitionTokenKind::Node,
16    "A stable logical identifier for one flow-graph node.
17
18Logical identity survives display-name changes. Runtime and database
19identifiers are never node identifiers."
20);
21/// The maximum number of step executions one logical step may start.
22///
23/// The default is `u32::MAX`: an effectively unrestricted step that remains a
24/// finite typed value rather than an absent bound.
25#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
26pub struct StartLimit(NonZeroU32);
27
28impl StartLimit {
29    /// The unrestricted default.
30    pub const UNRESTRICTED: Self = Self(NonZeroU32::MAX);
31
32    /// Constructs a nonzero start limit.
33    ///
34    /// # Errors
35    ///
36    /// Returns [`DefinitionError::ZeroStartLimit`] for zero, because a step
37    /// that can never start is a definition mistake rather than a policy.
38    pub fn new(value: u32) -> Result<Self, DefinitionError> {
39        NonZeroU32::new(value)
40            .map(Self)
41            .ok_or(DefinitionError::ZeroStartLimit)
42    }
43
44    /// Returns the limit.
45    #[must_use]
46    pub const fn get(self) -> u32 {
47        self.0.get()
48    }
49}
50
51impl Default for StartLimit {
52    fn default() -> Self {
53        Self::UNRESTRICTED
54    }
55}
56
57/// Restart-relevant start controls for one logical step.
58#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
59pub struct StartControls {
60    start_limit: StartLimit,
61    allow_start_if_complete: bool,
62}
63
64impl StartControls {
65    /// Constructs explicit start controls.
66    #[must_use]
67    pub const fn new(start_limit: StartLimit, allow_start_if_complete: bool) -> Self {
68        Self {
69            start_limit,
70            allow_start_if_complete,
71        }
72    }
73
74    /// Returns the maximum number of starts for one job instance.
75    #[must_use]
76    pub const fn start_limit(&self) -> StartLimit {
77        self.start_limit
78    }
79
80    /// Returns whether a restart path reruns an already completed step.
81    #[must_use]
82    pub const fn allow_start_if_complete(&self) -> bool {
83        self.allow_start_if_complete
84    }
85}
86
87/// A node that ends the job without starting further work.
88#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
89#[non_exhaustive]
90pub enum TerminalKind {
91    /// The job completes.
92    Complete,
93    /// The job fails.
94    Fail,
95    /// The job stops and remains restartable.
96    Stop,
97}
98
99impl TerminalKind {
100    /// Returns the stable manifest and telemetry name.
101    #[must_use]
102    pub const fn as_str(self) -> &'static str {
103        match self {
104            Self::Complete => "complete",
105            Self::Fail => "fail",
106            Self::Stop => "stop",
107        }
108    }
109}
110
111/// The destination one transition selects.
112#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
113pub enum FlowTarget {
114    /// Another graph node starts next.
115    Node(NodeId),
116    /// The job ends at a terminal.
117    Terminal(TerminalKind),
118}
119
120impl FlowTarget {
121    /// Returns the deterministic ordering key for canonical output.
122    #[must_use]
123    pub fn sort_key(&self) -> (u8, &str) {
124        match self {
125            Self::Node(id) => (0, id.as_str()),
126            Self::Terminal(kind) => (1, kind.as_str()),
127        }
128    }
129}
oxide_batch_core/flow.rs

oxide_batch_core/
flow.rs
