dpcs 0.12.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

//! Validation phase orchestration.

use crate::diagnostics::ValidationReport;
use crate::model::{AnalysisContext, PipelineContract};

use super::{
    com, control_flow, data_flow, document, execution, extensions, failure, governance, graph,
    lineage, quality, references, scheduling, security, structural,
};

/// Named validation phase used by orchestration and incremental caching.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub enum ValidationPhase {
    /// Document-level checks.
    Document,
    /// Canonical Object Model invariants.
    Com,
    /// Structural checks.
    Structural,
    /// Graph topology.
    Graph,
    /// Reference resolution.
    References,
    /// Data-flow wiring.
    DataFlow,
    /// Control-flow conflicts.
    ControlFlow,
    /// Execution requirements.
    Execution,
    /// Scheduling intents.
    Scheduling,
    /// Quality gates.
    Quality,
    /// Failure semantics.
    Failure,
    /// Lineage.
    Lineage,
    /// Security metadata.
    Security,
    /// Governance metadata.
    Governance,
    /// Extensions / compatibility.
    Extensions,
}

impl ValidationPhase {
    /// Phases in the SPEC-aligned orchestration order.
    pub const ALL: [Self; 15] = [
        Self::Document,
        Self::Com,
        Self::Structural,
        Self::Graph,
        Self::References,
        Self::DataFlow,
        Self::ControlFlow,
        Self::Execution,
        Self::Scheduling,
        Self::Quality,
        Self::Failure,
        Self::Lineage,
        Self::Security,
        Self::Governance,
        Self::Extensions,
    ];
}

/// Validate a Pipeline Contract using the DPCS phase model.
///
/// When the `parallel` feature is enabled, independent phases run concurrently
/// and diagnostics are merged then sorted deterministically. Without that
/// feature this is identical to [`validate_sequential`].
pub fn validate(contract: &PipelineContract) -> ValidationReport {
    let ctx = AnalysisContext::build(contract);
    validate_with_context(&ctx)
}

/// Validate using a prebuilt [`AnalysisContext`].
pub fn validate_with_context(ctx: &AnalysisContext<'_>) -> ValidationReport {
    #[cfg(feature = "parallel")]
    {
        validate_parallel(ctx)
    }
    #[cfg(not(feature = "parallel"))]
    {
        validate_sequential_with_context(ctx)
    }
}

/// Always-sequential validation (useful for benches and determinism checks).
pub fn validate_sequential(contract: &PipelineContract) -> ValidationReport {
    let ctx = AnalysisContext::build(contract);
    validate_sequential_with_context(&ctx)
}

/// Sequential validation using a prebuilt context.
pub fn validate_sequential_with_context(ctx: &AnalysisContext<'_>) -> ValidationReport {
    let mut report = ValidationReport::new();
    for phase in ValidationPhase::ALL {
        report.extend(run_phase(phase, ctx));
    }
    report.sort_deterministic();
    report
}

#[cfg(feature = "parallel")]
fn validate_parallel(ctx: &AnalysisContext<'_>) -> ValidationReport {
    use rayon::prelude::*;

    let mut reports: Vec<ValidationReport> = ValidationPhase::ALL
        .into_par_iter()
        .map(|phase| run_phase(phase, ctx))
        .collect();

    let mut report = ValidationReport::new();
    for phase_report in reports.drain(..) {
        report.extend(phase_report);
    }
    report.sort_deterministic();
    report
}

/// Run a single validation phase against a shared analysis context.
pub(crate) fn run_phase(phase: ValidationPhase, ctx: &AnalysisContext<'_>) -> ValidationReport {
    let contract = ctx.contract;
    match phase {
        ValidationPhase::Document => document::validate(contract),
        ValidationPhase::Com => com::validate(contract),
        ValidationPhase::Structural => structural::validate(contract),
        ValidationPhase::Graph => graph::validate_with_context(ctx),
        ValidationPhase::References => references::validate_with_context(ctx),
        ValidationPhase::DataFlow => data_flow::validate_with_context(ctx),
        ValidationPhase::ControlFlow => control_flow::validate_with_context(ctx),
        ValidationPhase::Execution => execution::validate(contract),
        ValidationPhase::Scheduling => scheduling::validate(contract),
        ValidationPhase::Quality => quality::validate_with_context(ctx),
        ValidationPhase::Failure => failure::validate_with_context(ctx),
        ValidationPhase::Lineage => lineage::validate_with_context(ctx),
        ValidationPhase::Security => security::validate(contract),
        ValidationPhase::Governance => governance::validate(contract),
        ValidationPhase::Extensions => extensions::validate(contract),
    }
}