1use super::{
4 AttemptJournalError, AttemptJournalGuard, BackupLayoutGuard, ExecutionProgressPersistenceError,
5 ExecutionSettlementCheckpointError, ExecutionSettlementPersistenceError, JournalLock,
6 JournalLockError, OperationPlanPersistenceError, PersistenceError, read_execution_progress,
7 read_execution_settlement, read_json, read_operation_plan, reconcile_execution_settlement,
8 replay_execution_settlement,
9};
10use crate::model::{
11 artifacts::ArtifactChecksumRecord,
12 execution_workflow::{
13 ExecutionStageBindingRecord, ExecutionStagePredecessorRecord, ExecutionWorkflowError,
14 ExecutionWorkflowRecord, MAX_EXECUTION_STAGE_BYTES, MAX_EXECUTION_WORKFLOW_BYTES,
15 },
16 operation_plan::{OperationPlanError, OperationPlanRecord},
17};
18use crate::policy::execution_progress::ExecutionProgressView;
19use std::{
20 collections::{BTreeMap, BTreeSet, btree_map::Entry},
21 fs,
22 path::PathBuf,
23};
24use thiserror::Error;
25
26const WORKFLOW_FILE: &str = "execution-workflow.json";
27const BINDING_FILE: &str = "stage-binding.json";
28
29#[derive(Clone, Debug, Eq, PartialEq)]
35pub struct ExecutionWorkflowSettlementView {
36 workflow: ArtifactChecksumRecord,
37 stages: Vec<ExecutionStageSettlementView>,
38}
39impl ExecutionWorkflowSettlementView {
40 #[must_use]
42 pub const fn workflow(&self) -> &ArtifactChecksumRecord {
43 &self.workflow
44 }
45 #[must_use]
47 pub fn stages(&self) -> &[ExecutionStageSettlementView] {
48 &self.stages
49 }
50}
51
52#[derive(Clone, Debug, Eq, PartialEq)]
54pub struct ExecutionStageSettlementView {
55 stage_sequence: u64,
56 binding: ArtifactChecksumRecord,
57 settlement: ArtifactChecksumRecord,
58}
59impl ExecutionStageSettlementView {
60 #[must_use]
62 pub const fn stage_sequence(&self) -> u64 {
63 self.stage_sequence
64 }
65 #[must_use]
67 pub const fn binding(&self) -> &ArtifactChecksumRecord {
68 &self.binding
69 }
70 #[must_use]
72 pub const fn settlement(&self) -> &ArtifactChecksumRecord {
73 &self.settlement
74 }
75}
76
77#[derive(Clone, Copy, Debug, Eq, PartialEq)]
78enum StagePreparationBarrier {
79 Plan,
80 Binding,
81 Journal(u64),
82}
83
84pub fn create_execution_workflow(
88 layout: &BackupLayoutGuard,
89 record: &ExecutionWorkflowRecord,
90) -> Result<(), ExecutionWorkflowPersistenceError> {
91 layout.check_root()?;
92 let bytes = super::json::encode_json(record, MAX_EXECUTION_WORKFLOW_BYTES)?;
93 let path = layout.root().join(WORKFLOW_FILE);
94 let _lock = JournalLock::acquire(&path)?;
95 super::json::create_encoded(&path, &bytes)?;
96 Ok(())
97}
98pub fn read_execution_workflow(
102 layout: &BackupLayoutGuard,
103 expected: &ArtifactChecksumRecord,
104) -> Result<ExecutionWorkflowRecord, ExecutionWorkflowPersistenceError> {
105 super::immutable_record::read(
106 layout,
107 WORKFLOW_FILE,
108 MAX_EXECUTION_WORKFLOW_BYTES,
109 |record: &ExecutionWorkflowRecord| &record.digest() == expected,
110 || ExecutionWorkflowPersistenceError::DigestMismatch,
111 )
112}
113
114pub(crate) fn read_execution_stage_occupancy(
122 layout: &BackupLayoutGuard,
123 expected: &ArtifactChecksumRecord,
124) -> Result<BTreeSet<u64>, ExecutionWorkflowPersistenceError> {
125 let workflow = read_execution_workflow(layout, expected)?;
126 let mut occupied = BTreeSet::new();
127 for stage in workflow.stages() {
128 let sequence = stage.operation_sequence();
129 match fs::symlink_metadata(stage_path(layout, sequence)) {
130 Ok(metadata) if metadata.is_dir() => {
131 occupied.insert(sequence);
132 }
133 Ok(_) => return Err(ExecutionWorkflowPersistenceError::UnsafeStage),
134 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
135 Err(error) => return Err(PersistenceError::from(error).into()),
136 }
137 }
138 read_execution_workflow(layout, expected)?;
139 Ok(occupied)
140}
141
142pub fn read_complete_execution_workflow(
155 layout: &BackupLayoutGuard,
156 expected: &ArtifactChecksumRecord,
157) -> Result<ExecutionWorkflowSettlementView, ExecutionStageCheckpointError> {
158 let workflow = read_execution_workflow(layout, expected)?;
159 let mut stages = Vec::with_capacity(workflow.stages().len());
160 let mut bindings = Vec::with_capacity(workflow.stages().len());
161 let mut by_sequence = BTreeMap::new();
162 for allocated in workflow.stages() {
163 let sequence = allocated.operation_sequence();
164 let stage = acquire_stage(layout, sequence)?;
165 let binding: ExecutionStageBindingRecord =
166 read_json(&stage.root().join(BINDING_FILE), MAX_EXECUTION_STAGE_BYTES)
167 .map_err(ExecutionWorkflowPersistenceError::from)?;
168 let (binding, plan) = read_binding(&stage, &workflow, sequence, &binding.digest())?;
169 let settlement = replay_execution_settlement(&stage, &plan.digest())?;
170 read_binding(&stage, &workflow, sequence, &binding.digest())?;
171 by_sequence.insert(sequence, stages.len());
172 stages.push(ExecutionStageSettlementView {
173 stage_sequence: sequence,
174 binding: binding.digest(),
175 settlement: settlement.digest(),
176 });
177 bindings.push(binding);
178 }
179 for binding in &bindings {
180 for predecessor in binding.predecessors() {
181 let Some(&index) = by_sequence.get(&predecessor.stage_sequence()) else {
182 return Err(ExecutionWorkflowPersistenceError::DigestMismatch.into());
183 };
184 let actual = &stages[index];
185 if predecessor.binding() != actual.binding()
186 || predecessor.settlement() != actual.settlement()
187 {
188 return Err(ExecutionWorkflowPersistenceError::DigestMismatch.into());
189 }
190 }
191 }
192 read_execution_workflow(layout, expected)?;
193 Ok(ExecutionWorkflowSettlementView {
194 workflow: expected.clone(),
195 stages,
196 })
197}
198
199#[derive(Debug)]
207pub struct ExecutionStageGuard<'a> {
208 workflow_layout: &'a BackupLayoutGuard,
209 stage_layout: BackupLayoutGuard,
210 binding: ExecutionStageBindingRecord,
211 plan: OperationPlanRecord,
212}
213impl<'a> ExecutionStageGuard<'a> {
214 pub fn reconcile_checkpoint(
224 &self,
225 learned_evidence: ArtifactChecksumRecord,
226 ) -> Result<ExecutionStagePredecessorRecord, ExecutionStageCheckpointError> {
227 self.reconcile_checkpoint_with(learned_evidence, || {})
228 }
229 fn reconcile_checkpoint_with(
230 &self,
231 learned_evidence: ArtifactChecksumRecord,
232 after_reconciliation: impl FnOnce(),
233 ) -> Result<ExecutionStagePredecessorRecord, ExecutionStageCheckpointError> {
234 let settlement = reconcile_execution_settlement(self.layout()?, &self.plan.digest())?;
235 after_reconciliation();
236 self.layout()?;
237 Ok(ExecutionStagePredecessorRecord::new(
238 self.binding.stage_sequence(),
239 self.binding.digest(),
240 settlement.digest(),
241 learned_evidence,
242 ))
243 }
244 pub fn replay_checkpoint(
253 &self,
254 learned_evidence: ArtifactChecksumRecord,
255 ) -> Result<ExecutionStagePredecessorRecord, ExecutionStageCheckpointError> {
256 let settlement = replay_execution_settlement(self.layout()?, &self.plan.digest())?;
257 self.layout()?;
258 Ok(ExecutionStagePredecessorRecord::new(
259 self.binding.stage_sequence(),
260 self.binding.digest(),
261 settlement.digest(),
262 learned_evidence,
263 ))
264 }
265 pub fn resume(
278 workflow_layout: &'a BackupLayoutGuard,
279 expected_workflow: &ArtifactChecksumRecord,
280 sequence: u64,
281 expected_binding: &ArtifactChecksumRecord,
282 ) -> Result<(Self, ExecutionProgressView), ExecutionStageResumeError> {
283 let stage = Self::open(
284 workflow_layout,
285 expected_workflow,
286 sequence,
287 expected_binding,
288 )?;
289 let progress = read_execution_progress(&stage.stage_layout, &stage.plan.digest())?;
290 stage.layout()?;
291 Ok((stage, progress))
292 }
293 pub fn prepare(
308 workflow_layout: &'a BackupLayoutGuard,
309 binding: ExecutionStageBindingRecord,
310 plan: OperationPlanRecord,
311 ) -> Result<Self, ExecutionStagePreparationError> {
312 Self::prepare_with(workflow_layout, binding, plan, |_| {})
313 }
314 fn prepare_with(
315 workflow_layout: &'a BackupLayoutGuard,
316 binding: ExecutionStageBindingRecord,
317 plan: OperationPlanRecord,
318 mut barrier: impl FnMut(StagePreparationBarrier),
319 ) -> Result<Self, ExecutionStagePreparationError> {
320 let authorities = plan.attempt_authorities()?;
321 let stage = Self::create_with(workflow_layout, binding, plan, &mut barrier)?;
322 let layout = stage.layout()?;
323 for authority in authorities {
324 let sequence = authority.binding().operation_sequence();
325 drop(AttemptJournalGuard::create(layout, authority)?);
326 barrier(StagePreparationBarrier::Journal(sequence));
327 }
328 stage.layout()?;
329 Ok(stage)
330 }
331 pub fn create(
340 workflow_layout: &'a BackupLayoutGuard,
341 binding: ExecutionStageBindingRecord,
342 plan: OperationPlanRecord,
343 ) -> Result<Self, ExecutionWorkflowPersistenceError> {
344 Self::create_with(workflow_layout, binding, plan, |_| {})
345 }
346 fn create_with(
347 workflow_layout: &'a BackupLayoutGuard,
348 binding: ExecutionStageBindingRecord,
349 plan: OperationPlanRecord,
350 mut barrier: impl FnMut(StagePreparationBarrier),
351 ) -> Result<Self, ExecutionWorkflowPersistenceError> {
352 let workflow = read_execution_workflow(workflow_layout, binding.workflow())?;
353 binding.validate(&workflow, &plan)?;
354 let binding_bytes = super::json::encode_json(&binding, MAX_EXECUTION_STAGE_BYTES)?;
355 let plan_bytes = super::json::encode_json(
357 &plan,
358 crate::model::operation_plan::MAX_OPERATION_PLAN_BYTES,
359 )?;
360 validate_predecessors(workflow_layout, &workflow, &binding)?;
361 let path = stage_path(workflow_layout, binding.stage_sequence());
362 ic_host_fs::durable::create_private_directory(&path)
363 .map_err(|error| PersistenceError::Io(crate::ops::directory_creation_io(error)))?;
364 let stage_layout = acquire_stage(workflow_layout, binding.stage_sequence())?;
365 super::operation_plan::create_encoded(&stage_layout, &plan_bytes)?;
366 barrier(StagePreparationBarrier::Plan);
367 super::json::create_encoded(&stage_layout.root().join(BINDING_FILE), &binding_bytes)?;
368 barrier(StagePreparationBarrier::Binding);
369 workflow_layout.check_root()?;
370 Ok(Self {
371 workflow_layout,
372 stage_layout,
373 binding,
374 plan,
375 })
376 }
377 pub fn open(
386 workflow_layout: &'a BackupLayoutGuard,
387 expected_workflow: &ArtifactChecksumRecord,
388 sequence: u64,
389 expected_binding: &ArtifactChecksumRecord,
390 ) -> Result<Self, ExecutionWorkflowPersistenceError> {
391 let workflow = read_execution_workflow(workflow_layout, expected_workflow)?;
392 workflow
393 .stage(sequence)
394 .map_err(|_| ExecutionWorkflowError::UnknownStage(sequence))?;
395 let stage_layout = acquire_stage(workflow_layout, sequence)?;
396 let (binding, plan) = read_binding(&stage_layout, &workflow, sequence, expected_binding)?;
397 validate_predecessors(workflow_layout, &workflow, &binding)?;
398 workflow_layout.check_root()?;
399 Ok(Self {
400 workflow_layout,
401 stage_layout,
402 binding,
403 plan,
404 })
405 }
406 #[must_use]
408 pub const fn plan(&self) -> &OperationPlanRecord {
409 &self.plan
410 }
411 #[must_use]
413 pub const fn binding(&self) -> &ExecutionStageBindingRecord {
414 &self.binding
415 }
416 pub fn layout(&self) -> Result<&BackupLayoutGuard, ExecutionWorkflowPersistenceError> {
423 let workflow = read_execution_workflow(self.workflow_layout, self.binding.workflow())?;
424 read_binding(
425 &self.stage_layout,
426 &workflow,
427 self.binding.stage_sequence(),
428 &self.binding.digest(),
429 )?;
430 validate_predecessors(self.workflow_layout, &workflow, &self.binding)?;
431 Ok(&self.stage_layout)
432 }
433}
434
435#[derive(Debug, Error)]
437pub enum ExecutionStageCheckpointError {
438 #[error(transparent)]
440 Stage(#[from] ExecutionWorkflowPersistenceError),
441 #[error(transparent)]
443 Settlement(#[from] ExecutionSettlementCheckpointError),
444}
445
446#[derive(Debug, Error)]
448pub enum ExecutionStagePreparationError {
449 #[error(transparent)]
451 Stage(#[from] ExecutionWorkflowPersistenceError),
452 #[error(transparent)]
454 Authority(#[from] OperationPlanError),
455 #[error(transparent)]
457 Journal(#[from] AttemptJournalError),
458}
459
460#[derive(Debug, Error)]
462pub enum ExecutionStageResumeError {
463 #[error(transparent)]
465 Stage(#[from] ExecutionWorkflowPersistenceError),
466 #[error(transparent)]
468 Progress(#[from] ExecutionProgressPersistenceError),
469}
470fn read_binding(
471 layout: &BackupLayoutGuard,
472 workflow: &ExecutionWorkflowRecord,
473 sequence: u64,
474 expected: &ArtifactChecksumRecord,
475) -> Result<(ExecutionStageBindingRecord, OperationPlanRecord), ExecutionWorkflowPersistenceError> {
476 layout.check_root()?;
477 let binding: ExecutionStageBindingRecord =
478 read_json(&layout.root().join(BINDING_FILE), MAX_EXECUTION_STAGE_BYTES)?;
479 super::json::check_json_size(&binding, MAX_EXECUTION_STAGE_BYTES)?;
480 if binding.stage_sequence() != sequence || &binding.digest() != expected {
481 return Err(ExecutionWorkflowPersistenceError::DigestMismatch);
482 }
483 let plan = read_operation_plan(layout, binding.plan())?;
484 binding.validate(workflow, &plan)?;
485 Ok((binding, plan))
486}
487fn validate_predecessors(
488 layout: &BackupLayoutGuard,
489 workflow: &ExecutionWorkflowRecord,
490 binding: &ExecutionStageBindingRecord,
491) -> Result<(), ExecutionWorkflowPersistenceError> {
492 let mut expected: BTreeMap<u64, ExecutionStagePredecessorRecord> = binding
496 .predecessors()
497 .iter()
498 .map(|row| (row.stage_sequence(), row.clone()))
499 .collect();
500 let mut pending: Vec<_> = expected.keys().copied().collect();
501 while let Some(sequence) = pending.pop() {
502 let row = &expected[&sequence];
503 let original = {
504 let predecessor = acquire_stage(layout, sequence)?;
507 let (original, _) = read_binding(&predecessor, workflow, sequence, row.binding())?;
508 read_execution_settlement(&predecessor, original.plan(), row.settlement())?;
509 original
510 };
511 for ancestor in original.predecessors() {
512 match expected.entry(ancestor.stage_sequence()) {
513 Entry::Vacant(entry) => {
514 pending.push(ancestor.stage_sequence());
515 entry.insert(ancestor.clone());
516 }
517 Entry::Occupied(entry) => {
518 if entry.get().binding() != ancestor.binding()
519 || entry.get().settlement() != ancestor.settlement()
520 {
521 return Err(ExecutionWorkflowPersistenceError::DigestMismatch);
522 }
523 }
524 }
525 }
526 }
527 Ok(())
528}
529fn stage_path(layout: &BackupLayoutGuard, sequence: u64) -> PathBuf {
530 layout.root().join(format!("execution-stage-{sequence}"))
531}
532fn acquire_stage(
533 layout: &BackupLayoutGuard,
534 sequence: u64,
535) -> Result<BackupLayoutGuard, ExecutionWorkflowPersistenceError> {
536 layout.check_root()?;
537 let path = stage_path(layout, sequence);
538 let metadata = fs::symlink_metadata(&path).map_err(PersistenceError::from)?;
539 if !metadata.is_dir() {
540 return Err(ExecutionWorkflowPersistenceError::UnsafeStage);
541 }
542 let stage = BackupLayoutGuard::acquire(&path)?;
543 if stage.root() != path {
546 return Err(ExecutionWorkflowPersistenceError::UnsafeStage);
547 }
548 layout.check_root()?;
549 Ok(stage)
550}
551#[derive(Debug, Error)]
553pub enum ExecutionWorkflowPersistenceError {
554 #[error("execution workflow/stage digest mismatch")]
556 DigestMismatch,
557 #[error("unsafe execution stage directory")]
559 UnsafeStage,
560 #[error(transparent)]
562 Model(#[from] ExecutionWorkflowError),
563 #[error(transparent)]
565 Plan(#[from] OperationPlanPersistenceError),
566 #[error(transparent)]
568 Settlement(#[from] ExecutionSettlementPersistenceError),
569 #[error(transparent)]
571 Lock(#[from] JournalLockError),
572 #[error(transparent)]
574 Persistence(#[from] PersistenceError),
575}
576
577#[cfg(all(test, unix))]
578mod tests;