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.append(self.allocation.digest().hash().as_bytes());
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 let stage = workflow
187 .stage(stage_sequence)
188 .map_err(|_| ExecutionWorkflowError::UnknownStage(stage_sequence))?;
189 if plan.context() != workflow.allocation.context()
190 || plan.inventory() != workflow.allocation.inventory()
191 || plan.selected_targets() != [stage.target()]
192 {
193 return Err(ExecutionWorkflowError::ChildIdentityMismatch);
194 }
195 if plan.budget().mutations() != stage.budget().mutations()
196 || plan.budget().observations() != stage.budget().observations()
197 {
198 return Err(ExecutionWorkflowError::ChildBudgetMismatch);
199 }
200 if plan.operations().iter().any(|operation| {
201 operation.budget().mutations() == 0 && operation.budget().observations() == 0
202 }) {
203 return Err(ExecutionWorkflowError::UnallocatedOperation);
204 }
205 canonical_predecessors(&mut predecessors)?;
206 let dependencies = workflow
207 .allocation
208 .graph()
209 .node(stage_sequence)
210 .map_err(|_| ExecutionWorkflowError::PredecessorMismatch)?
211 .depends_on();
212 if !predecessors
213 .iter()
214 .map(ExecutionStagePredecessorRecord::stage_sequence)
215 .eq(dependencies.iter().copied())
216 {
217 return Err(ExecutionWorkflowError::PredecessorMismatch);
218 }
219 Ok(Self {
220 version: 1,
221 workflow: workflow.digest(),
222 stage_sequence,
223 plan: plan.digest(),
224 predecessors,
225 })
226 }
227 pub fn validate(
231 &self,
232 workflow: &ExecutionWorkflowRecord,
233 plan: &OperationPlanRecord,
234 ) -> Result<(), ExecutionWorkflowError> {
235 if self.workflow != workflow.digest() || self.plan != plan.digest() {
236 return Err(ExecutionWorkflowError::BindingMismatch);
237 }
238 Self::new(
239 workflow,
240 self.stage_sequence,
241 plan,
242 self.predecessors.clone(),
243 )?;
244 Ok(())
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.append(self.workflow.hash().as_bytes());
272 bytes.append(&self.stage_sequence.to_be_bytes());
273 bytes.append(self.plan.hash().as_bytes());
274 bytes.append(&(self.predecessors.len() as u64).to_be_bytes());
275 for row in &self.predecessors {
276 bytes.append(&row.stage_sequence.to_be_bytes());
277 for digest in [&row.binding, &row.settlement, &row.learned_evidence] {
278 bytes.append(digest.hash().as_bytes());
279 }
280 }
281 bytes.finish()
282 }
283}
284fn canonical_predecessors(
285 rows: &mut [ExecutionStagePredecessorRecord],
286) -> Result<(), ExecutionWorkflowError> {
287 if rows.len() > MAX_EFFECT_DEPENDENCIES {
288 return Err(ExecutionWorkflowError::PredecessorMismatch);
289 }
290 rows.sort_by_key(ExecutionStagePredecessorRecord::stage_sequence);
291 if rows
292 .windows(2)
293 .any(|pair| pair[0].stage_sequence == pair[1].stage_sequence)
294 {
295 return Err(ExecutionWorkflowError::PredecessorMismatch);
296 }
297 Ok(())
298}
299fn bounded_predecessors<'de, D: Deserializer<'de>>(
300 deserializer: D,
301) -> Result<Vec<ExecutionStagePredecessorRecord>, D::Error> {
302 bounded_sequence::deserialize::<_, _, MAX_EFFECT_DEPENDENCIES>(
303 deserializer,
304 "at most 1,024 exact stage predecessors",
305 ExecutionWorkflowError::PredecessorMismatch,
306 )
307}
308#[derive(Debug, Error, Eq, PartialEq)]
310pub enum ExecutionWorkflowError {
311 #[error("unsupported execution workflow/stage version {0}")]
313 UnsupportedVersion(u16),
314 #[error("child plan differs from original stage identity")]
316 ChildIdentityMismatch,
317 #[error("child plan differs from original stage budget")]
319 ChildBudgetMismatch,
320 #[error("child plan contains an unallocated operation")]
322 UnallocatedOperation,
323 #[error("stage predecessors differ from original dependencies")]
325 PredecessorMismatch,
326 #[error("execution stage binding mismatch")]
328 BindingMismatch,
329 #[error("unknown execution stage {0}")]
331 UnknownStage(u64),
332}
333
334#[cfg(test)]
335pub(crate) mod tests;