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 layout.check_root()?;
106 let path = layout.root().join(WORKFLOW_FILE);
107 let _lock = JournalLock::acquire(&path)?;
108 let record: ExecutionWorkflowRecord = read_json(&path, MAX_EXECUTION_WORKFLOW_BYTES)?;
109 super::json::check_json_size(&record, MAX_EXECUTION_WORKFLOW_BYTES)?;
110 if &record.digest() != expected {
111 return Err(ExecutionWorkflowPersistenceError::DigestMismatch);
112 }
113 Ok(record)
114}
115
116pub(crate) fn read_execution_stage_occupancy(
124 layout: &BackupLayoutGuard,
125 expected: &ArtifactChecksumRecord,
126) -> Result<BTreeSet<u64>, ExecutionWorkflowPersistenceError> {
127 let workflow = read_execution_workflow(layout, expected)?;
128 let mut occupied = BTreeSet::new();
129 for stage in workflow.stages() {
130 let sequence = stage.operation_sequence();
131 match fs::symlink_metadata(stage_path(layout, sequence)) {
132 Ok(metadata) if metadata.is_dir() => {
133 occupied.insert(sequence);
134 }
135 Ok(_) => return Err(ExecutionWorkflowPersistenceError::UnsafeStage),
136 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
137 Err(error) => return Err(PersistenceError::from(error).into()),
138 }
139 }
140 read_execution_workflow(layout, expected)?;
141 Ok(occupied)
142}
143
144pub fn read_complete_execution_workflow(
157 layout: &BackupLayoutGuard,
158 expected: &ArtifactChecksumRecord,
159) -> Result<ExecutionWorkflowSettlementView, ExecutionStageCheckpointError> {
160 let workflow = read_execution_workflow(layout, expected)?;
161 let mut stages = Vec::with_capacity(workflow.stages().len());
162 let mut bindings = Vec::with_capacity(workflow.stages().len());
163 let mut by_sequence = BTreeMap::new();
164 for allocated in workflow.stages() {
165 let sequence = allocated.operation_sequence();
166 let stage = acquire_stage(layout, sequence)?;
167 let binding: ExecutionStageBindingRecord =
168 read_json(&stage.root().join(BINDING_FILE), MAX_EXECUTION_STAGE_BYTES)
169 .map_err(ExecutionWorkflowPersistenceError::from)?;
170 let (binding, plan) = read_binding(&stage, &workflow, sequence, &binding.digest())?;
171 let settlement = replay_execution_settlement(&stage, &plan.digest())?;
172 read_binding(&stage, &workflow, sequence, &binding.digest())?;
173 by_sequence.insert(sequence, stages.len());
174 stages.push(ExecutionStageSettlementView {
175 stage_sequence: sequence,
176 binding: binding.digest(),
177 settlement: settlement.digest(),
178 });
179 bindings.push(binding);
180 }
181 for binding in &bindings {
182 for predecessor in binding.predecessors() {
183 let Some(&index) = by_sequence.get(&predecessor.stage_sequence()) else {
184 return Err(ExecutionWorkflowPersistenceError::DigestMismatch.into());
185 };
186 let actual = &stages[index];
187 if predecessor.binding() != actual.binding()
188 || predecessor.settlement() != actual.settlement()
189 {
190 return Err(ExecutionWorkflowPersistenceError::DigestMismatch.into());
191 }
192 }
193 }
194 read_execution_workflow(layout, expected)?;
195 Ok(ExecutionWorkflowSettlementView {
196 workflow: expected.clone(),
197 stages,
198 })
199}
200
201#[derive(Debug)]
209pub struct ExecutionStageGuard<'a> {
210 workflow_layout: &'a BackupLayoutGuard,
211 stage_layout: BackupLayoutGuard,
212 binding: ExecutionStageBindingRecord,
213 plan: OperationPlanRecord,
214}
215impl<'a> ExecutionStageGuard<'a> {
216 pub fn reconcile_checkpoint(
226 &self,
227 learned_evidence: ArtifactChecksumRecord,
228 ) -> Result<ExecutionStagePredecessorRecord, ExecutionStageCheckpointError> {
229 self.reconcile_checkpoint_with(learned_evidence, || {})
230 }
231 fn reconcile_checkpoint_with(
232 &self,
233 learned_evidence: ArtifactChecksumRecord,
234 after_reconciliation: impl FnOnce(),
235 ) -> Result<ExecutionStagePredecessorRecord, ExecutionStageCheckpointError> {
236 let settlement = reconcile_execution_settlement(self.layout()?, &self.plan.digest())?;
237 after_reconciliation();
238 self.layout()?;
239 Ok(ExecutionStagePredecessorRecord::new(
240 self.binding.stage_sequence(),
241 self.binding.digest(),
242 settlement.digest(),
243 learned_evidence,
244 ))
245 }
246 pub fn replay_checkpoint(
255 &self,
256 learned_evidence: ArtifactChecksumRecord,
257 ) -> Result<ExecutionStagePredecessorRecord, ExecutionStageCheckpointError> {
258 let settlement = replay_execution_settlement(self.layout()?, &self.plan.digest())?;
259 self.layout()?;
260 Ok(ExecutionStagePredecessorRecord::new(
261 self.binding.stage_sequence(),
262 self.binding.digest(),
263 settlement.digest(),
264 learned_evidence,
265 ))
266 }
267 pub fn resume(
280 workflow_layout: &'a BackupLayoutGuard,
281 expected_workflow: &ArtifactChecksumRecord,
282 sequence: u64,
283 expected_binding: &ArtifactChecksumRecord,
284 ) -> Result<(Self, ExecutionProgressView), ExecutionStageResumeError> {
285 let stage = Self::open(
286 workflow_layout,
287 expected_workflow,
288 sequence,
289 expected_binding,
290 )?;
291 let progress = read_execution_progress(&stage.stage_layout, &stage.plan.digest())?;
292 stage.layout()?;
293 Ok((stage, progress))
294 }
295 pub fn prepare(
310 workflow_layout: &'a BackupLayoutGuard,
311 binding: ExecutionStageBindingRecord,
312 plan: OperationPlanRecord,
313 ) -> Result<Self, ExecutionStagePreparationError> {
314 Self::prepare_with(workflow_layout, binding, plan, |_| {})
315 }
316 fn prepare_with(
317 workflow_layout: &'a BackupLayoutGuard,
318 binding: ExecutionStageBindingRecord,
319 plan: OperationPlanRecord,
320 mut barrier: impl FnMut(StagePreparationBarrier),
321 ) -> Result<Self, ExecutionStagePreparationError> {
322 let authorities = plan.attempt_authorities()?;
323 let stage = Self::create_with(workflow_layout, binding, plan, &mut barrier)?;
324 let layout = stage.layout()?;
325 for authority in authorities {
326 let sequence = authority.binding().operation_sequence();
327 drop(AttemptJournalGuard::create(layout, authority)?);
328 barrier(StagePreparationBarrier::Journal(sequence));
329 }
330 stage.layout()?;
331 Ok(stage)
332 }
333 pub fn create(
342 workflow_layout: &'a BackupLayoutGuard,
343 binding: ExecutionStageBindingRecord,
344 plan: OperationPlanRecord,
345 ) -> Result<Self, ExecutionWorkflowPersistenceError> {
346 Self::create_with(workflow_layout, binding, plan, |_| {})
347 }
348 fn create_with(
349 workflow_layout: &'a BackupLayoutGuard,
350 binding: ExecutionStageBindingRecord,
351 plan: OperationPlanRecord,
352 mut barrier: impl FnMut(StagePreparationBarrier),
353 ) -> Result<Self, ExecutionWorkflowPersistenceError> {
354 let workflow = read_execution_workflow(workflow_layout, binding.workflow())?;
355 binding.validate(&workflow, &plan)?;
356 let binding_bytes = super::json::encode_json(&binding, MAX_EXECUTION_STAGE_BYTES)?;
357 let plan_bytes = super::json::encode_json(
359 &plan,
360 crate::model::operation_plan::MAX_OPERATION_PLAN_BYTES,
361 )?;
362 validate_predecessors(workflow_layout, &workflow, &binding)?;
363 let path = stage_path(workflow_layout, binding.stage_sequence());
364 ic_host_fs::durable::create_private_directory(&path)
365 .map_err(|error| PersistenceError::Io(crate::ops::directory_creation_io(error)))?;
366 let stage_layout = acquire_stage(workflow_layout, binding.stage_sequence())?;
367 super::operation_plan::create_encoded(&stage_layout, &plan_bytes)?;
368 barrier(StagePreparationBarrier::Plan);
369 super::json::create_encoded(&stage_layout.root().join(BINDING_FILE), &binding_bytes)?;
370 barrier(StagePreparationBarrier::Binding);
371 workflow_layout.check_root()?;
372 Ok(Self {
373 workflow_layout,
374 stage_layout,
375 binding,
376 plan,
377 })
378 }
379 pub fn open(
388 workflow_layout: &'a BackupLayoutGuard,
389 expected_workflow: &ArtifactChecksumRecord,
390 sequence: u64,
391 expected_binding: &ArtifactChecksumRecord,
392 ) -> Result<Self, ExecutionWorkflowPersistenceError> {
393 let workflow = read_execution_workflow(workflow_layout, expected_workflow)?;
394 workflow
395 .stage(sequence)
396 .map_err(|_| ExecutionWorkflowError::UnknownStage(sequence))?;
397 let stage_layout = acquire_stage(workflow_layout, sequence)?;
398 let (binding, plan) = read_binding(&stage_layout, &workflow, sequence, expected_binding)?;
399 validate_predecessors(workflow_layout, &workflow, &binding)?;
400 workflow_layout.check_root()?;
401 Ok(Self {
402 workflow_layout,
403 stage_layout,
404 binding,
405 plan,
406 })
407 }
408 #[must_use]
410 pub const fn plan(&self) -> &OperationPlanRecord {
411 &self.plan
412 }
413 #[must_use]
415 pub const fn binding(&self) -> &ExecutionStageBindingRecord {
416 &self.binding
417 }
418 pub fn layout(&self) -> Result<&BackupLayoutGuard, ExecutionWorkflowPersistenceError> {
425 let workflow = read_execution_workflow(self.workflow_layout, self.binding.workflow())?;
426 read_binding(
427 &self.stage_layout,
428 &workflow,
429 self.binding.stage_sequence(),
430 &self.binding.digest(),
431 )?;
432 validate_predecessors(self.workflow_layout, &workflow, &self.binding)?;
433 Ok(&self.stage_layout)
434 }
435}
436
437#[derive(Debug, Error)]
439pub enum ExecutionStageCheckpointError {
440 #[error(transparent)]
442 Stage(#[from] ExecutionWorkflowPersistenceError),
443 #[error(transparent)]
445 Settlement(#[from] ExecutionSettlementCheckpointError),
446}
447
448#[derive(Debug, Error)]
450pub enum ExecutionStagePreparationError {
451 #[error(transparent)]
453 Stage(#[from] ExecutionWorkflowPersistenceError),
454 #[error(transparent)]
456 Authority(#[from] OperationPlanError),
457 #[error(transparent)]
459 Journal(#[from] AttemptJournalError),
460}
461
462#[derive(Debug, Error)]
464pub enum ExecutionStageResumeError {
465 #[error(transparent)]
467 Stage(#[from] ExecutionWorkflowPersistenceError),
468 #[error(transparent)]
470 Progress(#[from] ExecutionProgressPersistenceError),
471}
472fn read_binding(
473 layout: &BackupLayoutGuard,
474 workflow: &ExecutionWorkflowRecord,
475 sequence: u64,
476 expected: &ArtifactChecksumRecord,
477) -> Result<(ExecutionStageBindingRecord, OperationPlanRecord), ExecutionWorkflowPersistenceError> {
478 layout.check_root()?;
479 let binding: ExecutionStageBindingRecord =
480 read_json(&layout.root().join(BINDING_FILE), MAX_EXECUTION_STAGE_BYTES)?;
481 super::json::check_json_size(&binding, MAX_EXECUTION_STAGE_BYTES)?;
482 if binding.stage_sequence() != sequence || &binding.digest() != expected {
483 return Err(ExecutionWorkflowPersistenceError::DigestMismatch);
484 }
485 let plan = read_operation_plan(layout, binding.plan())?;
486 binding.validate(workflow, &plan)?;
487 Ok((binding, plan))
488}
489fn validate_predecessors(
490 layout: &BackupLayoutGuard,
491 workflow: &ExecutionWorkflowRecord,
492 binding: &ExecutionStageBindingRecord,
493) -> Result<(), ExecutionWorkflowPersistenceError> {
494 let mut expected: BTreeMap<u64, ExecutionStagePredecessorRecord> = binding
498 .predecessors()
499 .iter()
500 .map(|row| (row.stage_sequence(), row.clone()))
501 .collect();
502 let mut pending: Vec<_> = expected.keys().copied().collect();
503 while let Some(sequence) = pending.pop() {
504 let row = &expected[&sequence];
505 let original = {
506 let predecessor = acquire_stage(layout, sequence)?;
509 let (original, _) = read_binding(&predecessor, workflow, sequence, row.binding())?;
510 read_execution_settlement(&predecessor, original.plan(), row.settlement())?;
511 original
512 };
513 for ancestor in original.predecessors() {
514 match expected.entry(ancestor.stage_sequence()) {
515 Entry::Vacant(entry) => {
516 pending.push(ancestor.stage_sequence());
517 entry.insert(ancestor.clone());
518 }
519 Entry::Occupied(entry) => {
520 if entry.get().binding() != ancestor.binding()
521 || entry.get().settlement() != ancestor.settlement()
522 {
523 return Err(ExecutionWorkflowPersistenceError::DigestMismatch);
524 }
525 }
526 }
527 }
528 }
529 Ok(())
530}
531fn stage_path(layout: &BackupLayoutGuard, sequence: u64) -> PathBuf {
532 layout.root().join(format!("execution-stage-{sequence}"))
533}
534fn acquire_stage(
535 layout: &BackupLayoutGuard,
536 sequence: u64,
537) -> Result<BackupLayoutGuard, ExecutionWorkflowPersistenceError> {
538 layout.check_root()?;
539 let path = stage_path(layout, sequence);
540 let metadata = fs::symlink_metadata(&path).map_err(PersistenceError::from)?;
541 if !metadata.is_dir() {
542 return Err(ExecutionWorkflowPersistenceError::UnsafeStage);
543 }
544 let stage = BackupLayoutGuard::acquire(&path)?;
545 if stage.root() != path {
548 return Err(ExecutionWorkflowPersistenceError::UnsafeStage);
549 }
550 layout.check_root()?;
551 Ok(stage)
552}
553#[derive(Debug, Error)]
555pub enum ExecutionWorkflowPersistenceError {
556 #[error("execution workflow/stage digest mismatch")]
558 DigestMismatch,
559 #[error("unsafe execution stage directory")]
561 UnsafeStage,
562 #[error(transparent)]
564 Model(#[from] ExecutionWorkflowError),
565 #[error(transparent)]
567 Plan(#[from] OperationPlanPersistenceError),
568 #[error(transparent)]
570 Settlement(#[from] ExecutionSettlementPersistenceError),
571 #[error(transparent)]
573 Lock(#[from] JournalLockError),
574 #[error(transparent)]
576 Persistence(#[from] PersistenceError),
577}
578
579#[cfg(all(test, unix))]
580mod tests;