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

ic_backup/ops/persistence/execution_workflow/
mod.rs
