runx-core 0.6.19

Pure Rust parity kernel for runx state-machine and policy decisions.
Documentation 
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
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134

use std::collections::BTreeSet;

use runx_contracts::JsonValue;

use super::super::types::{
    FanoutBranchResult, FanoutGate, FanoutGroupPolicy, FanoutSyncDecision, FanoutSyncOutcome,
    FanoutThresholdGate, GraphStepStatus,
};
use super::values::{json_value_as_f64, resolve_structured_field};
use super::{Counts, DecisionDetails, is_resolved, sync_decision};

pub(super) fn threshold_decision(
    policy: &FanoutGroupPolicy,
    results: &[FanoutBranchResult],
    resolved_gate_keys: Option<&BTreeSet<String>>,
    counts: Counts,
) -> Option<FanoutSyncDecision> {
    for gate in policy.threshold_gates.as_deref().unwrap_or(&[]) {
        let Some(above) = gate.above.as_f64() else {
            continue;
        };
        let Some(result) = results
            .iter()
            .find(|candidate| candidate.step_id == gate.step)
        else {
            continue;
        };
        if result.status != GraphStepStatus::Succeeded {
            continue;
        }
        let Some(value) = resolve_structured_field(result.outputs.as_ref(), &gate.field) else {
            return Some(threshold_missing_decision(policy, gate, counts));
        };
        let Some(number) = json_value_as_f64(value) else {
            return Some(threshold_non_numeric_decision(
                policy,
                gate,
                value.clone(),
                counts,
            ));
        };
        if number > above {
            let decision = threshold_above_decision(policy, gate, value.clone(), number, counts);
            if !is_resolved(&decision, resolved_gate_keys) {
                return Some(decision);
            }
        }
    }
    None
}

fn threshold_missing_decision(
    policy: &FanoutGroupPolicy,
    gate: &FanoutThresholdGate,
    counts: Counts,
) -> FanoutSyncDecision {
    sync_decision(
        policy,
        FanoutSyncOutcome::Halt,
        "threshold",
        counts,
        DecisionDetails::new(
            format!("threshold.{}.{}.missing", gate.step, gate.field),
            format!(
                "threshold field {}.{} was not produced",
                gate.step, gate.field
            ),
        )
        .with_gate(FanoutGate::Threshold {
            step_id: Some(gate.step.clone()),
            field: gate.field.clone(),
            value: None,
            compared_to: None,
            action: gate.action.clone(),
        }),
    )
}

fn threshold_non_numeric_decision(
    policy: &FanoutGroupPolicy,
    gate: &FanoutThresholdGate,
    value: JsonValue,
    counts: Counts,
) -> FanoutSyncDecision {
    sync_decision(
        policy,
        FanoutSyncOutcome::Halt,
        "threshold",
        counts,
        DecisionDetails::new(
            format!("threshold.{}.{}.non_numeric", gate.step, gate.field),
            format!(
                "threshold field {}.{} must be numeric",
                gate.step, gate.field
            ),
        )
        .with_gate(FanoutGate::Threshold {
            step_id: Some(gate.step.clone()),
            field: gate.field.clone(),
            value: Some(value),
            compared_to: None,
            action: gate.action.clone(),
        }),
    )
}

fn threshold_above_decision(
    policy: &FanoutGroupPolicy,
    gate: &FanoutThresholdGate,
    value: JsonValue,
    number: f64,
    counts: Counts,
) -> FanoutSyncDecision {
    sync_decision(
        policy,
        gate.action.clone().into(),
        "threshold",
        counts,
        DecisionDetails::new(
            format!("threshold.{}.{}.above", gate.step, gate.field),
            format!(
                "{}.{}={number} exceeded {}",
                gate.step, gate.field, gate.above
            ),
        )
        .with_gate(FanoutGate::Threshold {
            step_id: Some(gate.step.clone()),
            field: gate.field.clone(),
            value: Some(value),
            compared_to: Some(gate.above.clone()),
            action: gate.action.clone(),
        }),
    )
}