ic_backup/ops/persistence/execution_workflow/
mod.rs

1//! Fixed immutable stage admission using the existing layout, plan and settlement owners.
2
3use 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/// Exact local identities of every original stage's retained checkpoint.
30///
31/// This is a sequential local projection, not an immutable product completion
32/// receipt. It authenticates no learned evidence, remote effect, application
33/// safety or command/byte custody and grants no terminal or reference release.
34#[derive(Clone, Debug, Eq, PartialEq)]
35pub struct ExecutionWorkflowSettlementView {
36    workflow: ArtifactChecksumRecord,
37    stages: Vec<ExecutionStageSettlementView>,
38}
39impl ExecutionWorkflowSettlementView {
40    /// Read the exact original workflow declaration identity.
41    #[must_use]
42    pub const fn workflow(&self) -> &ArtifactChecksumRecord {
43        &self.workflow
44    }
45    /// Read one row per original stage in canonical allocation order.
46    #[must_use]
47    pub fn stages(&self) -> &[ExecutionStageSettlementView] {
48        &self.stages
49    }
50}
51
52/// One original stage's bound plan and complete chronological local checkpoint.
53#[derive(Clone, Debug, Eq, PartialEq)]
54pub struct ExecutionStageSettlementView {
55    stage_sequence: u64,
56    binding: ArtifactChecksumRecord,
57    settlement: ArtifactChecksumRecord,
58}
59impl ExecutionStageSettlementView {
60    /// Read the original opaque stage identity.
61    #[must_use]
62    pub const fn stage_sequence(&self) -> u64 {
63        self.stage_sequence
64    }
65    /// Read the exact retained binding identity.
66    #[must_use]
67    pub const fn binding(&self) -> &ArtifactChecksumRecord {
68        &self.binding
69    }
70    /// Read the complete original journal-history checkpoint identity.
71    #[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
84/// Durably create the full original workflow allocation without replacement.
85/// # Errors
86/// Rejects occupied/unsafe paths, bounds, ownership and publication failures.
87pub 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}
98/// Read exact original workflow allocation; missing evidence never supplies a new budget.
99/// # Errors
100/// Rejects missing/changed/unsafe/oversized records and ownership failures.
101pub 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
114/// Read which allocated stage paths are already occupied under the held layout.
115///
116/// This is a namespace preflight for a retained run, not evidence that an absent
117/// stage has never dispatched. An occupied partial stage remains occupied and is
118/// never repaired. Derived symlinks and non-directories refuse admission.
119/// # Errors
120/// Rejects changed workflow/layout, unsafe stage entries and IO failures.
121pub(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
142/// Replay every allocated stage's exact retained binding and all-Applied checkpoint.
143///
144/// The original workflow determines the complete stage set. Every fixed stage must
145/// exist with a valid child plan, immutable binding, complete original journal set
146/// and exact published settlement. Every binding's predecessor identities must
147/// match this complete original set. Stage locks are held one at a time under
148/// the caller's workflow layout guard. No absent stage means zero spending, and no
149/// occupied checkpoint is repaired or republished. The resulting digest rows
150/// are local evidence only; integration-owned authenticity, complete transfer,
151/// application safety, command quiescence and terminal/release proof remain separate.
152/// # Errors
153/// Rejects missing, unsafe, changed, held or unsettled original evidence without effects.
154pub fn read_complete_execution_workflow(
155    layout: &BackupLayoutGuard,
156    expected: &ArtifactChecksumRecord,
157) -> Result<ExecutionWorkflowSettlementView, ExecutionStageCheckpointError> {
158    read_complete_with(layout, expected, || {
159        read_execution_workflow(layout, expected)
160    })
161}
162
163/// Replay complete original settlement while retaining the exact workflow journal lock.
164///
165/// Terminal-evidence providers may use this instead of reacquiring their held journal
166/// lock. The guard must protect this layout's original `execution-workflow.json`.
167/// Both opening and closing reads retain the ordinary bounds, strict decoding and
168/// expected original digest checks; child locks and checkpoint admission are unchanged.
169/// This supplies local settlement only, never application or reference-release authority.
170/// # Errors
171/// Rejects a guard for another journal, changed originals, or incomplete settlement.
172pub fn read_complete_execution_workflow_locked(
173    layout: &BackupLayoutGuard,
174    journal: &JournalLock,
175    expected: &ArtifactChecksumRecord,
176) -> Result<ExecutionWorkflowSettlementView, ExecutionStageCheckpointError> {
177    if journal.journal() != layout.root().join(WORKFLOW_FILE) {
178        return Err(ExecutionWorkflowPersistenceError::DigestMismatch.into());
179    }
180    read_complete_with(layout, expected, || {
181        layout.check_root()?;
182        let workflow: ExecutionWorkflowRecord =
183            read_json(journal.journal(), MAX_EXECUTION_WORKFLOW_BYTES)?;
184        super::json::check_json_size(&workflow, MAX_EXECUTION_WORKFLOW_BYTES)?;
185        if &workflow.digest() != expected {
186            return Err(ExecutionWorkflowPersistenceError::DigestMismatch);
187        }
188        Ok(workflow)
189    })
190}
191
192fn read_complete_with(
193    layout: &BackupLayoutGuard,
194    expected: &ArtifactChecksumRecord,
195    read_original: impl Fn() -> Result<ExecutionWorkflowRecord, ExecutionWorkflowPersistenceError>,
196) -> Result<ExecutionWorkflowSettlementView, ExecutionStageCheckpointError> {
197    let workflow = read_original()?;
198    let mut stages = Vec::with_capacity(workflow.stages().len());
199    let mut bindings = Vec::with_capacity(workflow.stages().len());
200    let mut by_sequence = BTreeMap::new();
201    for allocated in workflow.stages() {
202        let sequence = allocated.operation_sequence();
203        let stage = acquire_stage(layout, sequence)?;
204        let binding: ExecutionStageBindingRecord =
205            read_json(&stage.root().join(BINDING_FILE), MAX_EXECUTION_STAGE_BYTES)
206                .map_err(ExecutionWorkflowPersistenceError::from)?;
207        let (binding, plan) = read_binding(&stage, &workflow, sequence, &binding.digest())?;
208        let settlement = replay_execution_settlement(&stage, &plan.digest())?;
209        read_binding(&stage, &workflow, sequence, &binding.digest())?;
210        by_sequence.insert(sequence, stages.len());
211        stages.push(ExecutionStageSettlementView {
212            stage_sequence: sequence,
213            binding: binding.digest(),
214            settlement: settlement.digest(),
215        });
216        bindings.push(binding);
217    }
218    for binding in &bindings {
219        for predecessor in binding.predecessors() {
220            let Some(&index) = by_sequence.get(&predecessor.stage_sequence()) else {
221                return Err(ExecutionWorkflowPersistenceError::DigestMismatch.into());
222            };
223            let actual = &stages[index];
224            if predecessor.binding() != actual.binding()
225                || predecessor.settlement() != actual.settlement()
226            {
227                return Err(ExecutionWorkflowPersistenceError::DigestMismatch.into());
228            }
229        }
230    }
231    read_original()?;
232    Ok(ExecutionWorkflowSettlementView {
233        workflow: expected.clone(),
234        stages,
235    })
236}
237
238/// One original stage's fixed layout, retained under both workflow and child exclusion.
239///
240/// Existing attempt journals remain the only spending owner. `create` supplies no
241/// journals; `prepare` creates the complete original set. Reopening never creates,
242/// resets or reconstructs journals. Integrations
243/// retain exact learned inputs, authenticate their relation to settled originals,
244/// and qualify fresh authority, command custody and proof of no previous dispatch.
245#[derive(Debug)]
246pub struct ExecutionStageGuard<'a> {
247    workflow_layout: &'a BackupLayoutGuard,
248    stage_layout: BackupLayoutGuard,
249    binding: ExecutionStageBindingRecord,
250    plan: OperationPlanRecord,
251}
252impl<'a> ExecutionStageGuard<'a> {
253    /// Complete an absent local checkpoint or replay its exact occupied record.
254    ///
255    /// This is local recovery after uncertain checkpoint publication: all original
256    /// journals and ancestor histories must still admit, and an occupied changed
257    /// checkpoint is never replaced. The supplied learned evidence remains opaque
258    /// integration-owned input, not an authenticated response or effect receipt.
259    /// No provider, reservation, refund, terminal or reference release is involved.
260    /// # Errors
261    /// Rejects incomplete originals, unsafe or conflicting checkpoints and IO failures.
262    pub fn reconcile_checkpoint(
263        &self,
264        learned_evidence: ArtifactChecksumRecord,
265    ) -> Result<ExecutionStagePredecessorRecord, ExecutionStageCheckpointError> {
266        self.reconcile_checkpoint_with(learned_evidence, || {})
267    }
268    fn reconcile_checkpoint_with(
269        &self,
270        learned_evidence: ArtifactChecksumRecord,
271        after_reconciliation: impl FnOnce(),
272    ) -> Result<ExecutionStagePredecessorRecord, ExecutionStageCheckpointError> {
273        let settlement = reconcile_execution_settlement(self.layout()?, &self.plan.digest())?;
274        after_reconciliation();
275        self.layout()?;
276        Ok(ExecutionStagePredecessorRecord::new(
277            self.binding.stage_sequence(),
278            self.binding.digest(),
279            settlement.digest(),
280            learned_evidence,
281        ))
282    }
283    /// Reconstruct an exact predecessor after a lost checkpoint response.
284    ///
285    /// Derives the settlement identity from all retained original journals and
286    /// reads the already published checkpoint. It never creates a checkpoint,
287    /// journal or receipt. The supplied learned evidence remains integration-owned
288    /// and must be independently qualified before a successor is dispatched.
289    /// # Errors
290    /// Rejects missing, held, unsettled or changed originals and checkpoint drift.
291    pub fn replay_checkpoint(
292        &self,
293        learned_evidence: ArtifactChecksumRecord,
294    ) -> Result<ExecutionStagePredecessorRecord, ExecutionStageCheckpointError> {
295        let settlement = replay_execution_settlement(self.layout()?, &self.plan.digest())?;
296        self.layout()?;
297        Ok(ExecutionStagePredecessorRecord::new(
298            self.binding.stage_sequence(),
299            self.binding.digest(),
300            settlement.digest(),
301            learned_evidence,
302        ))
303    }
304    /// Resume an exact retained stage with its complete original journal progress.
305    ///
306    /// Reuses record-only `open` and canonical complete-journal admission, then
307    /// rechecks the original stage and ancestor histories before returning. Hold
308    /// no attempt guards. Reads only retained local records, without creating
309    /// journals, reading artifact trees or calling providers. Pending and exhausted
310    /// spending is retained.
311    /// The returned progress is a sequential local projection, not atomic custody,
312    /// fresh permission, dispatch, authenticated receipt or terminal/release proof.
313    /// # Errors
314    /// Rejects missing/changed/unsafe originals, journal contention and invalid
315    /// retained accounting/causality without repair or replenishing allowance.
316    pub fn resume(
317        workflow_layout: &'a BackupLayoutGuard,
318        expected_workflow: &ArtifactChecksumRecord,
319        sequence: u64,
320        expected_binding: &ArtifactChecksumRecord,
321    ) -> Result<(Self, ExecutionProgressView), ExecutionStageResumeError> {
322        let stage = Self::open(
323            workflow_layout,
324            expected_workflow,
325            sequence,
326            expected_binding,
327        )?;
328        let progress = read_execution_progress(&stage.stage_layout, &stage.plan.digest())?;
329        stage.layout()?;
330        Ok((stage, progress))
331    }
332    /// Durably prepare a new stage and its complete original attempt-journal set.
333    ///
334    /// Reuses create-only stage admission and the existing journal owner, taking
335    /// one journal lock at a time. Every original authority is derived before
336    /// stage allocation. No reservation, receipt, backend call or fresh dispatch
337    /// authority follows. Hold no other attempt guards during preparation.
338    ///
339    /// Partial preparation remains occupied and is never repaired by retrying.
340    /// Reopen only retained originals and require complete execution admission;
341    /// missing journals never mean unused allowance. A lost successful response
342    /// can reopen the complete exact stage without creating another journal.
343    /// # Errors
344    /// Rejects invalid originals, occupied stages, journal contention and failed
345    /// durable publication, preserving all partial records and original limits.
346    pub fn prepare(
347        workflow_layout: &'a BackupLayoutGuard,
348        binding: ExecutionStageBindingRecord,
349        plan: OperationPlanRecord,
350    ) -> Result<Self, ExecutionStagePreparationError> {
351        Self::prepare_with(workflow_layout, binding, plan, |_| {})
352    }
353    fn prepare_with(
354        workflow_layout: &'a BackupLayoutGuard,
355        binding: ExecutionStageBindingRecord,
356        plan: OperationPlanRecord,
357        mut barrier: impl FnMut(StagePreparationBarrier),
358    ) -> Result<Self, ExecutionStagePreparationError> {
359        let authorities = plan.attempt_authorities()?;
360        let stage = Self::create_with(workflow_layout, binding, plan, &mut barrier)?;
361        let layout = stage.layout()?;
362        for authority in authorities {
363            let sequence = authority.binding().operation_sequence();
364            drop(AttemptJournalGuard::create(layout, authority)?);
365            barrier(StagePreparationBarrier::Journal(sequence));
366        }
367        stage.layout()?;
368        Ok(stage)
369    }
370    // Allocate immutable stage records internally before preparing original journals.
371    fn create_with(
372        workflow_layout: &'a BackupLayoutGuard,
373        binding: ExecutionStageBindingRecord,
374        plan: OperationPlanRecord,
375        mut barrier: impl FnMut(StagePreparationBarrier),
376    ) -> Result<Self, ExecutionWorkflowPersistenceError> {
377        let workflow = read_execution_workflow(workflow_layout, binding.workflow())?;
378        binding.validate(&workflow, &plan)?;
379        let binding_bytes = super::json::encode_json(&binding, MAX_EXECUTION_STAGE_BYTES)?;
380        // Encode both documents before allocating any stage paths.
381        let plan_bytes = super::json::encode_json(
382            &plan,
383            crate::model::operation_plan::MAX_OPERATION_PLAN_BYTES,
384        )?;
385        validate_predecessors(workflow_layout, &workflow, &binding)?;
386        let path = stage_path(workflow_layout, binding.stage_sequence());
387        ic_host_fs::durable::create_private_directory(&path)
388            .map_err(|error| PersistenceError::Io(crate::ops::directory_creation_io(error)))?;
389        let stage_layout = acquire_stage(workflow_layout, binding.stage_sequence())?;
390        super::operation_plan::create_encoded(&stage_layout, &plan_bytes)?;
391        barrier(StagePreparationBarrier::Plan);
392        super::json::create_encoded(&stage_layout.root().join(BINDING_FILE), &binding_bytes)?;
393        barrier(StagePreparationBarrier::Binding);
394        workflow_layout.check_root()?;
395        Ok(Self {
396            workflow_layout,
397            stage_layout,
398            binding,
399            plan,
400        })
401    }
402    /// Reopen an exact fixed stage without allocation, repair or journal creation.
403    ///
404    /// Lost successful creation responses reconcile here. A partially prepared
405    /// directory without both immutable records rejects and remains retained.
406    /// Missing pending journals cannot be treated as unused allowance by the
407    /// existing execution-progress or settlement owners.
408    /// # Errors
409    /// Rejects missing/changed original records, prerequisite history or ownership failures.
410    pub fn open(
411        workflow_layout: &'a BackupLayoutGuard,
412        expected_workflow: &ArtifactChecksumRecord,
413        sequence: u64,
414        expected_binding: &ArtifactChecksumRecord,
415    ) -> Result<Self, ExecutionWorkflowPersistenceError> {
416        let workflow = read_execution_workflow(workflow_layout, expected_workflow)?;
417        workflow
418            .stage(sequence)
419            .map_err(|_| ExecutionWorkflowError::UnknownStage(sequence))?;
420        let stage_layout = acquire_stage(workflow_layout, sequence)?;
421        let (binding, plan) = read_binding(&stage_layout, &workflow, sequence, expected_binding)?;
422        validate_predecessors(workflow_layout, &workflow, &binding)?;
423        workflow_layout.check_root()?;
424        Ok(Self {
425            workflow_layout,
426            stage_layout,
427            binding,
428            plan,
429        })
430    }
431    /// Read the exact bound original child plan; no fresh permission is implied.
432    #[must_use]
433    pub const fn plan(&self) -> &OperationPlanRecord {
434        &self.plan
435    }
436    /// Read the immutable original stage and predecessor commitments.
437    #[must_use]
438    pub const fn binding(&self) -> &ExecutionStageBindingRecord {
439        &self.binding
440    }
441    /// Re-admit both original records and expose the existing journal/layout owner.
442    ///
443    /// Callers must use complete original journal admission for resume. Never
444    /// create a missing journal on resume or substitute a freshly derived plan.
445    /// # Errors
446    /// Rejects changed roots, retained records or any ancestor settlement history.
447    pub fn layout(&self) -> Result<&BackupLayoutGuard, ExecutionWorkflowPersistenceError> {
448        let workflow = read_execution_workflow(self.workflow_layout, self.binding.workflow())?;
449        read_binding(
450            &self.stage_layout,
451            &workflow,
452            self.binding.stage_sequence(),
453            &self.binding.digest(),
454        )?;
455        validate_predecessors(self.workflow_layout, &workflow, &self.binding)?;
456        Ok(&self.stage_layout)
457    }
458}
459
460/// Stage-bound checkpoint publication/replay refusal; originals remain retained.
461#[derive(Debug, Error)]
462pub enum ExecutionStageCheckpointError {
463    /// Original workflow/stage or complete ancestor history cannot be admitted.
464    #[error(transparent)]
465    Stage(#[from] ExecutionWorkflowPersistenceError),
466    /// Complete original journal checkpoint derivation/publication failed.
467    #[error(transparent)]
468    Settlement(#[from] ExecutionSettlementCheckpointError),
469}
470
471/// Failed create-only preparation; retained records never grant repair or extra spending.
472#[derive(Debug, Error)]
473pub enum ExecutionStagePreparationError {
474    /// Original stage records, predecessor settlement or layout were rejected.
475    #[error(transparent)]
476    Stage(#[from] ExecutionWorkflowPersistenceError),
477    /// Complete original child authority derivation failed before allocation.
478    #[error(transparent)]
479    Authority(#[from] OperationPlanError),
480    /// Existing attempt-journal locking or durable create-only publication failed.
481    #[error(transparent)]
482    Journal(#[from] AttemptJournalError),
483}
484
485/// Complete retained stage-resume refusal; original evidence and spending remain unchanged.
486#[derive(Debug, Error)]
487pub enum ExecutionStageResumeError {
488    /// Original stage, ancestor history or layout cannot be admitted.
489    #[error(transparent)]
490    Stage(#[from] ExecutionWorkflowPersistenceError),
491    /// Complete original child journals, accounting or causality cannot be admitted.
492    #[error(transparent)]
493    Progress(#[from] ExecutionProgressPersistenceError),
494}
495fn read_binding(
496    layout: &BackupLayoutGuard,
497    workflow: &ExecutionWorkflowRecord,
498    sequence: u64,
499    expected: &ArtifactChecksumRecord,
500) -> Result<(ExecutionStageBindingRecord, OperationPlanRecord), ExecutionWorkflowPersistenceError> {
501    layout.check_root()?;
502    let binding: ExecutionStageBindingRecord =
503        read_json(&layout.root().join(BINDING_FILE), MAX_EXECUTION_STAGE_BYTES)?;
504    super::json::check_json_size(&binding, MAX_EXECUTION_STAGE_BYTES)?;
505    if binding.stage_sequence() != sequence || &binding.digest() != expected {
506        return Err(ExecutionWorkflowPersistenceError::DigestMismatch);
507    }
508    let plan = read_operation_plan(layout, binding.plan())?;
509    binding.validate(workflow, &plan)?;
510    Ok((binding, plan))
511}
512fn validate_predecessors(
513    layout: &BackupLayoutGuard,
514    workflow: &ExecutionWorkflowRecord,
515    binding: &ExecutionStageBindingRecord,
516) -> Result<(), ExecutionWorkflowPersistenceError> {
517    // Traverse the original acyclic catalog iteratively, admitting each ancestor
518    // once. Keep at most one row per catalog node, not one per dependency path.
519    // Learned evidence may differ between edges; binding/settlement identity cannot.
520    let mut expected: BTreeMap<u64, ExecutionStagePredecessorRecord> = binding
521        .predecessors()
522        .iter()
523        .map(|row| (row.stage_sequence(), row.clone()))
524        .collect();
525    let mut pending: Vec<_> = expected.keys().copied().collect();
526    while let Some(sequence) = pending.pop() {
527        let row = &expected[&sequence];
528        let original = {
529            // Release this layout and every journal lock before admitting another
530            // ancestor, including an ancestor shared by multiple branches.
531            let predecessor = acquire_stage(layout, sequence)?;
532            let (original, _) = read_binding(&predecessor, workflow, sequence, row.binding())?;
533            read_execution_settlement(&predecessor, original.plan(), row.settlement())?;
534            original
535        };
536        for ancestor in original.predecessors() {
537            match expected.entry(ancestor.stage_sequence()) {
538                Entry::Vacant(entry) => {
539                    pending.push(ancestor.stage_sequence());
540                    entry.insert(ancestor.clone());
541                }
542                Entry::Occupied(entry) => {
543                    if entry.get().binding() != ancestor.binding()
544                        || entry.get().settlement() != ancestor.settlement()
545                    {
546                        return Err(ExecutionWorkflowPersistenceError::DigestMismatch);
547                    }
548                }
549            }
550        }
551    }
552    Ok(())
553}
554fn stage_path(layout: &BackupLayoutGuard, sequence: u64) -> PathBuf {
555    layout.root().join(format!("execution-stage-{sequence}"))
556}
557fn acquire_stage(
558    layout: &BackupLayoutGuard,
559    sequence: u64,
560) -> Result<BackupLayoutGuard, ExecutionWorkflowPersistenceError> {
561    layout.check_root()?;
562    let path = stage_path(layout, sequence);
563    let metadata = fs::symlink_metadata(&path).map_err(PersistenceError::from)?;
564    if !metadata.is_dir() {
565        return Err(ExecutionWorkflowPersistenceError::UnsafeStage);
566    }
567    let stage = BackupLayoutGuard::acquire(&path)?;
568    // Unlike an explicitly operator-selected root, a derived stage symlink is
569    // never adopted. Noncooperating namespace stability remains caller-owned.
570    if stage.root() != path {
571        return Err(ExecutionWorkflowPersistenceError::UnsafeStage);
572    }
573    layout.check_root()?;
574    Ok(stage)
575}
576/// Typed immutable workflow/stage admission failure; no recovery evidence is removed.
577#[derive(Debug, Error)]
578pub enum ExecutionWorkflowPersistenceError {
579    /// Workflow/stage identity differs from the retained expected digest.
580    #[error("execution workflow/stage digest mismatch")]
581    DigestMismatch,
582    /// A derived stage directory is a symlink or another unsafe entry.
583    #[error("unsafe execution stage directory")]
584    UnsafeStage,
585    /// Original model allocation or learned binding was rejected.
586    #[error(transparent)]
587    Model(#[from] ExecutionWorkflowError),
588    /// Existing immutable operation-plan admission failed.
589    #[error(transparent)]
590    Plan(#[from] OperationPlanPersistenceError),
591    /// Existing complete chronological Applied settlement admission failed.
592    #[error(transparent)]
593    Settlement(#[from] ExecutionSettlementPersistenceError),
594    /// Cooperating layout or journal exclusion failed.
595    #[error(transparent)]
596    Lock(#[from] JournalLockError),
597    /// Existing bounded JSON/durable filesystem owner failed.
598    #[error(transparent)]
599    Persistence(#[from] PersistenceError),
600}
601
602#[cfg(all(test, unix))]
603mod tests;
ic_backup/ops/persistence/execution_workflow/mod.rs

ic_backup/ops/persistence/execution_workflow/
mod.rs
