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,
7    checkpoint_execution_settlement, create_json_durable, create_operation_plan,
8    read_execution_progress, read_execution_settlement, read_json, read_operation_plan,
9    reconcile_execution_settlement, replay_execution_settlement,
10};
11use crate::model::{
12    artifacts::ArtifactChecksumRecord,
13    execution_workflow::{
14        ExecutionStageBindingRecord, ExecutionStagePredecessorRecord, ExecutionWorkflowError,
15        ExecutionWorkflowRecord, MAX_EXECUTION_STAGE_BYTES, MAX_EXECUTION_WORKFLOW_BYTES,
16    },
17    operation_plan::{OperationPlanError, OperationPlanRecord},
18};
19use crate::policy::execution_progress::ExecutionProgressView;
20use std::{
21    collections::{BTreeMap, btree_map::Entry},
22    fs,
23    path::PathBuf,
24};
25use thiserror::Error;
26
27const WORKFLOW_FILE: &str = "execution-workflow.json";
28const BINDING_FILE: &str = "stage-binding.json";
29
30/// Exact local identities of every original stage's retained checkpoint.
31///
32/// This is a sequential local projection, not an immutable product completion
33/// receipt. It authenticates no learned evidence, remote effect, application
34/// safety or command/byte custody and grants no terminal or reference release.
35#[derive(Clone, Debug, Eq, PartialEq)]
36pub struct ExecutionWorkflowSettlementView {
37    workflow: ArtifactChecksumRecord,
38    stages: Vec<ExecutionStageSettlementView>,
39}
40impl ExecutionWorkflowSettlementView {
41    /// Read the exact original workflow declaration identity.
42    #[must_use]
43    pub const fn workflow(&self) -> &ArtifactChecksumRecord {
44        &self.workflow
45    }
46    /// Read one row per original stage in canonical allocation order.
47    #[must_use]
48    pub fn stages(&self) -> &[ExecutionStageSettlementView] {
49        &self.stages
50    }
51}
52
53/// One original stage's bound plan and complete chronological local checkpoint.
54#[derive(Clone, Debug, Eq, PartialEq)]
55pub struct ExecutionStageSettlementView {
56    stage_sequence: u64,
57    binding: ArtifactChecksumRecord,
58    settlement: ArtifactChecksumRecord,
59}
60impl ExecutionStageSettlementView {
61    /// Read the original opaque stage identity.
62    #[must_use]
63    pub const fn stage_sequence(&self) -> u64 {
64        self.stage_sequence
65    }
66    /// Read the exact retained binding identity.
67    #[must_use]
68    pub const fn binding(&self) -> &ArtifactChecksumRecord {
69        &self.binding
70    }
71    /// Read the complete original journal-history checkpoint identity.
72    #[must_use]
73    pub const fn settlement(&self) -> &ArtifactChecksumRecord {
74        &self.settlement
75    }
76}
77
78#[derive(Clone, Copy, Debug, Eq, PartialEq)]
79enum StagePreparationBarrier {
80    Plan,
81    Binding,
82    Journal(u64),
83}
84
85/// Durably create the full original workflow allocation without replacement.
86/// # Errors
87/// Rejects occupied/unsafe paths, bounds, ownership and publication failures.
88pub fn create_execution_workflow(
89    layout: &BackupLayoutGuard,
90    record: &ExecutionWorkflowRecord,
91) -> Result<(), ExecutionWorkflowPersistenceError> {
92    layout.check_root()?;
93    super::json::check_json_size(record, MAX_EXECUTION_WORKFLOW_BYTES)?;
94    let path = layout.root().join(WORKFLOW_FILE);
95    let _lock = JournalLock::acquire(&path)?;
96    create_json_durable(&path, record)?;
97    Ok(())
98}
99/// Read exact original workflow allocation; missing evidence never supplies a new budget.
100/// # Errors
101/// Rejects missing/changed/unsafe/oversized records and ownership failures.
102pub fn read_execution_workflow(
103    layout: &BackupLayoutGuard,
104    expected: &ArtifactChecksumRecord,
105) -> Result<ExecutionWorkflowRecord, ExecutionWorkflowPersistenceError> {
106    layout.check_root()?;
107    let path = layout.root().join(WORKFLOW_FILE);
108    let _lock = JournalLock::acquire(&path)?;
109    let record: ExecutionWorkflowRecord = read_json(&path, MAX_EXECUTION_WORKFLOW_BYTES)?;
110    super::json::check_json_size(&record, MAX_EXECUTION_WORKFLOW_BYTES)?;
111    if &record.digest() != expected {
112        return Err(ExecutionWorkflowPersistenceError::DigestMismatch);
113    }
114    Ok(record)
115}
116
117/// Replay every allocated stage's exact retained binding and all-Applied checkpoint.
118///
119/// The original workflow determines the complete stage set. Every fixed stage must
120/// exist with a valid child plan, immutable binding, complete original journal set
121/// and exact published settlement. Every binding's predecessor identities must
122/// match this complete original set. Stage locks are held one at a time under
123/// the caller's workflow layout guard. No absent stage means zero spending, and no
124/// occupied checkpoint is repaired or republished. The resulting digest rows
125/// are local evidence only; integration-owned authenticity, complete transfer,
126/// application safety, command quiescence and terminal/release proof remain separate.
127/// # Errors
128/// Rejects missing, unsafe, changed, held or unsettled original evidence without effects.
129pub fn read_complete_execution_workflow(
130    layout: &BackupLayoutGuard,
131    expected: &ArtifactChecksumRecord,
132) -> Result<ExecutionWorkflowSettlementView, ExecutionStageCheckpointError> {
133    let workflow = read_execution_workflow(layout, expected)?;
134    let mut stages = Vec::with_capacity(workflow.stages().len());
135    let mut bindings = Vec::with_capacity(workflow.stages().len());
136    let mut by_sequence = BTreeMap::new();
137    for allocated in workflow.stages() {
138        let sequence = allocated.operation_sequence();
139        let stage = acquire_stage(layout, sequence)?;
140        let binding: ExecutionStageBindingRecord =
141            read_json(&stage.root().join(BINDING_FILE), MAX_EXECUTION_STAGE_BYTES)
142                .map_err(ExecutionWorkflowPersistenceError::from)?;
143        let (binding, plan) = read_binding(&stage, &workflow, sequence, &binding.digest())?;
144        let settlement = replay_execution_settlement(&stage, &plan.digest())?;
145        read_binding(&stage, &workflow, sequence, &binding.digest())?;
146        by_sequence.insert(sequence, stages.len());
147        stages.push(ExecutionStageSettlementView {
148            stage_sequence: sequence,
149            binding: binding.digest(),
150            settlement: settlement.digest(),
151        });
152        bindings.push(binding);
153    }
154    for binding in &bindings {
155        for predecessor in binding.predecessors() {
156            let Some(&index) = by_sequence.get(&predecessor.stage_sequence()) else {
157                return Err(ExecutionWorkflowPersistenceError::DigestMismatch.into());
158            };
159            let actual = &stages[index];
160            if predecessor.binding() != actual.binding()
161                || predecessor.settlement() != actual.settlement()
162            {
163                return Err(ExecutionWorkflowPersistenceError::DigestMismatch.into());
164            }
165        }
166    }
167    read_execution_workflow(layout, expected)?;
168    Ok(ExecutionWorkflowSettlementView {
169        workflow: expected.clone(),
170        stages,
171    })
172}
173
174/// One original stage's fixed layout, retained under both workflow and child exclusion.
175///
176/// Existing attempt journals remain the only spending owner. `create` supplies no
177/// journals; `prepare` creates the complete original set. Reopening never creates,
178/// resets or reconstructs journals. Integrations
179/// retain exact learned inputs, authenticate their relation to settled originals,
180/// and qualify fresh authority, command custody and proof of no previous dispatch.
181#[derive(Debug)]
182pub struct ExecutionStageGuard<'a> {
183    workflow_layout: &'a BackupLayoutGuard,
184    stage_layout: BackupLayoutGuard,
185    binding: ExecutionStageBindingRecord,
186    plan: OperationPlanRecord,
187}
188impl<'a> ExecutionStageGuard<'a> {
189    /// Complete an absent local checkpoint or replay its exact occupied record.
190    ///
191    /// This is local recovery after uncertain checkpoint publication: all original
192    /// journals and ancestor histories must still admit, and an occupied changed
193    /// checkpoint is never replaced. The supplied learned evidence remains opaque
194    /// integration-owned input, not an authenticated response or effect receipt.
195    /// No provider, reservation, refund, terminal or reference release is involved.
196    /// # Errors
197    /// Rejects incomplete originals, unsafe or conflicting checkpoints and IO failures.
198    pub fn reconcile_checkpoint(
199        &self,
200        learned_evidence: ArtifactChecksumRecord,
201    ) -> Result<ExecutionStagePredecessorRecord, ExecutionStageCheckpointError> {
202        let settlement = reconcile_execution_settlement(self.layout()?, &self.plan.digest())?;
203        self.layout()?;
204        Ok(ExecutionStagePredecessorRecord::new(
205            self.binding.stage_sequence(),
206            self.binding.digest(),
207            settlement.digest(),
208            learned_evidence,
209        ))
210    }
211    /// Reconstruct an exact predecessor after a lost checkpoint response.
212    ///
213    /// Derives the settlement identity from all retained original journals and
214    /// reads the already published checkpoint. It never creates a checkpoint,
215    /// journal or receipt. The supplied learned evidence remains integration-owned
216    /// and must be independently qualified before a successor is dispatched.
217    /// # Errors
218    /// Rejects missing, held, unsettled or changed originals and checkpoint drift.
219    pub fn replay_checkpoint(
220        &self,
221        learned_evidence: ArtifactChecksumRecord,
222    ) -> Result<ExecutionStagePredecessorRecord, ExecutionStageCheckpointError> {
223        let settlement = replay_execution_settlement(self.layout()?, &self.plan.digest())?;
224        self.layout()?;
225        Ok(ExecutionStagePredecessorRecord::new(
226            self.binding.stage_sequence(),
227            self.binding.digest(),
228            settlement.digest(),
229            learned_evidence,
230        ))
231    }
232    /// Publish the original all-Applied checkpoint and derive its exact predecessor commitment.
233    ///
234    /// Re-admits the workflow, stage and complete ancestor histories before and after
235    /// delegating to the canonical original-journal checkpoint owner. Hold no attempt
236    /// guards. Failed post-publication admission retains the checkpoint and all originals;
237    /// an occupied checkpoint is never replaced. Identity-bound local replay remains
238    /// owned by `read_execution_settlement`.
239    ///
240    /// The supplied learned evidence stays integration-owned: this authenticates no
241    /// learned input, receipt or effect, and grants no dispatch or terminal/release proof.
242    /// # Errors
243    /// Rejects changed/missing/held originals, unsettled journals and publication failures.
244    pub fn checkpoint(
245        &self,
246        learned_evidence: ArtifactChecksumRecord,
247    ) -> Result<ExecutionStagePredecessorRecord, ExecutionStageCheckpointError> {
248        self.checkpoint_with(learned_evidence, || {})
249    }
250    fn checkpoint_with(
251        &self,
252        learned_evidence: ArtifactChecksumRecord,
253        after_publication: impl FnOnce(),
254    ) -> Result<ExecutionStagePredecessorRecord, ExecutionStageCheckpointError> {
255        let settlement = checkpoint_execution_settlement(self.layout()?, &self.plan.digest())?;
256        after_publication();
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    /// Resume an exact retained stage with its complete original journal progress.
266    ///
267    /// Reuses record-only `open` and canonical complete-journal admission, then
268    /// rechecks the original stage and ancestor histories before returning. Hold
269    /// no attempt guards. Reads only retained local records, without creating
270    /// journals, reading artifact trees or calling providers. Pending and exhausted
271    /// spending is retained.
272    /// The returned progress is a sequential local projection, not atomic custody,
273    /// fresh permission, dispatch, authenticated receipt or terminal/release proof.
274    /// # Errors
275    /// Rejects missing/changed/unsafe originals, journal contention and invalid
276    /// retained accounting/causality without repair or replenishing allowance.
277    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    /// Durably prepare a new stage and its complete original attempt-journal set.
294    ///
295    /// Reuses create-only stage admission and the existing journal owner, taking
296    /// one journal lock at a time. Every original authority is derived before
297    /// stage allocation. No reservation, receipt, backend call or fresh dispatch
298    /// authority follows. Hold no other attempt guards during preparation.
299    ///
300    /// Partial preparation remains occupied and is never repaired by retrying.
301    /// Reopen only retained originals and require complete execution admission;
302    /// missing journals never mean unused allowance. A lost successful response
303    /// can reopen the complete exact stage without creating another journal.
304    /// # Errors
305    /// Rejects invalid originals, occupied stages, journal contention and failed
306    /// durable publication, preserving all partial records and original limits.
307    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    /// Create a fixed private stage directory, exact child plan and immutable binding.
332    ///
333    /// Every original predecessor and ancestor needs its exact retained binding, plan and
334    /// complete chronological Applied settlement. The directory is create-only:
335    /// occupied or interrupted preparation is retained and cannot be recreated
336    /// with another plan. No journal or backend effect occurs before return.
337    /// # Errors
338    /// Rejects changed allocation, unavailable prerequisites, occupied stages and IO failures.
339    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        super::json::check_json_size(&binding, MAX_EXECUTION_STAGE_BYTES)?;
355        // Validate both documents before allocating any stage paths.
356        super::json::check_json_size(
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        create_private_directory(&path)?;
363        fs::File::open(workflow_layout.root())
364            .and_then(|directory| directory.sync_all())
365            .map_err(PersistenceError::from)?;
366        let stage_layout = acquire_stage(workflow_layout, binding.stage_sequence())?;
367        create_operation_plan(&stage_layout, &plan)?;
368        barrier(StagePreparationBarrier::Plan);
369        create_json_durable(&stage_layout.root().join(BINDING_FILE), &binding)?;
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}
553fn create_private_directory(
554    path: &std::path::Path,
555) -> Result<(), ExecutionWorkflowPersistenceError> {
556    #[cfg(unix)]
557    {
558        use std::os::unix::fs::DirBuilderExt;
559        fs::DirBuilder::new()
560            .mode(0o700)
561            .create(path)
562            .map_err(PersistenceError::from)?;
563        Ok(())
564    }
565    #[cfg(not(unix))]
566    {
567        let _ = path;
568        Err(PersistenceError::Io(std::io::Error::from(std::io::ErrorKind::Unsupported)).into())
569    }
570}
571
572/// Typed immutable workflow/stage admission failure; no recovery evidence is removed.
573#[derive(Debug, Error)]
574pub enum ExecutionWorkflowPersistenceError {
575    /// Workflow/stage identity differs from the retained expected digest.
576    #[error("execution workflow/stage digest mismatch")]
577    DigestMismatch,
578    /// A derived stage directory is a symlink or another unsafe entry.
579    #[error("unsafe execution stage directory")]
580    UnsafeStage,
581    /// Original model allocation or learned binding was rejected.
582    #[error(transparent)]
583    Model(#[from] ExecutionWorkflowError),
584    /// Existing immutable operation-plan admission failed.
585    #[error(transparent)]
586    Plan(#[from] OperationPlanPersistenceError),
587    /// Existing complete chronological Applied settlement admission failed.
588    #[error(transparent)]
589    Settlement(#[from] ExecutionSettlementPersistenceError),
590    /// Cooperating layout or journal exclusion failed.
591    #[error(transparent)]
592    Lock(#[from] JournalLockError),
593    /// Existing bounded JSON/durable filesystem owner failed.
594    #[error(transparent)]
595    Persistence(#[from] PersistenceError),
596}
597
598#[cfg(all(test, unix))]
599mod tests;
ic_backup/ops/persistence/execution_workflow/mod.rs

ic_backup/ops/persistence/execution_workflow/
mod.rs
