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 #[must_use]
52 pub const fn allocation(&self) -> &OperationPlanRecord {
53 &self.allocation
54 }
55 #[must_use]
57 pub const fn new(allocation: OperationPlanRecord) -> Self {
58 Self {
59 version: 1,
60 allocation,
61 }
62 }
63 pub fn stage(&self, sequence: u64) -> Result<&PlannedOperationRecord, OperationPlanError> {
67 self.allocation.operation(sequence)
68 }
69 #[must_use]
71 pub fn stages(&self) -> &[PlannedOperationRecord] {
72 self.allocation.operations()
73 }
74 #[must_use]
76 pub fn digest(&self) -> ArtifactChecksumRecord {
77 let mut bytes = EvidenceDigest::new(b"ic-backup/execution-workflow/v1\0");
78 bytes.checksum(&self.allocation.digest());
79 bytes.finish()
80 }
81}
82
83#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
88#[serde(deny_unknown_fields)]
89pub struct ExecutionStagePredecessorRecord {
90 stage_sequence: u64,
91 binding: ArtifactChecksumRecord,
92 settlement: ArtifactChecksumRecord,
93 learned_evidence: ArtifactChecksumRecord,
94}
95impl ExecutionStagePredecessorRecord {
96 #[must_use]
98 pub const fn new(
99 stage_sequence: u64,
100 binding: ArtifactChecksumRecord,
101 settlement: ArtifactChecksumRecord,
102 learned_evidence: ArtifactChecksumRecord,
103 ) -> Self {
104 Self {
105 stage_sequence,
106 binding,
107 settlement,
108 learned_evidence,
109 }
110 }
111 #[must_use]
113 pub const fn stage_sequence(&self) -> u64 {
114 self.stage_sequence
115 }
116 #[must_use]
118 pub const fn binding(&self) -> &ArtifactChecksumRecord {
119 &self.binding
120 }
121 #[must_use]
123 pub const fn settlement(&self) -> &ArtifactChecksumRecord {
124 &self.settlement
125 }
126 #[must_use]
128 pub const fn learned_evidence(&self) -> &ArtifactChecksumRecord {
129 &self.learned_evidence
130 }
131}
132
133#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
140#[serde(try_from = "StageFields")]
141pub struct ExecutionStageBindingRecord {
142 version: u16,
143 workflow: ArtifactChecksumRecord,
144 stage_sequence: u64,
145 plan: ArtifactChecksumRecord,
146 predecessors: Vec<ExecutionStagePredecessorRecord>,
147}
148#[derive(Deserialize)]
149#[serde(deny_unknown_fields)]
150struct StageFields {
151 version: u16,
152 workflow: ArtifactChecksumRecord,
153 stage_sequence: u64,
154 plan: ArtifactChecksumRecord,
155 #[serde(deserialize_with = "bounded_predecessors")]
156 predecessors: Vec<ExecutionStagePredecessorRecord>,
157}
158impl TryFrom<StageFields> for ExecutionStageBindingRecord {
159 type Error = ExecutionWorkflowError;
160 fn try_from(mut fields: StageFields) -> Result<Self, Self::Error> {
161 if fields.version != 1 {
162 return Err(ExecutionWorkflowError::UnsupportedVersion(fields.version));
163 }
164 canonical_predecessors(&mut fields.predecessors)?;
165 Ok(Self {
166 version: 1,
167 workflow: fields.workflow,
168 stage_sequence: fields.stage_sequence,
169 plan: fields.plan,
170 predecessors: fields.predecessors,
171 })
172 }
173}
174impl ExecutionStageBindingRecord {
175 pub fn new(
184 workflow: &ExecutionWorkflowRecord,
185 stage_sequence: u64,
186 plan: &OperationPlanRecord,
187 mut predecessors: Vec<ExecutionStagePredecessorRecord>,
188 ) -> Result<Self, ExecutionWorkflowError> {
189 validate_child(workflow, stage_sequence, plan)?;
190 canonical_predecessors(&mut predecessors)?;
191 validate_predecessor_set(workflow, stage_sequence, &predecessors)?;
192 Ok(Self {
193 version: 1,
194 workflow: workflow.digest(),
195 stage_sequence,
196 plan: plan.digest(),
197 predecessors,
198 })
199 }
200 pub fn validate(
204 &self,
205 workflow: &ExecutionWorkflowRecord,
206 plan: &OperationPlanRecord,
207 ) -> Result<(), ExecutionWorkflowError> {
208 if self.workflow != workflow.digest() || self.plan != plan.digest() {
209 return Err(ExecutionWorkflowError::BindingMismatch);
210 }
211 validate_child(workflow, self.stage_sequence, plan)?;
214 validate_predecessor_set(workflow, self.stage_sequence, &self.predecessors)?;
215 Ok(())
216 }
217
218 pub(crate) fn matches_learned_predecessor(
221 &self,
222 plan: &OperationPlanRecord,
223 request: &str,
224 learned: &ArtifactChecksumRecord,
225 predecessors: &[ExecutionStagePredecessorRecord],
226 ) -> bool {
227 matches!(plan.operations(), [operation] if operation.request() == request)
228 && predecessors.iter().any(|row| {
229 row.stage_sequence() == self.stage_sequence
230 && row.binding() == &self.digest()
231 && row.learned_evidence() == learned
232 })
233 }
234
235 pub(crate) fn has_learned_evidence(
236 &self,
237 sequence: u64,
238 learned: &ArtifactChecksumRecord,
239 ) -> bool {
240 self.stage_sequence == sequence
241 && self
242 .predecessors
243 .iter()
244 .any(|row| row.learned_evidence() == learned)
245 }
246 #[must_use]
248 pub const fn workflow(&self) -> &ArtifactChecksumRecord {
249 &self.workflow
250 }
251 #[must_use]
253 pub const fn stage_sequence(&self) -> u64 {
254 self.stage_sequence
255 }
256 #[must_use]
258 pub const fn plan(&self) -> &ArtifactChecksumRecord {
259 &self.plan
260 }
261 #[must_use]
263 pub fn predecessors(&self) -> &[ExecutionStagePredecessorRecord] {
264 &self.predecessors
265 }
266 #[must_use]
269 pub fn digest(&self) -> ArtifactChecksumRecord {
270 let mut bytes = EvidenceDigest::new(b"ic-backup/execution-stage/v1\0");
271 bytes.checksum(&self.workflow);
272 bytes.u64_be(self.stage_sequence);
273 bytes.checksum(&self.plan);
274 bytes.u64_be(self.predecessors.len() as u64);
275 for row in &self.predecessors {
276 bytes.u64_be(row.stage_sequence);
277 for digest in [&row.binding, &row.settlement, &row.learned_evidence] {
278 bytes.checksum(digest);
279 }
280 }
281 bytes.finish()
282 }
283}
284fn validate_child(
285 workflow: &ExecutionWorkflowRecord,
286 stage_sequence: u64,
287 plan: &OperationPlanRecord,
288) -> Result<(), ExecutionWorkflowError> {
289 let stage = workflow
290 .stage(stage_sequence)
291 .map_err(|_| ExecutionWorkflowError::UnknownStage(stage_sequence))?;
292 if plan.context() != workflow.allocation.context()
293 || plan.inventory() != workflow.allocation.inventory()
294 || plan.selected_targets() != [stage.target()]
295 {
296 return Err(ExecutionWorkflowError::ChildIdentityMismatch);
297 }
298 if plan.budget().mutations() != stage.budget().mutations()
299 || plan.budget().observations() != stage.budget().observations()
300 {
301 return Err(ExecutionWorkflowError::ChildBudgetMismatch);
302 }
303 if plan.operations().iter().any(|operation| {
304 operation.budget().mutations() == 0 && operation.budget().observations() == 0
305 }) {
306 return Err(ExecutionWorkflowError::UnallocatedOperation);
307 }
308 Ok(())
309}
310
311fn validate_predecessor_set(
312 workflow: &ExecutionWorkflowRecord,
313 stage_sequence: u64,
314 predecessors: &[ExecutionStagePredecessorRecord],
315) -> Result<(), ExecutionWorkflowError> {
316 let dependencies = workflow
317 .allocation
318 .graph()
319 .node(stage_sequence)
320 .map_err(|_| ExecutionWorkflowError::PredecessorMismatch)?
321 .depends_on();
322 if !predecessors
323 .iter()
324 .map(ExecutionStagePredecessorRecord::stage_sequence)
325 .eq(dependencies.iter().copied())
326 {
327 return Err(ExecutionWorkflowError::PredecessorMismatch);
328 }
329 Ok(())
330}
331
332fn canonical_predecessors(
333 rows: &mut [ExecutionStagePredecessorRecord],
334) -> Result<(), ExecutionWorkflowError> {
335 if rows.len() > MAX_EFFECT_DEPENDENCIES {
336 return Err(ExecutionWorkflowError::PredecessorMismatch);
337 }
338 rows.sort_by_key(ExecutionStagePredecessorRecord::stage_sequence);
339 if rows
340 .windows(2)
341 .any(|pair| pair[0].stage_sequence == pair[1].stage_sequence)
342 {
343 return Err(ExecutionWorkflowError::PredecessorMismatch);
344 }
345 Ok(())
346}
347fn bounded_predecessors<'de, D: Deserializer<'de>>(
348 deserializer: D,
349) -> Result<Vec<ExecutionStagePredecessorRecord>, D::Error> {
350 bounded_sequence::deserialize::<_, _, MAX_EFFECT_DEPENDENCIES>(
351 deserializer,
352 "at most 1,024 exact stage predecessors",
353 ExecutionWorkflowError::PredecessorMismatch,
354 )
355}
356#[derive(Debug, Error, Eq, PartialEq)]
358pub enum ExecutionWorkflowError {
359 #[error("unsupported execution workflow/stage version {0}")]
361 UnsupportedVersion(u16),
362 #[error("child plan differs from original stage identity")]
364 ChildIdentityMismatch,
365 #[error("child plan differs from original stage budget")]
367 ChildBudgetMismatch,
368 #[error("child plan contains an unallocated operation")]
370 UnallocatedOperation,
371 #[error("stage predecessors differ from original dependencies")]
373 PredecessorMismatch,
374 #[error("execution stage binding mismatch")]
376 BindingMismatch,
377 #[error("unknown execution stage {0}")]
379 UnknownStage(u64),
380}
381
382#[cfg(test)]
383mod tests;