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

ic_backup/ops/persistence/execution_workflow/
mod.rs
