dpcs 0.7.0

Reference implementation of the Data Pipeline Contract Standard (DPCS)
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
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152

//! Graph validation phase.

use crate::diagnostics::{categories, Diagnostic, ValidationReport};
use crate::model::{DependencyGraph, PipelineContract};

/// Validate graph edges and detect prohibited cycles.
pub fn validate(contract: &PipelineContract) -> ValidationReport {
    let mut report = ValidationReport::new();
    let step_ids = contract.step_ids();

    for (index, point) in contract.graph.entry_points.iter().enumerate() {
        if point.trim().is_empty() {
            continue;
        }
        if !step_ids.contains(point.as_str()) {
            report.push(
                Diagnostic::error(
                    "DPCS-GRP-007",
                    categories::GRAPH,
                    format!("graph entry point references unknown step `{point}`"),
                )
                .with_object_ref(format!("graph.entryPoints[{index}]"))
                .with_remediation(
                    "Declare graph.entryPoints that refer to declared step identifiers",
                ),
            );
        }
    }

    for (index, point) in contract.graph.exit_points.iter().enumerate() {
        if point.trim().is_empty() {
            continue;
        }
        if !step_ids.contains(point.as_str()) {
            report.push(
                Diagnostic::error(
                    "DPCS-GRP-008",
                    categories::GRAPH,
                    format!("graph exit point references unknown step `{point}`"),
                )
                .with_object_ref(format!("graph.exitPoints[{index}]"))
                .with_remediation(
                    "Declare graph.exitPoints that refer to declared step identifiers",
                ),
            );
        }
    }

    for (index, edge) in contract.graph.edges.iter().enumerate() {
        let object_ref = format!("graph.edges[{index}]");

        if edge.from.trim().is_empty() || edge.to.trim().is_empty() {
            report.push(
                Diagnostic::error(
                    "DPCS-GRP-001",
                    categories::GRAPH,
                    "graph edge endpoints must not be empty",
                )
                .with_object_ref(object_ref.clone()),
            );
            continue;
        }

        if !step_ids.contains(edge.from.as_str()) {
            report.push(
                Diagnostic::error(
                    "DPCS-GRP-002",
                    categories::GRAPH,
                    format!("graph edge references unknown step `{}`", edge.from),
                )
                .with_object_ref(object_ref.clone())
                .with_remediation("Ensure edge endpoints refer to declared step identifiers"),
            );
        }

        if !step_ids.contains(edge.to.as_str()) {
            report.push(
                Diagnostic::error(
                    "DPCS-GRP-003",
                    categories::GRAPH,
                    format!("graph edge references unknown step `{}`", edge.to),
                )
                .with_object_ref(object_ref)
                .with_remediation("Ensure edge endpoints refer to declared step identifiers"),
            );
        }
    }

    for duplicate in DependencyGraph::duplicate_edges(contract) {
        let kind_label = duplicate
            .kind
            .as_deref()
            .map(|kind| format!(" (kind `{kind}`)"))
            .unwrap_or_else(|| " (untyped)".to_string());
        report.push(
            Diagnostic::error(
                "DPCS-GRP-005",
                categories::GRAPH,
                format!(
                    "duplicate graph edge from `{}` to `{}`{kind_label}",
                    duplicate.from, duplicate.to
                ),
            )
            .with_object_ref(format!("graph.edges[{}]", duplicate.duplicate_index))
            .with_remediation("Remove duplicate edges from graph.edges"),
        );
    }

    let dependency_graph = DependencyGraph::from_contract(contract);
    let cycle = dependency_graph.find_cycle();
    let has_cycle = cycle.is_some();

    if let Some(cycle) = cycle {
        let cycle_node = cycle.first().cloned().unwrap_or_default();
        report.push(
            Diagnostic::error(
                "DPCS-GRP-004",
                categories::GRAPH,
                format!("pipeline graph contains a prohibited cycle involving `{cycle_node}`"),
            )
            .with_object_ref("graph")
            .with_remediation(
                "Remove cyclic dependencies from graph.edges, controlFlow, or dataFlow",
            ),
        );
    }

    if !has_cycle {
        let has_declared_entry_points = contract
            .graph
            .entry_points
            .iter()
            .any(|id| !id.trim().is_empty());
        let unreachable = dependency_graph.unreachable_steps(contract);
        for step_id in unreachable {
            let message = if has_declared_entry_points {
                format!("pipeline step `{step_id}` is unreachable from graph entry points")
            } else {
                format!("pipeline step `{step_id}` is unreachable from graph roots")
            };
            report.push(
                Diagnostic::error("DPCS-GRP-006", categories::GRAPH, message)
                    .with_object_ref(format!("steps.{step_id}"))
                    .with_remediation(
                        "Declare graph.entryPoints or add edges so every step is reachable",
                    ),
            );
        }
    }

    report
}