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, read_execution_progress, read_execution_settlement, read_json,
8    read_operation_plan, reconcile_execution_settlement, 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        let settlement = reconcile_execution_settlement(self.layout()?, &self.plan.digest())?;
230        self.layout()?;
231        Ok(ExecutionStagePredecessorRecord::new(
232            self.binding.stage_sequence(),
233            self.binding.digest(),
234            settlement.digest(),
235            learned_evidence,
236        ))
237    }
238    /// Reconstruct an exact predecessor after a lost checkpoint response.
239    ///
240    /// Derives the settlement identity from all retained original journals and
241    /// reads the already published checkpoint. It never creates a checkpoint,
242    /// journal or receipt. The supplied learned evidence remains integration-owned
243    /// and must be independently qualified before a successor is dispatched.
244    /// # Errors
245    /// Rejects missing, held, unsettled or changed originals and checkpoint drift.
246    pub fn replay_checkpoint(
247        &self,
248        learned_evidence: ArtifactChecksumRecord,
249    ) -> Result<ExecutionStagePredecessorRecord, ExecutionStageCheckpointError> {
250        let settlement = replay_execution_settlement(self.layout()?, &self.plan.digest())?;
251        self.layout()?;
252        Ok(ExecutionStagePredecessorRecord::new(
253            self.binding.stage_sequence(),
254            self.binding.digest(),
255            settlement.digest(),
256            learned_evidence,
257        ))
258    }
259    /// Publish the original all-Applied checkpoint and derive its exact predecessor commitment.
260    ///
261    /// Re-admits the workflow, stage and complete ancestor histories before and after
262    /// delegating to the canonical original-journal checkpoint owner. Hold no attempt
263    /// guards. Failed post-publication admission retains the checkpoint and all originals;
264    /// an occupied checkpoint is never replaced. Identity-bound local replay remains
265    /// owned by `read_execution_settlement`.
266    ///
267    /// The supplied learned evidence stays integration-owned: this authenticates no
268    /// learned input, receipt or effect, and grants no dispatch or terminal/release proof.
269    /// # Errors
270    /// Rejects changed/missing/held originals, unsettled journals and publication failures.
271    pub fn checkpoint(
272        &self,
273        learned_evidence: ArtifactChecksumRecord,
274    ) -> Result<ExecutionStagePredecessorRecord, ExecutionStageCheckpointError> {
275        self.checkpoint_with(learned_evidence, || {})
276    }
277    fn checkpoint_with(
278        &self,
279        learned_evidence: ArtifactChecksumRecord,
280        after_publication: impl FnOnce(),
281    ) -> Result<ExecutionStagePredecessorRecord, ExecutionStageCheckpointError> {
282        let settlement = checkpoint_execution_settlement(self.layout()?, &self.plan.digest())?;
283        after_publication();
284        self.layout()?;
285        Ok(ExecutionStagePredecessorRecord::new(
286            self.binding.stage_sequence(),
287            self.binding.digest(),
288            settlement.digest(),
289            learned_evidence,
290        ))
291    }
292    /// Resume an exact retained stage with its complete original journal progress.
293    ///
294    /// Reuses record-only `open` and canonical complete-journal admission, then
295    /// rechecks the original stage and ancestor histories before returning. Hold
296    /// no attempt guards. Reads only retained local records, without creating
297    /// journals, reading artifact trees or calling providers. Pending and exhausted
298    /// spending is retained.
299    /// The returned progress is a sequential local projection, not atomic custody,
300    /// fresh permission, dispatch, authenticated receipt or terminal/release proof.
301    /// # Errors
302    /// Rejects missing/changed/unsafe originals, journal contention and invalid
303    /// retained accounting/causality without repair or replenishing allowance.
304    pub fn resume(
305        workflow_layout: &'a BackupLayoutGuard,
306        expected_workflow: &ArtifactChecksumRecord,
307        sequence: u64,
308        expected_binding: &ArtifactChecksumRecord,
309    ) -> Result<(Self, ExecutionProgressView), ExecutionStageResumeError> {
310        let stage = Self::open(
311            workflow_layout,
312            expected_workflow,
313            sequence,
314            expected_binding,
315        )?;
316        let progress = read_execution_progress(&stage.stage_layout, &stage.plan.digest())?;
317        stage.layout()?;
318        Ok((stage, progress))
319    }
320    /// Durably prepare a new stage and its complete original attempt-journal set.
321    ///
322    /// Reuses create-only stage admission and the existing journal owner, taking
323    /// one journal lock at a time. Every original authority is derived before
324    /// stage allocation. No reservation, receipt, backend call or fresh dispatch
325    /// authority follows. Hold no other attempt guards during preparation.
326    ///
327    /// Partial preparation remains occupied and is never repaired by retrying.
328    /// Reopen only retained originals and require complete execution admission;
329    /// missing journals never mean unused allowance. A lost successful response
330    /// can reopen the complete exact stage without creating another journal.
331    /// # Errors
332    /// Rejects invalid originals, occupied stages, journal contention and failed
333    /// durable publication, preserving all partial records and original limits.
334    pub fn prepare(
335        workflow_layout: &'a BackupLayoutGuard,
336        binding: ExecutionStageBindingRecord,
337        plan: OperationPlanRecord,
338    ) -> Result<Self, ExecutionStagePreparationError> {
339        Self::prepare_with(workflow_layout, binding, plan, |_| {})
340    }
341    fn prepare_with(
342        workflow_layout: &'a BackupLayoutGuard,
343        binding: ExecutionStageBindingRecord,
344        plan: OperationPlanRecord,
345        mut barrier: impl FnMut(StagePreparationBarrier),
346    ) -> Result<Self, ExecutionStagePreparationError> {
347        let authorities = plan.attempt_authorities()?;
348        let stage = Self::create_with(workflow_layout, binding, plan, &mut barrier)?;
349        let layout = stage.layout()?;
350        for authority in authorities {
351            let sequence = authority.binding().operation_sequence();
352            drop(AttemptJournalGuard::create(layout, authority)?);
353            barrier(StagePreparationBarrier::Journal(sequence));
354        }
355        stage.layout()?;
356        Ok(stage)
357    }
358    /// Create a fixed private stage directory, exact child plan and immutable binding.
359    ///
360    /// Every original predecessor and ancestor needs its exact retained binding, plan and
361    /// complete chronological Applied settlement. The directory is create-only:
362    /// occupied or interrupted preparation is retained and cannot be recreated
363    /// with another plan. No journal or backend effect occurs before return.
364    /// # Errors
365    /// Rejects changed allocation, unavailable prerequisites, occupied stages and IO failures.
366    pub fn create(
367        workflow_layout: &'a BackupLayoutGuard,
368        binding: ExecutionStageBindingRecord,
369        plan: OperationPlanRecord,
370    ) -> Result<Self, ExecutionWorkflowPersistenceError> {
371        Self::create_with(workflow_layout, binding, plan, |_| {})
372    }
373    fn create_with(
374        workflow_layout: &'a BackupLayoutGuard,
375        binding: ExecutionStageBindingRecord,
376        plan: OperationPlanRecord,
377        mut barrier: impl FnMut(StagePreparationBarrier),
378    ) -> Result<Self, ExecutionWorkflowPersistenceError> {
379        let workflow = read_execution_workflow(workflow_layout, binding.workflow())?;
380        binding.validate(&workflow, &plan)?;
381        let binding_bytes = super::json::encode_json(&binding, MAX_EXECUTION_STAGE_BYTES)?;
382        // Encode both documents before allocating any stage paths.
383        let plan_bytes = super::json::encode_json(
384            &plan,
385            crate::model::operation_plan::MAX_OPERATION_PLAN_BYTES,
386        )?;
387        validate_predecessors(workflow_layout, &workflow, &binding)?;
388        let path = stage_path(workflow_layout, binding.stage_sequence());
389        ic_host_fs::durable::create_private_directory(&path)
390            .map_err(|error| PersistenceError::Io(crate::ops::directory_creation_io(error)))?;
391        let stage_layout = acquire_stage(workflow_layout, binding.stage_sequence())?;
392        super::operation_plan::create_encoded(&stage_layout, &plan_bytes)?;
393        barrier(StagePreparationBarrier::Plan);
394        super::json::create_encoded(&stage_layout.root().join(BINDING_FILE), &binding_bytes)?;
395        barrier(StagePreparationBarrier::Binding);
396        workflow_layout.check_root()?;
397        Ok(Self {
398            workflow_layout,
399            stage_layout,
400            binding,
401            plan,
402        })
403    }
404    /// Reopen an exact fixed stage without allocation, repair or journal creation.
405    ///
406    /// Lost successful creation responses reconcile here. A partially prepared
407    /// directory without both immutable records rejects and remains retained.
408    /// Missing pending journals cannot be treated as unused allowance by the
409    /// existing execution-progress or settlement owners.
410    /// # Errors
411    /// Rejects missing/changed original records, prerequisite history or ownership failures.
412    pub fn open(
413        workflow_layout: &'a BackupLayoutGuard,
414        expected_workflow: &ArtifactChecksumRecord,
415        sequence: u64,
416        expected_binding: &ArtifactChecksumRecord,
417    ) -> Result<Self, ExecutionWorkflowPersistenceError> {
418        let workflow = read_execution_workflow(workflow_layout, expected_workflow)?;
419        workflow
420            .stage(sequence)
421            .map_err(|_| ExecutionWorkflowError::UnknownStage(sequence))?;
422        let stage_layout = acquire_stage(workflow_layout, sequence)?;
423        let (binding, plan) = read_binding(&stage_layout, &workflow, sequence, expected_binding)?;
424        validate_predecessors(workflow_layout, &workflow, &binding)?;
425        workflow_layout.check_root()?;
426        Ok(Self {
427            workflow_layout,
428            stage_layout,
429            binding,
430            plan,
431        })
432    }
433    /// Read the exact bound original child plan; no fresh permission is implied.
434    #[must_use]
435    pub const fn plan(&self) -> &OperationPlanRecord {
436        &self.plan
437    }
438    /// Read the immutable original stage and predecessor commitments.
439    #[must_use]
440    pub const fn binding(&self) -> &ExecutionStageBindingRecord {
441        &self.binding
442    }
443    /// Re-admit both original records and expose the existing journal/layout owner.
444    ///
445    /// Callers must use complete original journal admission for resume. Never
446    /// create a missing journal on resume or substitute a freshly derived plan.
447    /// # Errors
448    /// Rejects changed roots, retained records or any ancestor settlement history.
449    pub fn layout(&self) -> Result<&BackupLayoutGuard, ExecutionWorkflowPersistenceError> {
450        let workflow = read_execution_workflow(self.workflow_layout, self.binding.workflow())?;
451        read_binding(
452            &self.stage_layout,
453            &workflow,
454            self.binding.stage_sequence(),
455            &self.binding.digest(),
456        )?;
457        validate_predecessors(self.workflow_layout, &workflow, &self.binding)?;
458        Ok(&self.stage_layout)
459    }
460}
461
462/// Stage-bound checkpoint publication/replay refusal; originals remain retained.
463#[derive(Debug, Error)]
464pub enum ExecutionStageCheckpointError {
465    /// Original workflow/stage or complete ancestor history cannot be admitted.
466    #[error(transparent)]
467    Stage(#[from] ExecutionWorkflowPersistenceError),
468    /// Complete original journal checkpoint derivation/publication failed.
469    #[error(transparent)]
470    Settlement(#[from] ExecutionSettlementCheckpointError),
471}
472
473/// Failed create-only preparation; retained records never grant repair or extra spending.
474#[derive(Debug, Error)]
475pub enum ExecutionStagePreparationError {
476    /// Original stage records, predecessor settlement or layout were rejected.
477    #[error(transparent)]
478    Stage(#[from] ExecutionWorkflowPersistenceError),
479    /// Complete original child authority derivation failed before allocation.
480    #[error(transparent)]
481    Authority(#[from] OperationPlanError),
482    /// Existing attempt-journal locking or durable create-only publication failed.
483    #[error(transparent)]
484    Journal(#[from] AttemptJournalError),
485}
486
487/// Complete retained stage-resume refusal; original evidence and spending remain unchanged.
488#[derive(Debug, Error)]
489pub enum ExecutionStageResumeError {
490    /// Original stage, ancestor history or layout cannot be admitted.
491    #[error(transparent)]
492    Stage(#[from] ExecutionWorkflowPersistenceError),
493    /// Complete original child journals, accounting or causality cannot be admitted.
494    #[error(transparent)]
495    Progress(#[from] ExecutionProgressPersistenceError),
496}
497fn read_binding(
498    layout: &BackupLayoutGuard,
499    workflow: &ExecutionWorkflowRecord,
500    sequence: u64,
501    expected: &ArtifactChecksumRecord,
502) -> Result<(ExecutionStageBindingRecord, OperationPlanRecord), ExecutionWorkflowPersistenceError> {
503    layout.check_root()?;
504    let binding: ExecutionStageBindingRecord =
505        read_json(&layout.root().join(BINDING_FILE), MAX_EXECUTION_STAGE_BYTES)?;
506    super::json::check_json_size(&binding, MAX_EXECUTION_STAGE_BYTES)?;
507    if binding.stage_sequence() != sequence || &binding.digest() != expected {
508        return Err(ExecutionWorkflowPersistenceError::DigestMismatch);
509    }
510    let plan = read_operation_plan(layout, binding.plan())?;
511    binding.validate(workflow, &plan)?;
512    Ok((binding, plan))
513}
514fn validate_predecessors(
515    layout: &BackupLayoutGuard,
516    workflow: &ExecutionWorkflowRecord,
517    binding: &ExecutionStageBindingRecord,
518) -> Result<(), ExecutionWorkflowPersistenceError> {
519    // Traverse the original acyclic catalog iteratively, admitting each ancestor
520    // once. Keep at most one row per catalog node, not one per dependency path.
521    // Learned evidence may differ between edges; binding/settlement identity cannot.
522    let mut expected: BTreeMap<u64, ExecutionStagePredecessorRecord> = binding
523        .predecessors()
524        .iter()
525        .map(|row| (row.stage_sequence(), row.clone()))
526        .collect();
527    let mut pending: Vec<_> = expected.keys().copied().collect();
528    while let Some(sequence) = pending.pop() {
529        let row = &expected[&sequence];
530        let original = {
531            // Release this layout and every journal lock before admitting another
532            // ancestor, including an ancestor shared by multiple branches.
533            let predecessor = acquire_stage(layout, sequence)?;
534            let (original, _) = read_binding(&predecessor, workflow, sequence, row.binding())?;
535            read_execution_settlement(&predecessor, original.plan(), row.settlement())?;
536            original
537        };
538        for ancestor in original.predecessors() {
539            match expected.entry(ancestor.stage_sequence()) {
540                Entry::Vacant(entry) => {
541                    pending.push(ancestor.stage_sequence());
542                    entry.insert(ancestor.clone());
543                }
544                Entry::Occupied(entry) => {
545                    if entry.get().binding() != ancestor.binding()
546                        || entry.get().settlement() != ancestor.settlement()
547                    {
548                        return Err(ExecutionWorkflowPersistenceError::DigestMismatch);
549                    }
550                }
551            }
552        }
553    }
554    Ok(())
555}
556fn stage_path(layout: &BackupLayoutGuard, sequence: u64) -> PathBuf {
557    layout.root().join(format!("execution-stage-{sequence}"))
558}
559fn acquire_stage(
560    layout: &BackupLayoutGuard,
561    sequence: u64,
562) -> Result<BackupLayoutGuard, ExecutionWorkflowPersistenceError> {
563    layout.check_root()?;
564    let path = stage_path(layout, sequence);
565    let metadata = fs::symlink_metadata(&path).map_err(PersistenceError::from)?;
566    if !metadata.is_dir() {
567        return Err(ExecutionWorkflowPersistenceError::UnsafeStage);
568    }
569    let stage = BackupLayoutGuard::acquire(&path)?;
570    // Unlike an explicitly operator-selected root, a derived stage symlink is
571    // never adopted. Noncooperating namespace stability remains caller-owned.
572    if stage.root() != path {
573        return Err(ExecutionWorkflowPersistenceError::UnsafeStage);
574    }
575    layout.check_root()?;
576    Ok(stage)
577}
578/// Typed immutable workflow/stage admission failure; no recovery evidence is removed.
579#[derive(Debug, Error)]
580pub enum ExecutionWorkflowPersistenceError {
581    /// Workflow/stage identity differs from the retained expected digest.
582    #[error("execution workflow/stage digest mismatch")]
583    DigestMismatch,
584    /// A derived stage directory is a symlink or another unsafe entry.
585    #[error("unsafe execution stage directory")]
586    UnsafeStage,
587    /// Original model allocation or learned binding was rejected.
588    #[error(transparent)]
589    Model(#[from] ExecutionWorkflowError),
590    /// Existing immutable operation-plan admission failed.
591    #[error(transparent)]
592    Plan(#[from] OperationPlanPersistenceError),
593    /// Existing complete chronological Applied settlement admission failed.
594    #[error(transparent)]
595    Settlement(#[from] ExecutionSettlementPersistenceError),
596    /// Cooperating layout or journal exclusion failed.
597    #[error(transparent)]
598    Lock(#[from] JournalLockError),
599    /// Existing bounded JSON/durable filesystem owner failed.
600    #[error(transparent)]
601    Persistence(#[from] PersistenceError),
602}
603
604#[cfg(all(test, unix))]
605mod tests;
ic_backup/ops/persistence/execution_workflow/mod.rs

ic_backup/ops/persistence/execution_workflow/
mod.rs
