ic_backup/model/execution_workflow/
mod.rs1use crate::model::evidence_digest::EvidenceDigest;
4use crate::model::{
5 artifacts::ArtifactChecksumRecord,
6 bounded_sequence,
7 effect_graph::MAX_EFFECT_DEPENDENCIES,
8 operation_plan::{OperationPlanError, OperationPlanRecord, PlannedOperationRecord},
9};
10use serde::{Deserialize, Deserializer, Serialize};
11use thiserror::Error;
12
13pub const MAX_EXECUTION_WORKFLOW_BYTES: u64 = 1024 * 1024;
15pub const MAX_EXECUTION_STAGE_BYTES: u64 = 512 * 1024;
17
18#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
28#[serde(try_from = "WorkflowFields")]
29pub struct ExecutionWorkflowRecord {
30 version: u16,
31 allocation: OperationPlanRecord,
32}
33#[derive(Deserialize)]
34#[serde(deny_unknown_fields)]
35struct WorkflowFields {
36 version: u16,
37 allocation: OperationPlanRecord,
38}
39impl TryFrom<WorkflowFields> for ExecutionWorkflowRecord {
40 type Error = ExecutionWorkflowError;
41 fn try_from(fields: WorkflowFields) -> Result<Self, Self::Error> {
42 if fields.version != 1 {
43 return Err(ExecutionWorkflowError::UnsupportedVersion(fields.version));
44 }
45 Ok(Self::new(fields.allocation))
46 }
47}
48impl ExecutionWorkflowRecord {
49 pub(crate) const fn allocation(&self) -> &OperationPlanRecord {
50 &self.allocation
51 }
52 #[must_use]
54 pub const fn new(allocation: OperationPlanRecord) -> Self {
55 Self {
56 version: 1,
57 allocation,
58 }
59 }
60 pub fn stage(&self, sequence: u64) -> Result<&PlannedOperationRecord, OperationPlanError> {
64 self.allocation.operation(sequence)
65 }
66 #[must_use]
68 pub fn stages(&self) -> &[PlannedOperationRecord] {
69 self.allocation.operations()
70 }
71 #[must_use]
73 pub fn digest(&self) -> ArtifactChecksumRecord {
74 let mut bytes = EvidenceDigest::new(b"ic-backup/execution-workflow/v1\0");
75 bytes.checksum(&self.allocation.digest());
76 bytes.finish()
77 }
78}
79
80#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
85#[serde(deny_unknown_fields)]
86pub struct ExecutionStagePredecessorRecord {
87 stage_sequence: u64,
88 binding: ArtifactChecksumRecord,
89 settlement: ArtifactChecksumRecord,
90 learned_evidence: ArtifactChecksumRecord,
91}
92impl ExecutionStagePredecessorRecord {
93 #[must_use]
95 pub const fn new(
96 stage_sequence: u64,
97 binding: ArtifactChecksumRecord,
98 settlement: ArtifactChecksumRecord,
99 learned_evidence: ArtifactChecksumRecord,
100 ) -> Self {
101 Self {
102 stage_sequence,
103 binding,
104 settlement,
105 learned_evidence,
106 }
107 }
108 #[must_use]
110 pub const fn stage_sequence(&self) -> u64 {
111 self.stage_sequence
112 }
113 #[must_use]
115 pub const fn binding(&self) -> &ArtifactChecksumRecord {
116 &self.binding
117 }
118 #[must_use]
120 pub const fn settlement(&self) -> &ArtifactChecksumRecord {
121 &self.settlement
122 }
123 #[must_use]
125 pub const fn learned_evidence(&self) -> &ArtifactChecksumRecord {
126 &self.learned_evidence
127 }
128}
129
130#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
137#[serde(try_from = "StageFields")]
138pub struct ExecutionStageBindingRecord {
139 version: u16,
140 workflow: ArtifactChecksumRecord,
141 stage_sequence: u64,
142 plan: ArtifactChecksumRecord,
143 predecessors: Vec<ExecutionStagePredecessorRecord>,
144}
145#[derive(Deserialize)]
146#[serde(deny_unknown_fields)]
147struct StageFields {
148 version: u16,
149 workflow: ArtifactChecksumRecord,
150 stage_sequence: u64,
151 plan: ArtifactChecksumRecord,
152 #[serde(deserialize_with = "bounded_predecessors")]
153 predecessors: Vec<ExecutionStagePredecessorRecord>,
154}
155impl TryFrom<StageFields> for ExecutionStageBindingRecord {
156 type Error = ExecutionWorkflowError;
157 fn try_from(mut fields: StageFields) -> Result<Self, Self::Error> {
158 if fields.version != 1 {
159 return Err(ExecutionWorkflowError::UnsupportedVersion(fields.version));
160 }
161 canonical_predecessors(&mut fields.predecessors)?;
162 Ok(Self {
163 version: 1,
164 workflow: fields.workflow,
165 stage_sequence: fields.stage_sequence,
166 plan: fields.plan,
167 predecessors: fields.predecessors,
168 })
169 }
170}
171impl ExecutionStageBindingRecord {
172 pub fn new(
181 workflow: &ExecutionWorkflowRecord,
182 stage_sequence: u64,
183 plan: &OperationPlanRecord,
184 mut predecessors: Vec<ExecutionStagePredecessorRecord>,
185 ) -> Result<Self, ExecutionWorkflowError> {
186 validate_child(workflow, stage_sequence, plan)?;
187 canonical_predecessors(&mut predecessors)?;
188 validate_predecessor_set(workflow, stage_sequence, &predecessors)?;
189 Ok(Self {
190 version: 1,
191 workflow: workflow.digest(),
192 stage_sequence,
193 plan: plan.digest(),
194 predecessors,
195 })
196 }
197 pub fn validate(
201 &self,
202 workflow: &ExecutionWorkflowRecord,
203 plan: &OperationPlanRecord,
204 ) -> Result<(), ExecutionWorkflowError> {
205 if self.workflow != workflow.digest() || self.plan != plan.digest() {
206 return Err(ExecutionWorkflowError::BindingMismatch);
207 }
208 validate_child(workflow, self.stage_sequence, plan)?;
211 validate_predecessor_set(workflow, self.stage_sequence, &self.predecessors)?;
212 Ok(())
213 }
214
215 pub(crate) fn matches_learned_predecessor(
218 &self,
219 plan: &OperationPlanRecord,
220 request: &str,
221 learned: &ArtifactChecksumRecord,
222 predecessors: &[ExecutionStagePredecessorRecord],
223 ) -> bool {
224 matches!(plan.operations(), [operation] if operation.request() == request)
225 && predecessors.iter().any(|row| {
226 row.stage_sequence() == self.stage_sequence
227 && row.binding() == &self.digest()
228 && row.learned_evidence() == learned
229 })
230 }
231
232 pub(crate) fn has_learned_evidence(
233 &self,
234 sequence: u64,
235 learned: &ArtifactChecksumRecord,
236 ) -> bool {
237 self.stage_sequence == sequence
238 && self
239 .predecessors
240 .iter()
241 .any(|row| row.learned_evidence() == learned)
242 }
243 #[must_use]
245 pub const fn workflow(&self) -> &ArtifactChecksumRecord {
246 &self.workflow
247 }
248 #[must_use]
250 pub const fn stage_sequence(&self) -> u64 {
251 self.stage_sequence
252 }
253 #[must_use]
255 pub const fn plan(&self) -> &ArtifactChecksumRecord {
256 &self.plan
257 }
258 #[must_use]
260 pub fn predecessors(&self) -> &[ExecutionStagePredecessorRecord] {
261 &self.predecessors
262 }
263 #[must_use]
266 pub fn digest(&self) -> ArtifactChecksumRecord {
267 let mut bytes = EvidenceDigest::new(b"ic-backup/execution-stage/v1\0");
268 bytes.checksum(&self.workflow);
269 bytes.u64_be(self.stage_sequence);
270 bytes.checksum(&self.plan);
271 bytes.u64_be(self.predecessors.len() as u64);
272 for row in &self.predecessors {
273 bytes.u64_be(row.stage_sequence);
274 for digest in [&row.binding, &row.settlement, &row.learned_evidence] {
275 bytes.checksum(digest);
276 }
277 }
278 bytes.finish()
279 }
280}
281fn validate_child(
282 workflow: &ExecutionWorkflowRecord,
283 stage_sequence: u64,
284 plan: &OperationPlanRecord,
285) -> Result<(), ExecutionWorkflowError> {
286 let stage = workflow
287 .stage(stage_sequence)
288 .map_err(|_| ExecutionWorkflowError::UnknownStage(stage_sequence))?;
289 if plan.context() != workflow.allocation.context()
290 || plan.inventory() != workflow.allocation.inventory()
291 || plan.selected_targets() != [stage.target()]
292 {
293 return Err(ExecutionWorkflowError::ChildIdentityMismatch);
294 }
295 if plan.budget().mutations() != stage.budget().mutations()
296 || plan.budget().observations() != stage.budget().observations()
297 {
298 return Err(ExecutionWorkflowError::ChildBudgetMismatch);
299 }
300 if plan.operations().iter().any(|operation| {
301 operation.budget().mutations() == 0 && operation.budget().observations() == 0
302 }) {
303 return Err(ExecutionWorkflowError::UnallocatedOperation);
304 }
305 Ok(())
306}
307
308fn validate_predecessor_set(
309 workflow: &ExecutionWorkflowRecord,
310 stage_sequence: u64,
311 predecessors: &[ExecutionStagePredecessorRecord],
312) -> Result<(), ExecutionWorkflowError> {
313 let dependencies = workflow
314 .allocation
315 .graph()
316 .node(stage_sequence)
317 .map_err(|_| ExecutionWorkflowError::PredecessorMismatch)?
318 .depends_on();
319 if !predecessors
320 .iter()
321 .map(ExecutionStagePredecessorRecord::stage_sequence)
322 .eq(dependencies.iter().copied())
323 {
324 return Err(ExecutionWorkflowError::PredecessorMismatch);
325 }
326 Ok(())
327}
328
329fn canonical_predecessors(
330 rows: &mut [ExecutionStagePredecessorRecord],
331) -> Result<(), ExecutionWorkflowError> {
332 if rows.len() > MAX_EFFECT_DEPENDENCIES {
333 return Err(ExecutionWorkflowError::PredecessorMismatch);
334 }
335 rows.sort_by_key(ExecutionStagePredecessorRecord::stage_sequence);
336 if rows
337 .windows(2)
338 .any(|pair| pair[0].stage_sequence == pair[1].stage_sequence)
339 {
340 return Err(ExecutionWorkflowError::PredecessorMismatch);
341 }
342 Ok(())
343}
344fn bounded_predecessors<'de, D: Deserializer<'de>>(
345 deserializer: D,
346) -> Result<Vec<ExecutionStagePredecessorRecord>, D::Error> {
347 bounded_sequence::deserialize::<_, _, MAX_EFFECT_DEPENDENCIES>(
348 deserializer,
349 "at most 1,024 exact stage predecessors",
350 ExecutionWorkflowError::PredecessorMismatch,
351 )
352}
353#[derive(Debug, Error, Eq, PartialEq)]
355pub enum ExecutionWorkflowError {
356 #[error("unsupported execution workflow/stage version {0}")]
358 UnsupportedVersion(u16),
359 #[error("child plan differs from original stage identity")]
361 ChildIdentityMismatch,
362 #[error("child plan differs from original stage budget")]
364 ChildBudgetMismatch,
365 #[error("child plan contains an unallocated operation")]
367 UnallocatedOperation,
368 #[error("stage predecessors differ from original dependencies")]
370 PredecessorMismatch,
371 #[error("execution stage binding mismatch")]
373 BindingMismatch,
374 #[error("unknown execution stage {0}")]
376 UnknownStage(u64),
377}
378
379#[cfg(test)]
380mod tests;