aves 0.5.0

Typed async application runtime with domain messaging, RPC, and durable workflow automation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

use super::graph::{FlowGraph, GraphNode, ParallelPolicy};
use super::registration::{CancellationDisposition, FlowNodeRegistry, SideEffectDisposition};
use super::types::ControllerError;

/// Validates the controller policies before a Flow is compiled for execution.
pub(super) fn validate_parallel_composition(
    graph: &FlowGraph,
    registry: &FlowNodeRegistry,
) -> Result<(), ControllerError> {
    validate_node(&graph.ast().root, registry)
}

fn validate_node(node: &GraphNode, registry: &FlowNodeRegistry) -> Result<(), ControllerError> {
    match node {
        GraphNode::AlwaysSucceed { child } => validate_node(child, registry),
        GraphNode::Sequence { children } | GraphNode::Selector { children } | GraphNode::Fallback { children } => {
            children
                .iter()
                .try_for_each(|child| validate_node(child, registry))
        }
        GraphNode::Parallel { policy, children } => {
            if matches!(policy, ParallelPolicy::WhenAny | ParallelPolicy::Race) {
                // Resolve each child's registrations before its discard-safety check.
                for child in children {
                    let mut registration_error = None;
                    child.for_each_leaf(&mut |node| {
                        if registration_error.is_none() {
                            registration_error = registry
                                .get(&node.node_kind)
                                .map(|_| ())
                                .map_err(ControllerError::from)
                                .err();
                        }
                    });
                    if let Some(error) = registration_error {
                        return Err(error);
                    }

                    let mut unsafe_kind = None;
                    child.for_each_leaf(&mut |node| {
                        let kind = node.node_kind.clone();
                        let descriptor = &registry
                            .get(&kind)
                            .expect("this child's registrations were validated above")
                            .descriptor;
                        if unsafe_kind.is_none()
                            && descriptor.cancellation == CancellationDisposition::NonCancellable
                            && descriptor.side_effects == SideEffectDisposition::SideEffecting
                        {
                            unsafe_kind = Some(kind);
                        }
                    });
                    if let Some(kind) = unsafe_kind {
                        return Err(ControllerError::UnsafeDiscard { kind });
                    }
                }
                return children
                    .iter()
                    .try_for_each(|child| validate_node(child, registry));
            }
            children
                .iter()
                .try_for_each(|child| validate_node(child, registry))
        }
        GraphNode::Leaf { node } => {
            registry.get(&node.node_kind)?;
            Ok(())
        }
    }
}