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    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
116/// Read which allocated stage paths are already occupied under the held layout.
117///
118/// This is a namespace preflight for a retained run, not evidence that an absent
119/// stage has never dispatched. An occupied partial stage remains occupied and is
120/// never repaired. Derived symlinks and non-directories refuse admission.
121/// # Errors
122/// Rejects changed workflow/layout, unsafe stage entries and IO failures.
123pub(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
144/// Replay every allocated stage's exact retained binding and all-Applied checkpoint.
145///
146/// The original workflow determines the complete stage set. Every fixed stage must
147/// exist with a valid child plan, immutable binding, complete original journal set
148/// and exact published settlement. Every binding's predecessor identities must
149/// match this complete original set. Stage locks are held one at a time under
150/// the caller's workflow layout guard. No absent stage means zero spending, and no
151/// occupied checkpoint is repaired or republished. The resulting digest rows
152/// are local evidence only; integration-owned authenticity, complete transfer,
153/// application safety, command quiescence and terminal/release proof remain separate.
154/// # Errors
155/// Rejects missing, unsafe, changed, held or unsettled original evidence without effects.
156pub 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/// One original stage's fixed layout, retained under both workflow and child exclusion.
202///
203/// Existing attempt journals remain the only spending owner. `create` supplies no
204/// journals; `prepare` creates the complete original set. Reopening never creates,
205/// resets or reconstructs journals. Integrations
206/// retain exact learned inputs, authenticate their relation to settled originals,
207/// and qualify fresh authority, command custody and proof of no previous dispatch.
208#[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    /// Complete an absent local checkpoint or replay its exact occupied record.
217    ///
218    /// This is local recovery after uncertain checkpoint publication: all original
219    /// journals and ancestor histories must still admit, and an occupied changed
220    /// checkpoint is never replaced. The supplied learned evidence remains opaque
221    /// integration-owned input, not an authenticated response or effect receipt.
222    /// No provider, reservation, refund, terminal or reference release is involved.
223    /// # Errors
224    /// Rejects incomplete originals, unsafe or conflicting checkpoints and IO failures.
225    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    /// Reconstruct an exact predecessor after a lost checkpoint response.
247    ///
248    /// Derives the settlement identity from all retained original journals and
249    /// reads the already published checkpoint. It never creates a checkpoint,
250    /// journal or receipt. The supplied learned evidence remains integration-owned
251    /// and must be independently qualified before a successor is dispatched.
252    /// # Errors
253    /// Rejects missing, held, unsettled or changed originals and checkpoint drift.
254    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    /// Resume an exact retained stage with its complete original journal progress.
268    ///
269    /// Reuses record-only `open` and canonical complete-journal admission, then
270    /// rechecks the original stage and ancestor histories before returning. Hold
271    /// no attempt guards. Reads only retained local records, without creating
272    /// journals, reading artifact trees or calling providers. Pending and exhausted
273    /// spending is retained.
274    /// The returned progress is a sequential local projection, not atomic custody,
275    /// fresh permission, dispatch, authenticated receipt or terminal/release proof.
276    /// # Errors
277    /// Rejects missing/changed/unsafe originals, journal contention and invalid
278    /// retained accounting/causality without repair or replenishing allowance.
279    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    /// Durably prepare a new stage and its complete original attempt-journal set.
296    ///
297    /// Reuses create-only stage admission and the existing journal owner, taking
298    /// one journal lock at a time. Every original authority is derived before
299    /// stage allocation. No reservation, receipt, backend call or fresh dispatch
300    /// authority follows. Hold no other attempt guards during preparation.
301    ///
302    /// Partial preparation remains occupied and is never repaired by retrying.
303    /// Reopen only retained originals and require complete execution admission;
304    /// missing journals never mean unused allowance. A lost successful response
305    /// can reopen the complete exact stage without creating another journal.
306    /// # Errors
307    /// Rejects invalid originals, occupied stages, journal contention and failed
308    /// durable publication, preserving all partial records and original limits.
309    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    /// Create a fixed private stage directory, exact child plan and immutable binding.
334    ///
335    /// Every original predecessor and ancestor needs its exact retained binding, plan and
336    /// complete chronological Applied settlement. The directory is create-only:
337    /// occupied or interrupted preparation is retained and cannot be recreated
338    /// with another plan. No journal or backend effect occurs before return.
339    /// # Errors
340    /// Rejects changed allocation, unavailable prerequisites, occupied stages and IO failures.
341    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        // Encode both documents before allocating any stage paths.
358        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    /// Reopen an exact fixed stage without allocation, repair or journal creation.
380    ///
381    /// Lost successful creation responses reconcile here. A partially prepared
382    /// directory without both immutable records rejects and remains retained.
383    /// Missing pending journals cannot be treated as unused allowance by the
384    /// existing execution-progress or settlement owners.
385    /// # Errors
386    /// Rejects missing/changed original records, prerequisite history or ownership failures.
387    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    /// Read the exact bound original child plan; no fresh permission is implied.
409    #[must_use]
410    pub const fn plan(&self) -> &OperationPlanRecord {
411        &self.plan
412    }
413    /// Read the immutable original stage and predecessor commitments.
414    #[must_use]
415    pub const fn binding(&self) -> &ExecutionStageBindingRecord {
416        &self.binding
417    }
418    /// Re-admit both original records and expose the existing journal/layout owner.
419    ///
420    /// Callers must use complete original journal admission for resume. Never
421    /// create a missing journal on resume or substitute a freshly derived plan.
422    /// # Errors
423    /// Rejects changed roots, retained records or any ancestor settlement history.
424    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/// Stage-bound checkpoint publication/replay refusal; originals remain retained.
438#[derive(Debug, Error)]
439pub enum ExecutionStageCheckpointError {
440    /// Original workflow/stage or complete ancestor history cannot be admitted.
441    #[error(transparent)]
442    Stage(#[from] ExecutionWorkflowPersistenceError),
443    /// Complete original journal checkpoint derivation/publication failed.
444    #[error(transparent)]
445    Settlement(#[from] ExecutionSettlementCheckpointError),
446}
447
448/// Failed create-only preparation; retained records never grant repair or extra spending.
449#[derive(Debug, Error)]
450pub enum ExecutionStagePreparationError {
451    /// Original stage records, predecessor settlement or layout were rejected.
452    #[error(transparent)]
453    Stage(#[from] ExecutionWorkflowPersistenceError),
454    /// Complete original child authority derivation failed before allocation.
455    #[error(transparent)]
456    Authority(#[from] OperationPlanError),
457    /// Existing attempt-journal locking or durable create-only publication failed.
458    #[error(transparent)]
459    Journal(#[from] AttemptJournalError),
460}
461
462/// Complete retained stage-resume refusal; original evidence and spending remain unchanged.
463#[derive(Debug, Error)]
464pub enum ExecutionStageResumeError {
465    /// Original stage, ancestor history or layout cannot be admitted.
466    #[error(transparent)]
467    Stage(#[from] ExecutionWorkflowPersistenceError),
468    /// Complete original child journals, accounting or causality cannot be admitted.
469    #[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    // Traverse the original acyclic catalog iteratively, admitting each ancestor
495    // once. Keep at most one row per catalog node, not one per dependency path.
496    // Learned evidence may differ between edges; binding/settlement identity cannot.
497    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            // Release this layout and every journal lock before admitting another
507            // ancestor, including an ancestor shared by multiple branches.
508            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    // Unlike an explicitly operator-selected root, a derived stage symlink is
546    // never adopted. Noncooperating namespace stability remains caller-owned.
547    if stage.root() != path {
548        return Err(ExecutionWorkflowPersistenceError::UnsafeStage);
549    }
550    layout.check_root()?;
551    Ok(stage)
552}
553/// Typed immutable workflow/stage admission failure; no recovery evidence is removed.
554#[derive(Debug, Error)]
555pub enum ExecutionWorkflowPersistenceError {
556    /// Workflow/stage identity differs from the retained expected digest.
557    #[error("execution workflow/stage digest mismatch")]
558    DigestMismatch,
559    /// A derived stage directory is a symlink or another unsafe entry.
560    #[error("unsafe execution stage directory")]
561    UnsafeStage,
562    /// Original model allocation or learned binding was rejected.
563    #[error(transparent)]
564    Model(#[from] ExecutionWorkflowError),
565    /// Existing immutable operation-plan admission failed.
566    #[error(transparent)]
567    Plan(#[from] OperationPlanPersistenceError),
568    /// Existing complete chronological Applied settlement admission failed.
569    #[error(transparent)]
570    Settlement(#[from] ExecutionSettlementPersistenceError),
571    /// Cooperating layout or journal exclusion failed.
572    #[error(transparent)]
573    Lock(#[from] JournalLockError),
574    /// Existing bounded JSON/durable filesystem owner failed.
575    #[error(transparent)]
576    Persistence(#[from] PersistenceError),
577}
578
579#[cfg(all(test, unix))]
580mod tests;
ic_backup/ops/persistence/execution_workflow/mod.rs

ic_backup/ops/persistence/execution_workflow/
mod.rs
