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

ic_backup/ops/persistence/execution_workflow/
mod.rs
