runledger-core 0.1.1

Core contracts and types for the Runledger durable job and workflow system
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

use std::collections::BTreeSet;

use super::super::identifiers::{JobType, StepKey};
use super::dag_validation::validate_workflow_run_enqueue;
use super::errors::WorkflowBuildError;
use super::types::{WorkflowRunEnqueue, WorkflowStepEnqueue, WorkflowStepExecutionKind};

pub(super) fn validate_step_enqueue(
    step: &WorkflowStepEnqueue<'_>,
    step_index: Option<usize>,
) -> Result<(), WorkflowBuildError> {
    if StepKey::try_new(step.step_key.as_str()).is_err() {
        return Err(WorkflowBuildError::BlankStepKey { step_index });
    }
    match step.execution_kind {
        WorkflowStepExecutionKind::Job => {
            let Some(job_type) = step.job_type else {
                return Err(WorkflowBuildError::BlankStepJobType {
                    step_key: step.step_key.as_str().to_owned(),
                });
            };
            if JobType::try_new(job_type.as_str()).is_err() {
                return Err(WorkflowBuildError::BlankStepJobType {
                    step_key: step.step_key.as_str().to_owned(),
                });
            }
            if let Some(max_attempts) = step.max_attempts
                && max_attempts <= 0
            {
                return Err(WorkflowBuildError::NonPositiveStepMaxAttempts {
                    step_key: step.step_key.as_str().to_owned(),
                    max_attempts,
                });
            }
            if let Some(timeout_seconds) = step.timeout_seconds
                && timeout_seconds <= 0
            {
                return Err(WorkflowBuildError::NonPositiveStepTimeoutSeconds {
                    step_key: step.step_key.as_str().to_owned(),
                    timeout_seconds,
                });
            }
        }
        WorkflowStepExecutionKind::External => {
            if step.job_type.is_some() {
                return Err(WorkflowBuildError::ExternalStepJobTypeNotAllowed {
                    step_key: step.step_key.as_str().to_owned(),
                });
            }
            if step.priority.is_some()
                || step.max_attempts.is_some()
                || step.timeout_seconds.is_some()
                || step.stage.is_some()
            {
                return Err(WorkflowBuildError::ExternalStepQueueSettingsNotAllowed {
                    step_key: step.step_key.as_str().to_owned(),
                });
            }
        }
    }
    let mut seen_dependencies: BTreeSet<&str> = BTreeSet::new();
    for dependency in &step.dependencies {
        if StepKey::try_new(dependency.prerequisite_step_key.as_str()).is_err() {
            return Err(WorkflowBuildError::BlankDependencyStepKey {
                step_key: step.step_key.as_str().to_owned(),
            });
        }
        if dependency.prerequisite_step_key == step.step_key {
            return Err(WorkflowBuildError::SelfDependency {
                step_key: step.step_key.as_str().to_owned(),
            });
        }
        if !seen_dependencies.insert(dependency.prerequisite_step_key.as_str()) {
            return Err(WorkflowBuildError::DuplicateDependency {
                step_key: step.step_key.as_str().to_owned(),
                prerequisite_step_key: dependency.prerequisite_step_key.as_str().to_owned(),
            });
        }
    }

    Ok(())
}

pub(super) fn validate_workflow_enqueue(
    payload: &WorkflowRunEnqueue<'_>,
) -> Result<(), WorkflowBuildError> {
    validate_workflow_run_enqueue(payload).map_err(Into::into)
}