ferrum_interfaces/vnext/execution/
validation.rs

1use super::{
2    invalid_plan, CapabilityCatalog, CompletionRetentionSpec, Deserialize, ExecutionPlan,
3    PlanBuildRequest, PlanNodeResolution, PlanSchemaVersion, PreparedModelFamily, RuntimePolicy,
4    TrustedExecutionWeightPlan, UnvalidatedExecutionPlan, UnvalidatedExecutionPlanWire, VNextError,
5    EXECUTION_PLAN_SCHEMA, MAX_EXECUTION_PLAN_WIRE_BYTES,
6};
7
8#[derive(Deserialize)]
9struct ExecutionPlanSchemaEnvelope {
10    payload: ExecutionPlanSchemaHeader,
11}
12
13#[derive(Deserialize)]
14struct ExecutionPlanSchemaHeader {
15    schema: PlanSchemaVersion,
16}
17
18pub(super) fn validate_execution_plan_wire_size(
19    wire_size: usize,
20    context: &'static str,
21) -> Result<(), VNextError> {
22    if wire_size > MAX_EXECUTION_PLAN_WIRE_BYTES {
23        return Err(VNextError::Serialization {
24            context,
25            message: format!(
26                "execution plan wire size {wire_size} exceeds limit {MAX_EXECUTION_PLAN_WIRE_BYTES}"
27            ),
28        });
29    }
30    Ok(())
31}
32
33impl ExecutionPlan {
34    pub fn to_json(&self) -> Result<Vec<u8>, VNextError> {
35        let bytes = serde_json::to_vec(self).map_err(|error| VNextError::Serialization {
36            context: "serialize execution plan",
37            message: error.to_string(),
38        })?;
39        validate_execution_plan_wire_size(bytes.len(), "serialize execution plan")?;
40        Ok(bytes)
41    }
42
43    pub fn decode_untrusted(bytes: &[u8]) -> Result<UnvalidatedExecutionPlan, VNextError> {
44        const CONTEXT: &str = "decode untrusted execution plan";
45        validate_execution_plan_wire_size(bytes.len(), CONTEXT)?;
46        let header =
47            serde_json::from_slice::<ExecutionPlanSchemaEnvelope>(bytes).map_err(|error| {
48                VNextError::Serialization {
49                    context: CONTEXT,
50                    message: error.to_string(),
51                }
52            })?;
53        if header.payload.schema != EXECUTION_PLAN_SCHEMA {
54            return Err(VNextError::UnsupportedPlanSchema {
55                expected_major: EXECUTION_PLAN_SCHEMA.major,
56                expected_minor: EXECUTION_PLAN_SCHEMA.minor,
57                actual_major: header.payload.schema.major,
58                actual_minor: header.payload.schema.minor,
59            });
60        }
61        serde_json::from_slice::<UnvalidatedExecutionPlanWire>(bytes)
62            .map(UnvalidatedExecutionPlan::from)
63            .map_err(|error| VNextError::Serialization {
64                context: CONTEXT,
65                message: error.to_string(),
66            })
67    }
68
69    pub fn from_json_validated<P: RuntimePolicy>(
70        bytes: &[u8],
71        family: &PreparedModelFamily,
72        capabilities: &CapabilityCatalog,
73        policy: &P,
74        node_resolutions: Vec<PlanNodeResolution>,
75    ) -> Result<Self, VNextError> {
76        Self::decode_untrusted(bytes)?.revalidate(family, capabilities, policy, node_resolutions)
77    }
78
79    pub fn from_json_validated_with_completion_retention<P: RuntimePolicy>(
80        bytes: &[u8],
81        family: &PreparedModelFamily,
82        capabilities: &CapabilityCatalog,
83        policy: &P,
84        node_resolutions: Vec<PlanNodeResolution>,
85        completion_retention: CompletionRetentionSpec,
86    ) -> Result<Self, VNextError> {
87        Self::decode_untrusted(bytes)?.revalidate_with_completion_retention(
88            family,
89            capabilities,
90            policy,
91            node_resolutions,
92            completion_retention,
93        )
94    }
95
96    pub fn from_json_validated_with_execution_weights<P: RuntimePolicy>(
97        bytes: &[u8],
98        family: &PreparedModelFamily,
99        capabilities: &CapabilityCatalog,
100        policy: &P,
101        node_resolutions: Vec<PlanNodeResolution>,
102        completion_retention: CompletionRetentionSpec,
103        execution_weights: TrustedExecutionWeightPlan,
104    ) -> Result<Self, VNextError> {
105        Self::decode_untrusted(bytes)?.revalidate_with_execution_weights(
106            family,
107            capabilities,
108            policy,
109            node_resolutions,
110            completion_retention,
111            execution_weights,
112        )
113    }
114
115    pub fn validate_against<P: RuntimePolicy>(
116        &self,
117        family: &PreparedModelFamily,
118        capabilities: &CapabilityCatalog,
119        policy: &P,
120        node_resolutions: &[PlanNodeResolution],
121    ) -> Result<(), VNextError> {
122        self.validate_against_with_completion_retention(
123            family,
124            capabilities,
125            policy,
126            node_resolutions,
127            CompletionRetentionSpec::default(),
128        )
129    }
130
131    pub fn validate_against_with_completion_retention<P: RuntimePolicy>(
132        &self,
133        family: &PreparedModelFamily,
134        capabilities: &CapabilityCatalog,
135        policy: &P,
136        node_resolutions: &[PlanNodeResolution],
137        completion_retention: CompletionRetentionSpec,
138    ) -> Result<(), VNextError> {
139        let rebuilt = ExecutionPlan::build(
140            PlanBuildRequest::new(family, capabilities, policy, node_resolutions.to_vec())?
141                .with_execution_weights(self.trusted_execution_weights.clone())?
142                .with_completion_retention(completion_retention)?,
143        )?;
144        if rebuilt.operation_registry_authority != self.operation_registry_authority {
145            return Err(invalid_plan(
146                "execution plan belongs to a different operation runtime registry",
147            ));
148        }
149        if &rebuilt != self {
150            return Err(invalid_plan(
151                "execution plan is not identical to its semantic rebuild",
152            ));
153        }
154        Ok(())
155    }
156}
157
158impl UnvalidatedExecutionPlan {
159    pub fn schema(&self) -> PlanSchemaVersion {
160        self.payload.schema
161    }
162
163    pub fn revalidate<P: RuntimePolicy>(
164        self,
165        family: &PreparedModelFamily,
166        capabilities: &CapabilityCatalog,
167        policy: &P,
168        node_resolutions: Vec<PlanNodeResolution>,
169    ) -> Result<ExecutionPlan, VNextError> {
170        self.revalidate_with_completion_retention(
171            family,
172            capabilities,
173            policy,
174            node_resolutions,
175            CompletionRetentionSpec::default(),
176        )
177    }
178
179    pub fn revalidate_with_completion_retention<P: RuntimePolicy>(
180        self,
181        family: &PreparedModelFamily,
182        capabilities: &CapabilityCatalog,
183        policy: &P,
184        node_resolutions: Vec<PlanNodeResolution>,
185        completion_retention: CompletionRetentionSpec,
186    ) -> Result<ExecutionPlan, VNextError> {
187        let execution_weights = TrustedExecutionWeightPlan::identity(family)?;
188        self.revalidate_with_execution_weights(
189            family,
190            capabilities,
191            policy,
192            node_resolutions,
193            completion_retention,
194            execution_weights,
195        )
196    }
197
198    pub fn revalidate_with_execution_weights<P: RuntimePolicy>(
199        self,
200        family: &PreparedModelFamily,
201        capabilities: &CapabilityCatalog,
202        policy: &P,
203        node_resolutions: Vec<PlanNodeResolution>,
204        completion_retention: CompletionRetentionSpec,
205        execution_weights: TrustedExecutionWeightPlan,
206    ) -> Result<ExecutionPlan, VNextError> {
207        if self.payload.schema != EXECUTION_PLAN_SCHEMA {
208            return Err(VNextError::UnsupportedPlanSchema {
209                expected_major: EXECUTION_PLAN_SCHEMA.major,
210                expected_minor: EXECUTION_PLAN_SCHEMA.minor,
211                actual_major: self.payload.schema.major,
212                actual_minor: self.payload.schema.minor,
213            });
214        }
215        let rebuilt = ExecutionPlan::build(
216            PlanBuildRequest::new(family, capabilities, policy, node_resolutions)?
217                .with_execution_weights(execution_weights)?
218                .with_completion_retention(completion_retention)?,
219        )?;
220        let untrusted_payload =
221            serde_json::to_value(&self.payload).map_err(|error| VNextError::Serialization {
222                context: "serialize unvalidated execution plan payload",
223                message: error.to_string(),
224            })?;
225        let rebuilt_payload =
226            serde_json::to_value(&rebuilt.payload).map_err(|error| VNextError::Serialization {
227                context: "serialize rebuilt execution plan payload",
228                message: error.to_string(),
229            })?;
230        if untrusted_payload != rebuilt_payload {
231            return Err(invalid_plan(
232                "untrusted plan differs from a semantic rebuild against current dependencies",
233            ));
234        }
235        if rebuilt.plan_hash != self.plan_hash {
236            return Err(VNextError::PlanHashMismatch {
237                expected: rebuilt.plan_hash.to_string(),
238                actual: self.plan_hash.to_string(),
239            });
240        }
241        Ok(rebuilt)
242    }
243}
ferrum_interfaces/vnext/execution/validation.rs

ferrum_interfaces/vnext/execution/
validation.rs
