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    for allocated in workflow.stages() {
197        let sequence = allocated.operation_sequence();
198        let stage = acquire_stage(layout, sequence)?;
199        let binding: ExecutionStageBindingRecord =
200            read_json(&stage.root().join(BINDING_FILE), MAX_EXECUTION_STAGE_BYTES)
201                .map_err(ExecutionWorkflowPersistenceError::from)?;
202        let (binding, plan) = read_binding(&stage, &workflow, sequence, &binding.digest())?;
203        let settlement = replay_execution_settlement(&stage, &plan.digest())?;
204        read_binding(&stage, &workflow, sequence, &binding.digest())?;
205        let view = ExecutionStageSettlementView {
206            stage_sequence: sequence,
207            binding: binding.digest(),
208            settlement: settlement.digest(),
209        };
210        stages.push((binding, view));
211    }
212    for (binding, _) in &stages {
213        for predecessor in binding.predecessors() {
214            // The original catalog supplies canonical sequence order.
215            let index = stages
216                .binary_search_by_key(&predecessor.stage_sequence(), |(_, view)| {
217                    view.stage_sequence()
218                })
219                .map_err(|_| ExecutionWorkflowPersistenceError::DigestMismatch)?;
220            let actual = &stages[index].1;
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: stages.into_iter().map(|(_, view)| view).collect(),
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
439    /// Borrow the held original workflow layout for read-only predecessor admission.
440    #[must_use]
441    pub const fn workflow_layout(&self) -> &BackupLayoutGuard {
442        self.workflow_layout
443    }
444    /// Re-admit both original records and expose the existing journal/layout owner.
445    ///
446    /// Callers must use complete original journal admission for resume. Never
447    /// create a missing journal on resume or substitute a freshly derived plan.
448    /// # Errors
449    /// Rejects changed roots, retained records or any ancestor settlement history.
450    pub fn layout(&self) -> Result<&BackupLayoutGuard, ExecutionWorkflowPersistenceError> {
451        let workflow = read_execution_workflow(self.workflow_layout, self.binding.workflow())?;
452        read_binding(
453            &self.stage_layout,
454            &workflow,
455            self.binding.stage_sequence(),
456            &self.binding.digest(),
457        )?;
458        validate_predecessors(self.workflow_layout, &workflow, &self.binding)?;
459        Ok(&self.stage_layout)
460    }
461}
462
463/// Stage-bound checkpoint publication/replay refusal; originals remain retained.
464#[derive(Debug, Error)]
465pub enum ExecutionStageCheckpointError {
466    /// Original workflow/stage or complete ancestor history cannot be admitted.
467    #[error(transparent)]
468    Stage(#[from] ExecutionWorkflowPersistenceError),
469    /// Complete original journal checkpoint derivation/publication failed.
470    #[error(transparent)]
471    Settlement(#[from] ExecutionSettlementCheckpointError),
472}
473
474/// Failed create-only preparation; retained records never grant repair or extra spending.
475#[derive(Debug, Error)]
476pub enum ExecutionStagePreparationError {
477    /// Original stage records, predecessor settlement or layout were rejected.
478    #[error(transparent)]
479    Stage(#[from] ExecutionWorkflowPersistenceError),
480    /// Complete original child authority derivation failed before allocation.
481    #[error(transparent)]
482    Authority(#[from] OperationPlanError),
483    /// Existing attempt-journal locking or durable create-only publication failed.
484    #[error(transparent)]
485    Journal(#[from] AttemptJournalError),
486}
487
488/// Complete retained stage-resume refusal; original evidence and spending remain unchanged.
489#[derive(Debug, Error)]
490pub enum ExecutionStageResumeError {
491    /// Original stage, ancestor history or layout cannot be admitted.
492    #[error(transparent)]
493    Stage(#[from] ExecutionWorkflowPersistenceError),
494    /// Complete original child journals, accounting or causality cannot be admitted.
495    #[error(transparent)]
496    Progress(#[from] ExecutionProgressPersistenceError),
497}
498fn read_binding(
499    layout: &BackupLayoutGuard,
500    workflow: &ExecutionWorkflowRecord,
501    sequence: u64,
502    expected: &ArtifactChecksumRecord,
503) -> Result<(ExecutionStageBindingRecord, OperationPlanRecord), ExecutionWorkflowPersistenceError> {
504    layout.check_root()?;
505    let binding: ExecutionStageBindingRecord =
506        read_json(&layout.root().join(BINDING_FILE), MAX_EXECUTION_STAGE_BYTES)?;
507    super::json::check_json_size(&binding, MAX_EXECUTION_STAGE_BYTES)?;
508    if binding.stage_sequence() != sequence || &binding.digest() != expected {
509        return Err(ExecutionWorkflowPersistenceError::DigestMismatch);
510    }
511    let plan = read_operation_plan(layout, binding.plan())?;
512    binding.validate(workflow, &plan)?;
513    Ok((binding, plan))
514}
515fn validate_predecessors(
516    layout: &BackupLayoutGuard,
517    workflow: &ExecutionWorkflowRecord,
518    binding: &ExecutionStageBindingRecord,
519) -> Result<(), ExecutionWorkflowPersistenceError> {
520    // Traverse the original acyclic catalog iteratively, admitting each ancestor
521    // once. Keep at most one row per catalog node, not one per dependency path.
522    // Learned evidence may differ between edges; binding/settlement identity cannot.
523    let mut expected: BTreeMap<_, _> = binding
524        .predecessors()
525        .iter()
526        .map(|row| {
527            (
528                row.stage_sequence(),
529                (row.binding().clone(), row.settlement().clone()),
530            )
531        })
532        .collect();
533    let mut pending: Vec<_> = expected.keys().copied().collect();
534    while let Some(sequence) = pending.pop() {
535        let (expected_binding, expected_settlement) = &expected[&sequence];
536        let original = {
537            // Release this layout and every journal lock before admitting another
538            // ancestor, including an ancestor shared by multiple branches.
539            let predecessor = acquire_stage(layout, sequence)?;
540            let (original, _) = read_binding(&predecessor, workflow, sequence, expected_binding)?;
541            read_execution_settlement(&predecessor, original.plan(), expected_settlement)?;
542            original
543        };
544        for ancestor in original.predecessors() {
545            match expected.entry(ancestor.stage_sequence()) {
546                Entry::Vacant(entry) => {
547                    pending.push(ancestor.stage_sequence());
548                    entry.insert((ancestor.binding().clone(), ancestor.settlement().clone()));
549                }
550                Entry::Occupied(entry) => {
551                    let (expected_binding, expected_settlement) = entry.get();
552                    if expected_binding != ancestor.binding()
553                        || expected_settlement != ancestor.settlement()
554                    {
555                        return Err(ExecutionWorkflowPersistenceError::DigestMismatch);
556                    }
557                }
558            }
559        }
560    }
561    Ok(())
562}
563fn stage_path(layout: &BackupLayoutGuard, sequence: u64) -> PathBuf {
564    layout.root().join(format!("execution-stage-{sequence}"))
565}
566fn acquire_stage(
567    layout: &BackupLayoutGuard,
568    sequence: u64,
569) -> Result<BackupLayoutGuard, ExecutionWorkflowPersistenceError> {
570    layout.check_root()?;
571    let path = stage_path(layout, sequence);
572    let metadata = fs::symlink_metadata(&path).map_err(PersistenceError::from)?;
573    if !metadata.is_dir() {
574        return Err(ExecutionWorkflowPersistenceError::UnsafeStage);
575    }
576    let stage = BackupLayoutGuard::acquire(&path)?;
577    // Unlike an explicitly operator-selected root, a derived stage symlink is
578    // never adopted. Noncooperating namespace stability remains caller-owned.
579    if stage.root() != path {
580        return Err(ExecutionWorkflowPersistenceError::UnsafeStage);
581    }
582    layout.check_root()?;
583    Ok(stage)
584}
585/// Typed immutable workflow/stage admission failure; no recovery evidence is removed.
586#[derive(Debug, Error)]
587pub enum ExecutionWorkflowPersistenceError {
588    /// Workflow/stage identity differs from the retained expected digest.
589    #[error("execution workflow/stage digest mismatch")]
590    DigestMismatch,
591    /// A derived stage directory is a symlink or another unsafe entry.
592    #[error("unsafe execution stage directory")]
593    UnsafeStage,
594    /// Original model allocation or learned binding was rejected.
595    #[error(transparent)]
596    Model(#[from] ExecutionWorkflowError),
597    /// Existing immutable operation-plan admission failed.
598    #[error(transparent)]
599    Plan(#[from] OperationPlanPersistenceError),
600    /// Existing complete chronological Applied settlement admission failed.
601    #[error(transparent)]
602    Settlement(#[from] ExecutionSettlementPersistenceError),
603    /// Cooperating layout or journal exclusion failed.
604    #[error(transparent)]
605    Lock(#[from] JournalLockError),
606    /// Existing bounded JSON/durable filesystem owner failed.
607    #[error(transparent)]
608    Persistence(#[from] PersistenceError),
609}
610
611#[cfg(all(test, unix))]
612mod tests;
ic_backup/ops/persistence/execution_workflow/mod.rs

ic_backup/ops/persistence/execution_workflow/
mod.rs
