oxide_batch_plan/
lib.rs

1//! Internal implementation crate for `OxideBatch`.
2//!
3//! **This crate is implementation detail. Use
4//! [`oxide-batch`](https://crates.io/crates/oxide-batch) instead.**
5//!
6//! It exists on crates.io only because the published `oxide-batch` facade
7//! depends on it. Its API carries no stability promise: items may be added,
8//! changed, or removed in any release, without a deprecation period. It has no
9//! supported-configuration matrix, no compatibility ledger row, and no
10//! independent release cadence.
11//!
12//! Everything here that `OxideBatch` supports is re-exported from `oxide-batch`
13//! under a stable path.
14//!
15//! The crate holds immutable flow graphs and the compiled execution plans they
16//! lower into. An application declares a [`FlowGraph`] of step and decision
17//! nodes joined by exit-pattern transitions, then compiles it into an immutable
18//! [`CompiledExecutionPlan`]. Compilation normalizes the graph, rejects every
19//! structural error the accepted basic-flow contract names, and produces the
20//! canonical manifest whose SHA-256 digest is the definition fingerprint.
21//!
22//! The M3 graph remains acyclic. M4 adds only the accepted bounded split and
23//! local-partition forms; nested splits, decisions inside branches, dynamic
24//! partitioning, and remote execution remain outside this crate's contract.
25//! Existing one-step `TaskletJob` and `ChunkJob` definitions lower into a
26//! compatibility plan that retains their original format-1 manifest bytes and
27//! fingerprint.
28//!
29//! The crate depends on no async runtime, database driver, command-line
30//! framework, telemetry SDK, broker client, or web framework, and on no
31//! `OxideBatch` crate other than `oxide-batch-core`. The flow engine that
32//! executes a compiled plan, the metadata ports that persist its decisions,
33//! and the runtime live above this crate.
34//!
35//! # Items marked `#[doc(hidden)]`
36//!
37//! Some items exist as `#[doc(hidden)] pub` only because the facade's own code
38//! was split from these types by the extraction boundary: private access that
39//! one crate resolved by module privacy now crosses a crate boundary. They are
40//! not part of any surface, supported or otherwise, and the facade never
41//! re-exports one under its own name. The staged crate-extraction contract
42//! records each one.
43
44#![forbid(unsafe_code)]
45
46use std::collections::{BTreeMap, BTreeSet};
47use std::error::Error;
48use std::fmt;
49use std::num::NonZeroU32;
50
51use serde_json::{Value, json};
52
53use oxide_batch_core::{
54    ChunkComponentRevisions, ChunkSize, ComponentRevision, DefinitionError, DefinitionIdentity,
55    DefinitionRevision, DefinitionTokenKind, ExitCode, FaultPolicy, FlowTarget, InFlightPolicy,
56    JobName, MAX_NODES, MAX_PARTITIONS, MAX_TRANSITIONS, NodeId, StartControls, StepName,
57    TerminalKind, definition_token, validate_token,
58};
59
60/// The maximum number of transitions leaving one node.
61pub const MAX_OUTGOING_TRANSITIONS: usize = 64;
62/// The maximum length of one exit pattern in UTF-8 bytes.
63pub const MAX_PATTERN_BYTES: usize = 64;
64/// The maximum number of branches in one M4 split.
65pub const MAX_SPLIT_BRANCHES: usize = 8;
66/// The maximum number of linear steps in one split branch.
67pub const MAX_BRANCH_STEPS: usize = 8;
68/// The maximum number of concurrent local partition workers.
69pub const MAX_PARTITION_WORKERS: u8 = 64;
70
71/// The sibling behavior selected after one local child fails.
72#[derive(Clone, Copy, Debug, Default, Eq, Hash, Ord, PartialEq, PartialOrd)]
73#[non_exhaustive]
74pub enum LocalFailurePolicy {
75    /// Request cooperative cancellation of siblings, then join all children.
76    #[default]
77    CancelSiblings,
78    /// Allow siblings to reach their next boundary, then join all children.
79    DrainSiblings,
80}
81
82impl LocalFailurePolicy {
83    const fn as_str(self) -> &'static str {
84        match self {
85            Self::CancelSiblings => "cancel_siblings",
86            Self::DrainSiblings => "drain_siblings",
87        }
88    }
89}
90
91/// The finite concurrency and connection budget for one M4 split.
92#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
93pub struct SplitBudget {
94    max_parallel_branches: u8,
95    repository_pool_size: u32,
96}
97
98impl SplitBudget {
99    /// Constructs validated branch-concurrency and connection bounds.
100    ///
101    /// # Errors
102    ///
103    /// Rejects zero/over-limit concurrency or a pool that cannot supply every
104    /// active branch plus the owning parent connection.
105    pub fn new(max_parallel_branches: u8, repository_pool_size: u32) -> Result<Self, PlanError> {
106        if max_parallel_branches == 0 || usize::from(max_parallel_branches) > MAX_SPLIT_BRANCHES {
107            return Err(PlanError::InvalidParallelBranchBudget {
108                max: MAX_SPLIT_BRANCHES,
109            });
110        }
111        let required = u32::from(max_parallel_branches).saturating_add(1);
112        if repository_pool_size < required {
113            return Err(PlanError::InsufficientPoolCapacity {
114                required,
115                configured: repository_pool_size,
116            });
117        }
118        Ok(Self {
119            max_parallel_branches,
120            repository_pool_size,
121        })
122    }
123
124    /// Returns the maximum concurrent split branches.
125    #[must_use]
126    pub const fn max_parallel_branches(self) -> u8 {
127        self.max_parallel_branches
128    }
129
130    /// Returns the validated repository pool size.
131    #[must_use]
132    pub const fn repository_pool_size(self) -> u32 {
133        self.repository_pool_size
134    }
135}
136
137impl Default for SplitBudget {
138    fn default() -> Self {
139        Self {
140            max_parallel_branches: 1,
141            repository_pool_size: 2,
142        }
143    }
144}
145
146/// The finite worker and connection budget for one M4 partition manager.
147#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
148pub struct PartitionBudget {
149    max_partition_workers: u8,
150    repository_pool_size: u32,
151}
152
153impl PartitionBudget {
154    /// Constructs validated worker-concurrency and connection bounds.
155    ///
156    /// # Errors
157    ///
158    /// Rejects zero/over-limit concurrency or a pool that cannot supply every
159    /// active worker plus the owning parent connection.
160    pub fn new(max_partition_workers: u8, repository_pool_size: u32) -> Result<Self, PlanError> {
161        if !(1..=MAX_PARTITION_WORKERS).contains(&max_partition_workers) {
162            return Err(PlanError::InvalidPartitionWorkerBudget {
163                max: MAX_PARTITION_WORKERS,
164            });
165        }
166        let required = u32::from(max_partition_workers).saturating_add(1);
167        if repository_pool_size < required {
168            return Err(PlanError::InsufficientPoolCapacity {
169                required,
170                configured: repository_pool_size,
171            });
172        }
173        Ok(Self {
174            max_partition_workers,
175            repository_pool_size,
176        })
177    }
178
179    /// Returns the maximum concurrent partition workers.
180    #[must_use]
181    pub const fn max_partition_workers(self) -> u8 {
182        self.max_partition_workers
183    }
184
185    /// Returns the validated repository pool size.
186    #[must_use]
187    pub const fn repository_pool_size(self) -> u32 {
188        self.repository_pool_size
189    }
190}
191
192impl Default for PartitionBudget {
193    fn default() -> Self {
194        Self {
195            max_partition_workers: 4,
196            repository_pool_size: 5,
197        }
198    }
199}
200
201definition_token!(
202    DeciderRevision,
203    DefinitionTokenKind::Decider,
204    "An application-owned revision token for one deterministic decider."
205);
206
207/// The version of the durable input contract one decider reads.
208#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
209pub struct DecisionInputVersion(NonZeroU32);
210
211impl DecisionInputVersion {
212    /// Constructs a nonzero durable input-contract version.
213    ///
214    /// # Errors
215    ///
216    /// Returns [`PlanError::ZeroDecisionInputVersion`] for zero.
217    pub fn new(value: u32) -> Result<Self, PlanError> {
218        NonZeroU32::new(value)
219            .map(Self)
220            .ok_or(PlanError::ZeroDecisionInputVersion)
221    }
222
223    /// Returns the version.
224    #[must_use]
225    pub const fn get(self) -> u32 {
226        self.0.get()
227    }
228}
229
230/// A bounded exit-outcome pattern used to select one transition.
231///
232/// A pattern contains literal characters plus `*` for zero or more characters
233/// and `?` for exactly one character. It matches the bounded
234/// [`ExitCode`], never [`BatchStatus`](oxide_batch_core::BatchStatus).
235///
236/// The worked example lives in the `oxide-batch` crate documentation, so that
237/// it keeps demonstrating the supported import path.
238#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
239pub struct ExitPattern(String);
240
241impl ExitPattern {
242    /// Validates and constructs an exit pattern.
243    ///
244    /// # Errors
245    ///
246    /// Returns [`PlanError::InvalidPattern`] for an empty pattern, a pattern
247    /// longer than 64 UTF-8 bytes, surrounding whitespace, or a control
248    /// character.
249    pub fn new(value: impl Into<String>) -> Result<Self, PlanError> {
250        let value = value.into();
251        if value.is_empty()
252            || value.len() > MAX_PATTERN_BYTES
253            || value.trim() != value
254            || value.chars().any(char::is_control)
255        {
256            return Err(PlanError::InvalidPattern {
257                max_bytes: MAX_PATTERN_BYTES,
258            });
259        }
260        Ok(Self(value))
261    }
262
263    /// Borrows the validated pattern.
264    #[must_use]
265    pub fn as_str(&self) -> &str {
266        &self.0
267    }
268
269    /// Returns whether this pattern matches one exit code.
270    #[must_use]
271    pub fn matches(&self, code: &ExitCode) -> bool {
272        let pattern: Vec<char> = self.0.chars().collect();
273        let value: Vec<char> = code.as_str().chars().collect();
274        matches_from(&pattern, &value)
275    }
276
277    /// Returns the computed specificity used to order transitions.
278    ///
279    /// A greater specificity is evaluated first.
280    #[must_use]
281    pub fn specificity(&self) -> PatternSpecificity {
282        let wildcards = self
283            .0
284            .chars()
285            .filter(|character| matches!(character, '*' | '?'))
286            .count();
287        let literals = self.0.chars().count() - wildcards;
288        PatternSpecificity {
289            literals,
290            wildcards,
291            bytes: self.0.len(),
292        }
293    }
294
295    /// Returns whether some exit code exists that both patterns match.
296    #[must_use]
297    pub fn intersects(&self, other: &Self) -> bool {
298        let left: Vec<char> = self.0.chars().collect();
299        let right: Vec<char> = other.0.chars().collect();
300        let mut memo = vec![None; (left.len() + 1) * (right.len() + 1)];
301        intersects_from(&left, &right, 0, 0, &mut memo)
302    }
303}
304
305impl fmt::Display for ExitPattern {
306    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
307        formatter.write_str(&self.0)
308    }
309}
310
311fn matches_from(pattern: &[char], value: &[char]) -> bool {
312    let mut pattern_index = 0_usize;
313    let mut value_index = 0_usize;
314    let mut star: Option<(usize, usize)> = None;
315    while value_index < value.len() {
316        match pattern.get(pattern_index) {
317            Some('*') => {
318                star = Some((pattern_index, value_index));
319                pattern_index += 1;
320            }
321            Some('?') => {
322                pattern_index += 1;
323                value_index += 1;
324            }
325            Some(literal) if *literal == value[value_index] => {
326                pattern_index += 1;
327                value_index += 1;
328            }
329            _ => match star {
330                Some((star_index, resume)) => {
331                    pattern_index = star_index + 1;
332                    value_index = resume + 1;
333                    star = Some((star_index, resume + 1));
334                }
335                None => return false,
336            },
337        }
338    }
339    pattern[pattern_index..]
340        .iter()
341        .all(|character| *character == '*')
342}
343
344fn intersects_from(
345    left: &[char],
346    right: &[char],
347    left_index: usize,
348    right_index: usize,
349    memo: &mut [Option<bool>],
350) -> bool {
351    let key = left_index * (right.len() + 1) + right_index;
352    if let Some(cached) = memo[key] {
353        return cached;
354    }
355    let answer = match (left.get(left_index), right.get(right_index)) {
356        (None, None) => true,
357        (None, Some(_)) => right[right_index..].iter().all(|value| *value == '*'),
358        (Some(_), None) => left[left_index..].iter().all(|value| *value == '*'),
359        (Some('*'), _) => {
360            intersects_from(left, right, left_index + 1, right_index, memo)
361                || intersects_from(left, right, left_index, right_index + 1, memo)
362        }
363        (_, Some('*')) => {
364            intersects_from(left, right, left_index, right_index + 1, memo)
365                || intersects_from(left, right, left_index + 1, right_index, memo)
366        }
367        (Some(left_character), Some(right_character)) => {
368            (*left_character == '?' || *right_character == '?' || left_character == right_character)
369                && intersects_from(left, right, left_index + 1, right_index + 1, memo)
370        }
371    };
372    memo[key] = Some(answer);
373    answer
374}
375
376/// The computed specificity of one exit pattern.
377///
378/// Ordering compares more literal characters first, then fewer wildcards, then
379/// a longer UTF-8 byte length. A greater value is evaluated first.
380#[derive(Clone, Copy, Debug, Eq, PartialEq)]
381pub struct PatternSpecificity {
382    literals: usize,
383    wildcards: usize,
384    bytes: usize,
385}
386
387impl PatternSpecificity {
388    /// Returns the number of literal characters.
389    #[must_use]
390    pub const fn literals(self) -> usize {
391        self.literals
392    }
393
394    /// Returns the number of `*` and `?` characters.
395    #[must_use]
396    pub const fn wildcards(self) -> usize {
397        self.wildcards
398    }
399
400    /// Returns the pattern length in UTF-8 bytes.
401    #[must_use]
402    pub const fn bytes(self) -> usize {
403        self.bytes
404    }
405}
406
407impl Ord for PatternSpecificity {
408    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
409        self.literals
410            .cmp(&other.literals)
411            .then_with(|| other.wildcards.cmp(&self.wildcards))
412            .then_with(|| self.bytes.cmp(&other.bytes))
413    }
414}
415
416impl PartialOrd for PatternSpecificity {
417    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
418        Some(self.cmp(other))
419    }
420}
421
422/// The executable kind and restart-relevant declaration of one step node.
423#[derive(Clone, Debug, Eq, PartialEq)]
424#[non_exhaustive]
425pub enum StepComponents {
426    /// A single-invocation tasklet body.
427    Tasklet(ComponentRevision),
428    /// A restartable reader, processor, and writer pipeline.
429    Chunk {
430        /// The committed chunk size.
431        size: ChunkSize,
432        /// The restart-relevant component revisions and state schemas.
433        revisions: Box<ChunkComponentRevisions>,
434    },
435}
436
437impl StepComponents {
438    fn kind_name(&self) -> &'static str {
439        match self {
440            Self::Tasklet(_) => "tasklet",
441            Self::Chunk { .. } => "chunk",
442        }
443    }
444
445    fn manifest_value(&self) -> Value {
446        match self {
447            Self::Tasklet(revision) => json!({
448                "component": revision.as_str(),
449                "delivery_mode": "best_effort",
450                "transaction_boundary": "tasklet_completion"
451            }),
452            Self::Chunk { size, revisions } => {
453                let mut chunk = chunk_declaration_manifest(revisions);
454                if let Some(members) = chunk.as_object_mut() {
455                    members.insert("size".to_owned(), json!(size.get()));
456                    members.insert(
457                        "transaction_boundary".to_owned(),
458                        Value::String("chunk".to_owned()),
459                    );
460                }
461                chunk
462            }
463        }
464    }
465}
466
467/// One executable node of a compiled plan.
468#[derive(Clone, Debug, Eq, PartialEq)]
469pub struct StepNode {
470    id: NodeId,
471    step_name: StepName,
472    components: StepComponents,
473    start: StartControls,
474    fault: Option<FaultPolicy>,
475    listeners: Vec<ComponentRevision>,
476}
477
478impl StepNode {
479    /// Declares one step node.
480    #[must_use]
481    pub fn new(id: NodeId, step_name: StepName, components: StepComponents) -> Self {
482        Self {
483            id,
484            step_name,
485            components,
486            start: StartControls::default(),
487            fault: None,
488            listeners: Vec::new(),
489        }
490    }
491
492    /// Declares explicit start controls.
493    #[must_use]
494    pub const fn with_start_controls(mut self, start: StartControls) -> Self {
495        self.start = start;
496        self
497    }
498
499    /// Declares the fault policy this step's fingerprint captures.
500    #[must_use]
501    pub fn with_fault_policy(mut self, policy: FaultPolicy) -> Self {
502        self.fault = Some(policy);
503        self
504    }
505
506    /// Declares one authoritative listener revision in registration order.
507    #[must_use]
508    pub fn with_listener_revision(mut self, revision: ComponentRevision) -> Self {
509        self.listeners.push(revision);
510        self
511    }
512
513    /// Borrows the stable node identifier.
514    #[must_use]
515    pub const fn id(&self) -> &NodeId {
516        &self.id
517    }
518
519    /// Borrows the durable step name.
520    #[must_use]
521    pub const fn step_name(&self) -> &StepName {
522        &self.step_name
523    }
524
525    /// Borrows the executable component declaration.
526    #[must_use]
527    pub const fn components(&self) -> &StepComponents {
528        &self.components
529    }
530
531    /// Returns the restart-relevant start controls.
532    #[must_use]
533    pub const fn start_controls(&self) -> StartControls {
534        self.start
535    }
536
537    /// Borrows the declared fault policy.
538    #[must_use]
539    pub const fn fault_policy(&self) -> Option<&FaultPolicy> {
540        self.fault.as_ref()
541    }
542
543    /// Borrows the authoritative listener revisions in registration order.
544    #[must_use]
545    pub fn listener_revisions(&self) -> &[ComponentRevision] {
546        &self.listeners
547    }
548
549    fn manifest_value(&self) -> Value {
550        json!({
551            "id": self.id.as_str(),
552            "kind": "step",
553            "listeners": self
554                .listeners
555                .iter()
556                .map(|revision| Value::String(revision.as_str().to_owned()))
557                .collect::<Vec<_>>(),
558            "policy": self.fault.as_ref().map_or(Value::Null, fault_manifest_value),
559            "start": start_controls_manifest(self.start),
560            "step": {
561                "declaration": self.components.manifest_value(),
562                "kind": self.components.kind_name(),
563                "name": self.step_name.as_str()
564            }
565        })
566    }
567}
568
569/// One deterministic decision node of a compiled plan.
570///
571/// M3 compiles and fingerprints decision nodes; executing them is owned by the
572/// durable-flow workstream.
573#[derive(Clone, Debug, Eq, PartialEq)]
574pub struct DecisionNode {
575    id: NodeId,
576    revision: DeciderRevision,
577    input_version: DecisionInputVersion,
578}
579
580impl DecisionNode {
581    /// Declares one decision node.
582    #[must_use]
583    pub const fn new(
584        id: NodeId,
585        revision: DeciderRevision,
586        input_version: DecisionInputVersion,
587    ) -> Self {
588        Self {
589            id,
590            revision,
591            input_version,
592        }
593    }
594
595    /// Borrows the stable node identifier.
596    #[must_use]
597    pub const fn id(&self) -> &NodeId {
598        &self.id
599    }
600
601    /// Borrows the application-owned decider revision.
602    #[must_use]
603    pub const fn revision(&self) -> &DeciderRevision {
604        &self.revision
605    }
606
607    /// Returns the durable input-contract version.
608    #[must_use]
609    pub const fn input_version(&self) -> DecisionInputVersion {
610        self.input_version
611    }
612
613    fn manifest_value(&self) -> Value {
614        json!({
615            "decision": {
616                "input_version": self.input_version.get(),
617                "revision": self.revision.as_str()
618            },
619            "id": self.id.as_str(),
620            "kind": "decision"
621        })
622    }
623}
624
625/// One declared linear branch of an M4 split.
626#[derive(Clone, Debug, Eq, PartialEq)]
627pub struct SplitBranch {
628    steps: Vec<StepNode>,
629}
630
631impl SplitBranch {
632    /// Declares a branch from its ordered tasklet or chunk steps.
633    ///
634    /// Cardinality and identifier uniqueness are checked by
635    /// [`FlowGraph::compile`], so builders can assemble a complete diagnostic
636    /// instead of panicking while under construction.
637    #[must_use]
638    pub fn new(steps: Vec<StepNode>) -> Self {
639        Self { steps }
640    }
641
642    /// Borrows the branch steps in declared execution order.
643    #[must_use]
644    pub fn steps(&self) -> &[StepNode] {
645        &self.steps
646    }
647
648    /// Borrows the branch identity, which is its first logical step ID.
649    #[must_use]
650    pub fn id(&self) -> Option<&NodeId> {
651        self.steps.first().map(StepNode::id)
652    }
653
654    fn manifest_value(&self) -> Value {
655        Value::Array(self.steps.iter().map(StepNode::manifest_value).collect())
656    }
657}
658
659/// A bounded M4 split whose branches converge at exactly one join node.
660#[derive(Clone, Debug, Eq, PartialEq)]
661pub struct SplitNode {
662    id: NodeId,
663    branches: Vec<SplitBranch>,
664    join: NodeId,
665    budget: SplitBudget,
666    failure_policy: LocalFailurePolicy,
667}
668
669impl SplitNode {
670    /// Declares a split, its ordered branches, and its unique join.
671    #[must_use]
672    pub fn new(id: NodeId, branches: Vec<SplitBranch>, join: NodeId, budget: SplitBudget) -> Self {
673        Self {
674            id,
675            branches,
676            join,
677            budget,
678            failure_policy: LocalFailurePolicy::default(),
679        }
680    }
681
682    /// Selects sibling failure behavior.
683    #[must_use]
684    pub const fn with_failure_policy(mut self, failure_policy: LocalFailurePolicy) -> Self {
685        self.failure_policy = failure_policy;
686        self
687    }
688
689    /// Borrows the stable split identifier.
690    #[must_use]
691    pub const fn id(&self) -> &NodeId {
692        &self.id
693    }
694
695    /// Borrows branches in deterministic aggregation order.
696    #[must_use]
697    pub fn branches(&self) -> &[SplitBranch] {
698        &self.branches
699    }
700
701    /// Borrows the split's unique join identifier.
702    #[must_use]
703    pub const fn join(&self) -> &NodeId {
704        &self.join
705    }
706
707    /// Returns the finite local resource budget.
708    #[must_use]
709    pub const fn budget(&self) -> SplitBudget {
710        self.budget
711    }
712
713    /// Returns sibling failure behavior.
714    #[must_use]
715    pub const fn failure_policy(&self) -> LocalFailurePolicy {
716        self.failure_policy
717    }
718
719    /// Projects the restart-relevant split declaration.
720    ///
721    /// The budget is a throughput bound rather than a durable-meaning value, so
722    /// [ADR-0009](https://github.com/luceat-lux-vestra/oxide-batch/blob/main/docs/architecture/decisions/0009-definition-fingerprint-input-set.md)
723    /// excludes it. Branch membership and order select assignment and remain.
724    fn manifest_value(&self) -> Value {
725        json!({
726            "branches": self.branches.iter().map(SplitBranch::manifest_value).collect::<Vec<_>>(),
727            "failure_policy": self.failure_policy.as_str(),
728            "id": self.id.as_str(),
729            "join": self.join.as_str(),
730            "kind": "split"
731        })
732    }
733}
734
735/// The structural join owned by one M4 split.
736#[derive(Clone, Debug, Eq, PartialEq)]
737pub struct JoinNode {
738    id: NodeId,
739}
740
741impl JoinNode {
742    /// Declares a structural join.
743    #[must_use]
744    pub const fn new(id: NodeId) -> Self {
745        Self { id }
746    }
747
748    /// Borrows the stable join identifier.
749    #[must_use]
750    pub const fn id(&self) -> &NodeId {
751        &self.id
752    }
753
754    fn manifest_value(&self) -> Value {
755        json!({
756            "id": self.id.as_str(),
757            "kind": "join"
758        })
759    }
760}
761
762/// A finite durable partition count for one M4 partitioned step.
763#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
764pub struct PartitionCount(u16);
765
766impl PartitionCount {
767    /// Constructs a count in the accepted `1..=1024` range.
768    ///
769    /// # Errors
770    ///
771    /// Returns [`PlanError::InvalidPartitionCount`] outside that range.
772    pub fn new(value: u16) -> Result<Self, PlanError> {
773        if value == 0 || value > MAX_PARTITIONS {
774            return Err(PlanError::InvalidPartitionCount {
775                max: MAX_PARTITIONS,
776            });
777        }
778        Ok(Self(value))
779    }
780
781    /// Returns the declared partition count.
782    #[must_use]
783    pub const fn get(self) -> u16 {
784        self.0
785    }
786}
787
788/// A bounded local partition manager and its ordinary worker-step definition.
789#[derive(Clone, Debug, Eq, PartialEq)]
790pub struct PartitionedStepNode {
791    id: NodeId,
792    step_name: StepName,
793    worker: StepNode,
794    partitioner: ComponentRevision,
795    aggregation: ComponentRevision,
796    partitions: PartitionCount,
797    budget: PartitionBudget,
798    failure_policy: LocalFailurePolicy,
799    start: StartControls,
800}
801
802impl PartitionedStepNode {
803    /// Declares the complete restart-relevant local partition shape.
804    #[allow(clippy::too_many_arguments)]
805    #[must_use]
806    pub fn new(
807        id: NodeId,
808        step_name: StepName,
809        worker: StepNode,
810        partitioner: ComponentRevision,
811        aggregation: ComponentRevision,
812        partitions: PartitionCount,
813        budget: PartitionBudget,
814    ) -> Self {
815        Self {
816            id,
817            step_name,
818            worker,
819            partitioner,
820            aggregation,
821            partitions,
822            budget,
823            failure_policy: LocalFailurePolicy::default(),
824            start: StartControls::default(),
825        }
826    }
827
828    /// Selects sibling failure behavior.
829    #[must_use]
830    pub const fn with_failure_policy(mut self, failure_policy: LocalFailurePolicy) -> Self {
831        self.failure_policy = failure_policy;
832        self
833    }
834
835    /// Declares the partition manager's start controls.
836    #[must_use]
837    pub const fn with_start_controls(mut self, start: StartControls) -> Self {
838        self.start = start;
839        self
840    }
841
842    /// Borrows the stable manager node identifier.
843    #[must_use]
844    pub const fn id(&self) -> &NodeId {
845        &self.id
846    }
847
848    /// Borrows the manager's durable step name.
849    #[must_use]
850    pub const fn step_name(&self) -> &StepName {
851        &self.step_name
852    }
853
854    /// Borrows the ordinary tasklet or chunk worker declaration.
855    #[must_use]
856    pub const fn worker(&self) -> &StepNode {
857        &self.worker
858    }
859
860    /// Borrows the deterministic partitioner revision.
861    #[must_use]
862    pub const fn partitioner(&self) -> &ComponentRevision {
863        &self.partitioner
864    }
865
866    /// Borrows the deterministic aggregation revision.
867    #[must_use]
868    pub const fn aggregation(&self) -> &ComponentRevision {
869        &self.aggregation
870    }
871
872    /// Returns the finite partition count.
873    #[must_use]
874    pub const fn partition_count(&self) -> PartitionCount {
875        self.partitions
876    }
877
878    /// Returns the finite local resource budget.
879    #[must_use]
880    pub const fn budget(&self) -> PartitionBudget {
881        self.budget
882    }
883
884    /// Returns sibling failure behavior.
885    #[must_use]
886    pub const fn failure_policy(&self) -> LocalFailurePolicy {
887        self.failure_policy
888    }
889
890    /// Returns the partition manager's start controls.
891    #[must_use]
892    pub const fn start_controls(&self) -> StartControls {
893        self.start
894    }
895
896    /// Projects the restart-relevant partition declaration.
897    ///
898    /// The partition count selects durable assignment and the partitioner and
899    /// aggregation revisions decide how that assignment and its results are
900    /// interpreted, so all three remain. The worker and connection budget is a
901    /// throughput bound that [ADR-0009](https://github.com/luceat-lux-vestra/oxide-batch/blob/main/docs/architecture/decisions/0009-definition-fingerprint-input-set.md)
902    /// excludes.
903    fn manifest_value(&self) -> Value {
904        json!({
905            "aggregation": self.aggregation.as_str(),
906            "failure_policy": self.failure_policy.as_str(),
907            "id": self.id.as_str(),
908            "kind": "partitioned_step",
909            "partition_count": self.partitions.get(),
910            "partitioner": self.partitioner.as_str(),
911            "start": start_controls_manifest(self.start),
912            "step_name": self.step_name.as_str(),
913            "worker": self.worker.manifest_value()
914        })
915    }
916}
917
918/// One node of a declared flow graph.
919#[derive(Clone, Debug, Eq, PartialEq)]
920#[non_exhaustive]
921pub enum FlowNode {
922    /// A tasklet or chunk step.
923    Step(Box<StepNode>),
924    /// A deterministic decision.
925    Decision(DecisionNode),
926    /// A bounded set of linear branches and its structural join.
927    Split(Box<SplitNode>),
928    /// A structural join owned by exactly one split.
929    Join(JoinNode),
930    /// A bounded durable local-partition manager.
931    PartitionedStep(Box<PartitionedStepNode>),
932}
933
934impl FlowNode {
935    /// Declares a step node.
936    #[must_use]
937    pub fn step(node: StepNode) -> Self {
938        Self::Step(Box::new(node))
939    }
940
941    /// Declares a decision node.
942    #[must_use]
943    pub const fn decision(node: DecisionNode) -> Self {
944        Self::Decision(node)
945    }
946
947    /// Declares a bounded split node.
948    #[must_use]
949    pub fn split(node: SplitNode) -> Self {
950        Self::Split(Box::new(node))
951    }
952
953    /// Declares a structural join node.
954    #[must_use]
955    pub const fn join(node: JoinNode) -> Self {
956        Self::Join(node)
957    }
958
959    /// Declares a bounded local partitioned-step node.
960    #[must_use]
961    pub fn partitioned_step(node: PartitionedStepNode) -> Self {
962        Self::PartitionedStep(Box::new(node))
963    }
964
965    /// Borrows the node's stable identifier.
966    #[must_use]
967    pub const fn id(&self) -> &NodeId {
968        match self {
969            Self::Step(node) => node.id(),
970            Self::Decision(node) => node.id(),
971            Self::Split(node) => node.id(),
972            Self::Join(node) => node.id(),
973            Self::PartitionedStep(node) => node.id(),
974        }
975    }
976
977    fn manifest_value(&self) -> Value {
978        match self {
979            Self::Step(node) => node.manifest_value(),
980            Self::Decision(node) => node.manifest_value(),
981            Self::Split(node) => node.manifest_value(),
982            Self::Join(node) => node.manifest_value(),
983            Self::PartitionedStep(node) => node.manifest_value(),
984        }
985    }
986}
987
988/// One declared transition edge.
989#[derive(Clone, Debug, Eq, PartialEq)]
990pub struct FlowTransition {
991    source: NodeId,
992    pattern: ExitPattern,
993    target: FlowTarget,
994}
995
996impl FlowTransition {
997    /// Declares one directed transition selected by an exit pattern.
998    #[must_use]
999    pub const fn new(source: NodeId, pattern: ExitPattern, target: FlowTarget) -> Self {
1000        Self {
1001            source,
1002            pattern,
1003            target,
1004        }
1005    }
1006
1007    /// Borrows the source node identifier.
1008    #[must_use]
1009    pub const fn source(&self) -> &NodeId {
1010        &self.source
1011    }
1012
1013    /// Borrows the selecting pattern.
1014    #[must_use]
1015    pub const fn pattern(&self) -> &ExitPattern {
1016        &self.pattern
1017    }
1018
1019    /// Borrows the selected target.
1020    #[must_use]
1021    pub const fn target(&self) -> &FlowTarget {
1022        &self.target
1023    }
1024
1025    fn manifest_value(&self) -> Value {
1026        json!({
1027            "pattern": self.pattern.as_str(),
1028            "source": self.source.as_str(),
1029            "target": flow_target_manifest(&self.target)
1030        })
1031    }
1032}
1033
1034/// An immutable declaration of the M3 flow subset.
1035///
1036/// The worked example lives in the `oxide-batch` crate documentation, so that
1037/// it keeps demonstrating the supported import path.
1038#[derive(Clone, Debug, Default, Eq, PartialEq)]
1039pub struct FlowGraph {
1040    entry: Option<NodeId>,
1041    nodes: Vec<FlowNode>,
1042    transitions: Vec<FlowTransition>,
1043}
1044
1045impl FlowGraph {
1046    /// Starts a graph at its entry node.
1047    #[must_use]
1048    pub fn new(entry: NodeId) -> Self {
1049        Self {
1050            entry: Some(entry),
1051            nodes: Vec::new(),
1052            transitions: Vec::new(),
1053        }
1054    }
1055
1056    /// Declares one node.
1057    #[must_use]
1058    pub fn with_node(mut self, node: FlowNode) -> Self {
1059        self.nodes.push(node);
1060        self
1061    }
1062
1063    /// Declares one explicit transition.
1064    #[must_use]
1065    pub fn with_transition(mut self, transition: FlowTransition) -> Self {
1066        self.transitions.push(transition);
1067        self
1068    }
1069
1070    /// Declares the convenience sequential edge.
1071    ///
1072    /// A sequential edge compiles to an exact `FAILED` transition leading to
1073    /// [`TerminalKind::Fail`] and a less specific `*` transition leading to
1074    /// `next`. Custom successful exit codes therefore continue, while a failed
1075    /// step ends the job.
1076    ///
1077    /// # Errors
1078    ///
1079    /// Returns [`PlanError::InvalidPattern`] only if the framework-owned
1080    /// patterns cannot be constructed.
1081    pub fn with_sequence(self, source: NodeId, next: FlowTarget) -> Result<Self, PlanError> {
1082        Ok(self
1083            .with_transition(FlowTransition::new(
1084                source.clone(),
1085                ExitPattern::new("FAILED")?,
1086                FlowTarget::Terminal(TerminalKind::Fail),
1087            ))
1088            .with_transition(FlowTransition::new(source, ExitPattern::new("*")?, next)))
1089    }
1090
1091    /// Validates and normalizes the graph into an immutable plan.
1092    ///
1093    /// # Errors
1094    ///
1095    /// Returns [`PlanError`] for a missing entry node, a duplicate logical
1096    /// identifier, an undefined source or target, a node without outgoing
1097    /// transitions, two equally specific patterns that can match one value, an
1098    /// unreachable node, a cycle, an exceeded bound, or a manifest that cannot
1099    /// be encoded within the durable limit.
1100    pub fn compile(
1101        self,
1102        job_name: &JobName,
1103        revision: DefinitionRevision,
1104    ) -> Result<CompiledExecutionPlan, PlanError> {
1105        let entry = self.entry.ok_or(PlanError::MissingEntryNode)?;
1106        if self.nodes.len() > MAX_NODES {
1107            return Err(PlanError::TooManyNodes { max: MAX_NODES });
1108        }
1109        if self.transitions.len() > MAX_TRANSITIONS {
1110            return Err(PlanError::TooManyTransitions {
1111                max: MAX_TRANSITIONS,
1112            });
1113        }
1114
1115        let mut nodes = BTreeMap::new();
1116        for node in self.nodes {
1117            if nodes.insert(node.id().clone(), node.clone()).is_some() {
1118                return Err(PlanError::DuplicateNodeId {
1119                    node: node.id().clone(),
1120                });
1121            }
1122        }
1123        if !nodes.contains_key(&entry) {
1124            return Err(PlanError::UndefinedNode {
1125                node: entry.clone(),
1126            });
1127        }
1128        let local_scale = check_local_scale_subset(&entry, &nodes)?;
1129
1130        let mut outgoing: BTreeMap<NodeId, Vec<FlowTransition>> = BTreeMap::new();
1131        for transition in self.transitions {
1132            if !nodes.contains_key(transition.source()) {
1133                return Err(PlanError::UndefinedNode {
1134                    node: transition.source().clone(),
1135                });
1136            }
1137            if let FlowTarget::Node(target) = transition.target()
1138                && !nodes.contains_key(target)
1139            {
1140                return Err(PlanError::UndefinedNode {
1141                    node: target.clone(),
1142                });
1143            }
1144            if let FlowTarget::Node(target) = transition.target()
1145                && matches!(nodes.get(target), Some(FlowNode::Join(_)))
1146            {
1147                return Err(PlanError::JoinHasExternalEntry {
1148                    join: target.clone(),
1149                });
1150            }
1151            if matches!(nodes.get(transition.source()), Some(FlowNode::Split(_))) {
1152                return Err(PlanError::SplitHasExplicitTransition {
1153                    split: transition.source().clone(),
1154                });
1155            }
1156            let edges = outgoing.entry(transition.source().clone()).or_default();
1157            if edges.len() == MAX_OUTGOING_TRANSITIONS {
1158                return Err(PlanError::TooManyOutgoingTransitions {
1159                    node: transition.source().clone(),
1160                    max: MAX_OUTGOING_TRANSITIONS,
1161                });
1162            }
1163            edges.push(transition);
1164        }
1165
1166        for (id, node) in &nodes {
1167            if matches!(node, FlowNode::Split(_)) {
1168                continue;
1169            }
1170            let edges = outgoing
1171                .get(id)
1172                .filter(|edges| !edges.is_empty())
1173                .ok_or_else(|| PlanError::MissingTransition { node: id.clone() })?;
1174            check_unambiguous(id, edges)?;
1175        }
1176
1177        let mut compiled: BTreeMap<NodeId, Vec<FlowTransition>> = outgoing;
1178        for edges in compiled.values_mut() {
1179            edges.sort_by(|left, right| {
1180                right
1181                    .pattern()
1182                    .specificity()
1183                    .cmp(&left.pattern().specificity())
1184                    .then_with(|| left.pattern().cmp(right.pattern()))
1185                    .then_with(|| left.target().sort_key().cmp(&right.target().sort_key()))
1186            });
1187        }
1188
1189        check_reachable_and_acyclic(&entry, &nodes, &compiled)?;
1190
1191        let manifest = flow_manifest(job_name, &entry, &nodes, &compiled, local_scale);
1192        let canonical = serde_json::to_vec(&manifest)
1193            .map_err(|_| PlanError::Manifest(DefinitionError::ManifestEncoding))?;
1194        let definition = DefinitionIdentity::from_flow_manifest(job_name, revision, &canonical)
1195            .map_err(PlanError::Manifest)?;
1196        Ok(CompiledExecutionPlan {
1197            definition,
1198            entry,
1199            nodes,
1200            transitions: compiled,
1201        })
1202    }
1203}
1204
1205fn check_local_scale_subset(
1206    entry: &NodeId,
1207    nodes: &BTreeMap<NodeId, FlowNode>,
1208) -> Result<bool, PlanError> {
1209    let mut embedded_ids = BTreeSet::new();
1210    let mut join_owners: BTreeMap<NodeId, NodeId> = BTreeMap::new();
1211    let mut local_scale = false;
1212    for (id, node) in nodes {
1213        match node {
1214            FlowNode::Split(split) => {
1215                local_scale = true;
1216                if id == entry {
1217                    return Err(PlanError::SplitIsEntry { split: id.clone() });
1218                }
1219                if !(2..=MAX_SPLIT_BRANCHES).contains(&split.branches().len()) {
1220                    return Err(PlanError::InvalidSplitBranchCount {
1221                        split: id.clone(),
1222                        min: 2,
1223                        max: MAX_SPLIT_BRANCHES,
1224                    });
1225                }
1226                if usize::from(split.budget().max_parallel_branches()) > split.branches().len() {
1227                    return Err(PlanError::ParallelBudgetExceedsBranches {
1228                        split: id.clone(),
1229                        branches: split.branches().len(),
1230                    });
1231                }
1232                if !matches!(nodes.get(split.join()), Some(FlowNode::Join(_))) {
1233                    return Err(PlanError::InvalidSplitJoin {
1234                        split: id.clone(),
1235                        join: split.join().clone(),
1236                    });
1237                }
1238                if let Some(first) = join_owners.insert(split.join().clone(), id.clone()) {
1239                    return Err(PlanError::JoinHasMultipleOwners {
1240                        join: split.join().clone(),
1241                        first,
1242                        second: id.clone(),
1243                    });
1244                }
1245                for branch in split.branches() {
1246                    if !(1..=MAX_BRANCH_STEPS).contains(&branch.steps().len()) {
1247                        return Err(PlanError::InvalidBranchLength {
1248                            split: id.clone(),
1249                            max: MAX_BRANCH_STEPS,
1250                        });
1251                    }
1252                    for step in branch.steps() {
1253                        if nodes.contains_key(step.id()) || !embedded_ids.insert(step.id().clone())
1254                        {
1255                            return Err(PlanError::DuplicateNodeId {
1256                                node: step.id().clone(),
1257                            });
1258                        }
1259                    }
1260                }
1261            }
1262            FlowNode::Join(_) => {
1263                local_scale = true;
1264            }
1265            FlowNode::PartitionedStep(partitioned) => {
1266                local_scale = true;
1267                let worker = partitioned.worker().id();
1268                if nodes.contains_key(worker) || !embedded_ids.insert(worker.clone()) {
1269                    return Err(PlanError::DuplicateNodeId {
1270                        node: worker.clone(),
1271                    });
1272                }
1273            }
1274            FlowNode::Step(_) | FlowNode::Decision(_) => {}
1275        }
1276    }
1277    if nodes.len().saturating_add(embedded_ids.len()) > MAX_NODES {
1278        return Err(PlanError::TooManyNodes { max: MAX_NODES });
1279    }
1280    for (id, node) in nodes {
1281        if matches!(node, FlowNode::Join(_)) && !join_owners.contains_key(id) {
1282            return Err(PlanError::OrphanJoin { join: id.clone() });
1283        }
1284    }
1285    Ok(local_scale)
1286}
1287
1288fn check_unambiguous(node: &NodeId, edges: &[FlowTransition]) -> Result<(), PlanError> {
1289    for (index, left) in edges.iter().enumerate() {
1290        for right in &edges[index + 1..] {
1291            if left.pattern().specificity() == right.pattern().specificity()
1292                && left.pattern().intersects(right.pattern())
1293            {
1294                return Err(PlanError::AmbiguousTransition {
1295                    node: node.clone(),
1296                    first: left.pattern().clone(),
1297                    second: right.pattern().clone(),
1298                });
1299            }
1300        }
1301    }
1302    Ok(())
1303}
1304
1305fn check_reachable_and_acyclic(
1306    entry: &NodeId,
1307    nodes: &BTreeMap<NodeId, FlowNode>,
1308    transitions: &BTreeMap<NodeId, Vec<FlowTransition>>,
1309) -> Result<(), PlanError> {
1310    let mut visited = BTreeSet::new();
1311    let mut on_path = BTreeSet::new();
1312    visit(entry, nodes, transitions, &mut visited, &mut on_path)?;
1313    for id in nodes.keys() {
1314        if !visited.contains(id) {
1315            return Err(PlanError::UnreachableNode { node: id.clone() });
1316        }
1317    }
1318    Ok(())
1319}
1320
1321fn visit(
1322    node: &NodeId,
1323    nodes: &BTreeMap<NodeId, FlowNode>,
1324    transitions: &BTreeMap<NodeId, Vec<FlowTransition>>,
1325    visited: &mut BTreeSet<NodeId>,
1326    on_path: &mut BTreeSet<NodeId>,
1327) -> Result<(), PlanError> {
1328    if on_path.contains(node) {
1329        return Err(PlanError::CyclicGraph { node: node.clone() });
1330    }
1331    if !visited.insert(node.clone()) {
1332        return Ok(());
1333    }
1334    on_path.insert(node.clone());
1335    if let Some(FlowNode::Split(split)) = nodes.get(node) {
1336        visit(split.join(), nodes, transitions, visited, on_path)?;
1337    }
1338    if let Some(edges) = transitions.get(node) {
1339        for edge in edges {
1340            if let FlowTarget::Node(target) = edge.target() {
1341                visit(target, nodes, transitions, visited, on_path)?;
1342            }
1343        }
1344    }
1345    on_path.remove(node);
1346    Ok(())
1347}
1348
1349/// Projects restart-relevant start controls into their manifest member.
1350///
1351/// The projection lives here rather than on the value because the canonical
1352/// manifest is this crate's contract. Keeping it here is also what keeps the
1353/// serializer out of the core's public signatures, and out of the facade that
1354/// re-exports them.
1355fn start_controls_manifest(controls: StartControls) -> Value {
1356    json!({
1357        "allow_start_if_complete": controls.allow_start_if_complete(),
1358        "start_limit": controls.start_limit().get()
1359    })
1360}
1361
1362/// Projects one transition target into its manifest member.
1363fn flow_target_manifest(target: &FlowTarget) -> Value {
1364    match target {
1365        FlowTarget::Node(id) => json!({ "node": id.as_str() }),
1366        FlowTarget::Terminal(kind) => json!({ "terminal": kind.as_str() }),
1367    }
1368}
1369
1370/// Projects the restart-relevant chunk declaration into manifest members.
1371///
1372/// `in_flight_policy` is present only when it is the non-default rollback
1373/// policy, because format 1 recorded nothing for the default and the two
1374/// formats must agree on what a chunk declaration means.
1375fn chunk_declaration_manifest(revisions: &ChunkComponentRevisions) -> Value {
1376    let mut value = json!({
1377        "checkpoint": {
1378            "schema": revisions.checkpoint_schema().as_str(),
1379            "version": revisions.checkpoint_schema_version().get()
1380        },
1381        "components": {
1382            "checkpoint": revisions.checkpoint().as_str(),
1383            "processor": revisions.processor().as_str(),
1384            "reader": revisions.reader().as_str(),
1385            "writer": revisions.writer().as_str()
1386        },
1387        "context": {
1388            "schema": revisions.context_schema().as_str(),
1389            "version": revisions.context_schema_version().get()
1390        },
1391        "delivery_mode": revisions.delivery_mode().manifest_name()
1392    });
1393    if revisions.in_flight_policy() == InFlightPolicy::RollbackChunk
1394        && let Some(object) = value.as_object_mut()
1395    {
1396        object.insert(
1397            "in_flight_policy".to_owned(),
1398            Value::String("rollback_chunk".to_owned()),
1399        );
1400    }
1401    value
1402}
1403
1404fn fault_manifest_value(policy: &FaultPolicy) -> Value {
1405    let backoff = policy.backoff();
1406    let rules: Vec<Value> = policy
1407        .classifier()
1408        .rules()
1409        .iter()
1410        .map(|rule| {
1411            json!({
1412                "category": rule.category().as_str(),
1413                "phase": rule.phase().as_str(),
1414                "retryable": rule.action().is_retryable(),
1415                "skip": rule
1416                    .action()
1417                    .skip_disposition()
1418                    .map_or(Value::Null, |skip| Value::String(skip.as_str().to_owned()))
1419            })
1420        })
1421        .collect();
1422    json!({
1423        "backoff": {
1424            "initial_ms": u64::try_from(backoff.initial().as_millis()).unwrap_or(u64::MAX),
1425            "kind": backoff.kind().as_str(),
1426            "maximum_ms": u64::try_from(backoff.maximum().as_millis()).unwrap_or(u64::MAX),
1427            "multiplier": backoff.multiplier()
1428        },
1429        "classifier": {
1430            "revision": policy.classifier().revision().as_str(),
1431            "rules": rules
1432        },
1433        "retry_limit": policy.retry_limit().get(),
1434        "retry_state_limit": policy.retry_state_limit().get(),
1435        "skip_limit": policy.skip_limit().get()
1436    })
1437}
1438
1439/// Projects the compiled graph into its canonical restart-relevant manifest.
1440///
1441/// The projection carries exactly the values that select or reinterpret durable
1442/// state. Framework capacity bounds are deliberately absent: they belong to the
1443/// runtime that reads a manifest, not to the definition it identifies, so
1444/// raising one in a later release must not change a fingerprint. `MAX_NODES`
1445/// and `MAX_TRANSITIONS` are enforced against the graph a manifest declares by
1446/// [`DefinitionManifest::read`](oxide_batch_core::DefinitionManifest::read).
1447fn flow_manifest(
1448    job_name: &JobName,
1449    entry: &NodeId,
1450    nodes: &BTreeMap<NodeId, FlowNode>,
1451    transitions: &BTreeMap<NodeId, Vec<FlowTransition>>,
1452    local_scale: bool,
1453) -> Value {
1454    let node_values: Vec<Value> = nodes.values().map(FlowNode::manifest_value).collect();
1455    let transition_values: Vec<Value> = transitions
1456        .values()
1457        .flat_map(|edges| edges.iter().map(FlowTransition::manifest_value))
1458        .collect();
1459    json!({
1460        "entry": entry.as_str(),
1461        "format": if local_scale {
1462            oxide_batch_core::MANIFEST_FORMAT_LOCAL_SCALE
1463        } else {
1464            oxide_batch_core::MANIFEST_FORMAT_FLOW
1465        },
1466        "job": job_name.as_str(),
1467        "nodes": node_values,
1468        "transitions": transition_values
1469    })
1470}
1471
1472/// A validated, immutable execution plan.
1473///
1474/// A plan owns the exact canonical manifest and fingerprint that identify the
1475/// definition across restart. A plan lowered from a one-step wrapper retains
1476/// that wrapper's original format-1 manifest bytes instead of emitting new
1477/// ones, so lowering never changes a persisted identity.
1478#[derive(Clone, Debug, Eq, PartialEq)]
1479pub struct CompiledExecutionPlan {
1480    definition: DefinitionIdentity,
1481    entry: NodeId,
1482    nodes: BTreeMap<NodeId, FlowNode>,
1483    transitions: BTreeMap<NodeId, Vec<FlowTransition>>,
1484}
1485
1486impl CompiledExecutionPlan {
1487    /// Lowers one validated wrapper step into an in-memory compatibility plan.
1488    ///
1489    /// The plan reuses `definition` unchanged, so its manifest bytes, format,
1490    /// and fingerprint stay exactly what the wrapper persisted. The synthetic
1491    /// graph maps the framework's own exit codes onto terminals and adds no
1492    /// node an application could observe as a new durable decision.
1493    #[doc(hidden)]
1494    pub fn compatibility_one_step(
1495        definition: DefinitionIdentity,
1496        step: StepNode,
1497    ) -> Result<Self, PlanError> {
1498        let entry = step.id().clone();
1499        let mut nodes = BTreeMap::new();
1500        nodes.insert(entry.clone(), FlowNode::step(step));
1501        let mut edges = Vec::with_capacity(3);
1502        for (code, terminal) in [
1503            ("COMPLETED", TerminalKind::Complete),
1504            ("FAILED", TerminalKind::Fail),
1505            ("STOPPED", TerminalKind::Stop),
1506        ] {
1507            edges.push(FlowTransition::new(
1508                entry.clone(),
1509                ExitPattern::new(code)?,
1510                FlowTarget::Terminal(terminal),
1511            ));
1512        }
1513        check_unambiguous(&entry, &edges)?;
1514        let mut transitions = BTreeMap::new();
1515        transitions.insert(entry.clone(), edges);
1516        Ok(Self {
1517            definition,
1518            entry,
1519            nodes,
1520            transitions,
1521        })
1522    }
1523
1524    /// Borrows the restart-relevant definition identity.
1525    #[must_use]
1526    pub const fn definition_identity(&self) -> &DefinitionIdentity {
1527        &self.definition
1528    }
1529
1530    /// Returns the canonical manifest format this plan is identified by.
1531    #[must_use]
1532    pub const fn manifest_format(&self) -> u16 {
1533        self.definition.manifest_format()
1534    }
1535
1536    /// Returns the SHA-256 definition fingerprint.
1537    #[must_use]
1538    pub const fn fingerprint(&self) -> &[u8; 32] {
1539        self.definition.manifest_digest()
1540    }
1541
1542    /// Borrows the entry node identifier.
1543    #[must_use]
1544    pub const fn entry(&self) -> &NodeId {
1545        &self.entry
1546    }
1547
1548    /// Returns the compiled node count.
1549    #[must_use]
1550    pub fn node_count(&self) -> usize {
1551        self.nodes.len()
1552    }
1553
1554    /// Returns the compiled transition count.
1555    #[must_use]
1556    pub fn transition_count(&self) -> usize {
1557        self.transitions.values().map(Vec::len).sum()
1558    }
1559
1560    /// Borrows one compiled node.
1561    #[must_use]
1562    pub fn node(&self, id: &NodeId) -> Option<&FlowNode> {
1563        self.nodes.get(id)
1564    }
1565
1566    /// Iterates over compiled nodes in stable logical-identifier order.
1567    ///
1568    /// The returned order is canonical and independent of builder declaration
1569    /// order. It is useful when binding executable components to an immutable
1570    /// plan before launch.
1571    #[must_use]
1572    pub fn nodes(&self) -> impl ExactSizeIterator<Item = (&NodeId, &FlowNode)> {
1573        self.nodes.iter()
1574    }
1575
1576    /// Borrows one node's transitions in evaluation order.
1577    ///
1578    /// The first matching transition wins, so the slice is ordered from the
1579    /// most specific pattern to the least specific one.
1580    #[must_use]
1581    pub fn transitions(&self, id: &NodeId) -> &[FlowTransition] {
1582        self.transitions.get(id).map_or(&[], Vec::as_slice)
1583    }
1584
1585    /// Selects the target one node's exit outcome reaches.
1586    ///
1587    /// # Errors
1588    ///
1589    /// Returns [`FlowSelectionError::UnknownNode`] when `id` is not compiled
1590    /// into this plan and [`FlowSelectionError::UnmappedExitOutcome`] when no
1591    /// declared pattern matches `code`. The plan never selects an arbitrary
1592    /// default.
1593    pub fn select_target(
1594        &self,
1595        id: &NodeId,
1596        code: &ExitCode,
1597    ) -> Result<&FlowTarget, FlowSelectionError> {
1598        let edges = self
1599            .transitions
1600            .get(id)
1601            .ok_or_else(|| FlowSelectionError::UnknownNode { node: id.clone() })?;
1602        edges
1603            .iter()
1604            .find(|edge| edge.pattern().matches(code))
1605            .map(FlowTransition::target)
1606            .ok_or_else(|| FlowSelectionError::UnmappedExitOutcome {
1607                node: id.clone(),
1608                code: code.clone(),
1609            })
1610    }
1611}
1612
1613/// A flow graph that cannot be compiled into an executable plan.
1614#[derive(Clone, Debug, Eq, PartialEq)]
1615#[non_exhaustive]
1616pub enum PlanError {
1617    /// No entry node was declared.
1618    MissingEntryNode,
1619    /// Two nodes declared the same logical identifier.
1620    DuplicateNodeId {
1621        /// Repeated identifier.
1622        node: NodeId,
1623    },
1624    /// A transition referenced a node the graph does not declare.
1625    UndefinedNode {
1626        /// Missing identifier.
1627        node: NodeId,
1628    },
1629    /// A node declared no outgoing transition.
1630    MissingTransition {
1631        /// Identifier of the node without a transition.
1632        node: NodeId,
1633    },
1634    /// Two equally specific patterns can match one exit outcome.
1635    AmbiguousTransition {
1636        /// Identifier of the node with the ambiguity.
1637        node: NodeId,
1638        /// First conflicting pattern.
1639        first: ExitPattern,
1640        /// Second conflicting pattern.
1641        second: ExitPattern,
1642    },
1643    /// A node cannot be reached from the entry node.
1644    UnreachableNode {
1645        /// Unreachable identifier.
1646        node: NodeId,
1647    },
1648    /// The graph contains a cycle.
1649    CyclicGraph {
1650        /// Identifier revisited on one path.
1651        node: NodeId,
1652    },
1653    /// The graph declared more nodes than M3 accepts.
1654    TooManyNodes {
1655        /// Maximum accepted node count.
1656        max: usize,
1657    },
1658    /// The graph declared more transitions than M3 accepts.
1659    TooManyTransitions {
1660        /// Maximum accepted transition count.
1661        max: usize,
1662    },
1663    /// One node declared more outgoing transitions than M3 accepts.
1664    TooManyOutgoingTransitions {
1665        /// Identifier of the node with too many transitions.
1666        node: NodeId,
1667        /// Maximum accepted outgoing transition count.
1668        max: usize,
1669    },
1670    /// An exit pattern violated its bounded format.
1671    InvalidPattern {
1672        /// Maximum accepted pattern length in UTF-8 bytes.
1673        max_bytes: usize,
1674    },
1675    /// A decision input-contract version of zero is not a version.
1676    ZeroDecisionInputVersion,
1677    /// A split declared fewer than two or more than eight branches.
1678    InvalidSplitBranchCount {
1679        /// Split whose branch count was invalid.
1680        split: NodeId,
1681        /// Minimum accepted branch count.
1682        min: usize,
1683        /// Maximum accepted branch count.
1684        max: usize,
1685    },
1686    /// A split branch was empty or longer than the accepted bound.
1687    InvalidBranchLength {
1688        /// Owning split.
1689        split: NodeId,
1690        /// Maximum accepted branch length.
1691        max: usize,
1692    },
1693    /// A split was declared as the graph entry.
1694    SplitIsEntry {
1695        /// Rejected split.
1696        split: NodeId,
1697    },
1698    /// A split did not reference a declared structural join.
1699    InvalidSplitJoin {
1700        /// Owning split.
1701        split: NodeId,
1702        /// Missing or wrongly typed join.
1703        join: NodeId,
1704    },
1705    /// More than one split tried to own the same join.
1706    JoinHasMultipleOwners {
1707        /// Multiply owned join.
1708        join: NodeId,
1709        /// First owning split.
1710        first: NodeId,
1711        /// Second owning split.
1712        second: NodeId,
1713    },
1714    /// A join was not owned by any split.
1715    OrphanJoin {
1716        /// Unowned join.
1717        join: NodeId,
1718    },
1719    /// A normal transition attempted to enter a structural join.
1720    JoinHasExternalEntry {
1721        /// Join with an external incoming edge.
1722        join: NodeId,
1723    },
1724    /// A split tried to bypass its implicit join edge.
1725    SplitHasExplicitTransition {
1726        /// Split with the explicit edge.
1727        split: NodeId,
1728    },
1729    /// The branch concurrency budget exceeded the declared branch count.
1730    ParallelBudgetExceedsBranches {
1731        /// Split with the contradictory budget.
1732        split: NodeId,
1733        /// Declared branch count.
1734        branches: usize,
1735    },
1736    /// A branch concurrency budget was zero or above the M4 ceiling.
1737    InvalidParallelBranchBudget {
1738        /// Maximum accepted branch concurrency.
1739        max: usize,
1740    },
1741    /// A partition-worker budget was zero or above the M4 ceiling.
1742    InvalidPartitionWorkerBudget {
1743        /// Maximum accepted worker concurrency.
1744        max: u8,
1745    },
1746    /// The declared pool cannot supply active children plus their parent.
1747    InsufficientPoolCapacity {
1748        /// Minimum required connection count.
1749        required: u32,
1750        /// Declared connection count.
1751        configured: u32,
1752    },
1753    /// A durable partition count was zero or above the M4 ceiling.
1754    InvalidPartitionCount {
1755        /// Maximum accepted partition count.
1756        max: u16,
1757    },
1758    /// A logical identifier or revision token was invalid.
1759    Token(DefinitionError),
1760    /// The canonical manifest could not be encoded within its bound.
1761    Manifest(DefinitionError),
1762}
1763
1764impl fmt::Display for PlanError {
1765    #[allow(
1766        clippy::too_many_lines,
1767        reason = "each typed plan rejection retains one stable redacted diagnostic"
1768    )]
1769    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1770        match self {
1771            Self::MissingEntryNode => formatter.write_str("flow graph has no entry node"),
1772            Self::DuplicateNodeId { node } => {
1773                write!(
1774                    formatter,
1775                    "node {} is declared more than once",
1776                    node.as_str()
1777                )
1778            }
1779            Self::UndefinedNode { node } => {
1780                write!(formatter, "node {} is not declared", node.as_str())
1781            }
1782            Self::MissingTransition { node } => {
1783                write!(
1784                    formatter,
1785                    "node {} has no outgoing transition",
1786                    node.as_str()
1787                )
1788            }
1789            Self::AmbiguousTransition {
1790                node,
1791                first,
1792                second,
1793            } => write!(
1794                formatter,
1795                "node {} patterns {first} and {second} are equally specific and overlap",
1796                node.as_str()
1797            ),
1798            Self::UnreachableNode { node } => {
1799                write!(
1800                    formatter,
1801                    "node {} is unreachable from the entry node",
1802                    node.as_str()
1803                )
1804            }
1805            Self::CyclicGraph { node } => {
1806                write!(formatter, "node {} closes a cycle", node.as_str())
1807            }
1808            Self::TooManyNodes { max } => write!(formatter, "flow graph exceeds {max} nodes"),
1809            Self::TooManyTransitions { max } => {
1810                write!(formatter, "flow graph exceeds {max} transitions")
1811            }
1812            Self::TooManyOutgoingTransitions { node, max } => write!(
1813                formatter,
1814                "node {} exceeds {max} outgoing transitions",
1815                node.as_str()
1816            ),
1817            Self::InvalidPattern { max_bytes } => write!(
1818                formatter,
1819                "exit pattern must be 1 to {max_bytes} bytes without control characters"
1820            ),
1821            Self::ZeroDecisionInputVersion => {
1822                formatter.write_str("decision input version must be nonzero")
1823            }
1824            Self::InvalidSplitBranchCount { split, min, max } => write!(
1825                formatter,
1826                "split {} must declare {min} to {max} branches",
1827                split.as_str()
1828            ),
1829            Self::InvalidBranchLength { split, max } => write!(
1830                formatter,
1831                "split {} branches must declare 1 to {max} steps",
1832                split.as_str()
1833            ),
1834            Self::SplitIsEntry { split } => {
1835                write!(
1836                    formatter,
1837                    "split {} cannot be the entry node",
1838                    split.as_str()
1839                )
1840            }
1841            Self::InvalidSplitJoin { split, join } => write!(
1842                formatter,
1843                "split {} does not own declared join {}",
1844                split.as_str(),
1845                join.as_str()
1846            ),
1847            Self::JoinHasMultipleOwners {
1848                join,
1849                first,
1850                second,
1851            } => write!(
1852                formatter,
1853                "join {} is owned by both splits {} and {}",
1854                join.as_str(),
1855                first.as_str(),
1856                second.as_str()
1857            ),
1858            Self::OrphanJoin { join } => {
1859                write!(formatter, "join {} has no owning split", join.as_str())
1860            }
1861            Self::JoinHasExternalEntry { join } => write!(
1862                formatter,
1863                "join {} can be entered only by its owning split",
1864                join.as_str()
1865            ),
1866            Self::SplitHasExplicitTransition { split } => write!(
1867                formatter,
1868                "split {} reaches only its declared join",
1869                split.as_str()
1870            ),
1871            Self::ParallelBudgetExceedsBranches { split, branches } => write!(
1872                formatter,
1873                "split {} parallel budget exceeds its {branches} branches",
1874                split.as_str()
1875            ),
1876            Self::InvalidParallelBranchBudget { max } => {
1877                write!(formatter, "parallel branch budget must be 1 to {max}")
1878            }
1879            Self::InvalidPartitionWorkerBudget { max } => {
1880                write!(formatter, "partition worker budget must be 1 to {max}")
1881            }
1882            Self::InsufficientPoolCapacity {
1883                required,
1884                configured,
1885            } => write!(
1886                formatter,
1887                "repository pool size {configured} cannot supply required capacity {required}"
1888            ),
1889            Self::InvalidPartitionCount { max } => {
1890                write!(formatter, "partition count must be 1 to {max}")
1891            }
1892            Self::Token(error) => write!(formatter, "flow graph token is invalid: {error}"),
1893            Self::Manifest(error) => {
1894                write!(formatter, "flow manifest could not be encoded: {error}")
1895            }
1896        }
1897    }
1898}
1899
1900impl Error for PlanError {
1901    fn source(&self) -> Option<&(dyn Error + 'static)> {
1902        match self {
1903            Self::Token(error) | Self::Manifest(error) => Some(error),
1904            _ => None,
1905        }
1906    }
1907}
1908
1909impl From<DefinitionError> for PlanError {
1910    fn from(error: DefinitionError) -> Self {
1911        Self::Token(error)
1912    }
1913}
1914
1915/// A compiled plan that cannot route one observed exit outcome.
1916#[derive(Clone, Debug, Eq, PartialEq)]
1917#[non_exhaustive]
1918pub enum FlowSelectionError {
1919    /// The requested node is not part of this plan.
1920    UnknownNode {
1921        /// Requested identifier.
1922        node: NodeId,
1923    },
1924    /// No declared pattern matches the produced exit outcome.
1925    UnmappedExitOutcome {
1926        /// Node whose outcome could not be routed.
1927        node: NodeId,
1928        /// Produced exit code.
1929        code: ExitCode,
1930    },
1931}
1932
1933impl fmt::Display for FlowSelectionError {
1934    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1935        match self {
1936            Self::UnknownNode { node } => {
1937                write!(
1938                    formatter,
1939                    "node {} is not part of the compiled plan",
1940                    node.as_str()
1941                )
1942            }
1943            Self::UnmappedExitOutcome { node, code } => write!(
1944                formatter,
1945                "node {} declares no transition for exit outcome {code}",
1946                node.as_str()
1947            ),
1948        }
1949    }
1950}
1951
1952impl Error for FlowSelectionError {}
oxide_batch_plan/lib.rs

oxide_batch_plan/
lib.rs
