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        create_private_directory(&path)?;
391        fs::File::open(workflow_layout.root())
392            .and_then(|directory| directory.sync_all())
393            .map_err(PersistenceError::from)?;
394        let stage_layout = acquire_stage(workflow_layout, binding.stage_sequence())?;
395        create_operation_plan(&stage_layout, &plan)?;
396        barrier(StagePreparationBarrier::Plan);
397        create_json_durable(&stage_layout.root().join(BINDING_FILE), &binding)?;
398        barrier(StagePreparationBarrier::Binding);
399        workflow_layout.check_root()?;
400        Ok(Self {
401            workflow_layout,
402            stage_layout,
403            binding,
404            plan,
405        })
406    }
407    /// Reopen an exact fixed stage without allocation, repair or journal creation.
408    ///
409    /// Lost successful creation responses reconcile here. A partially prepared
410    /// directory without both immutable records rejects and remains retained.
411    /// Missing pending journals cannot be treated as unused allowance by the
412    /// existing execution-progress or settlement owners.
413    /// # Errors
414    /// Rejects missing/changed original records, prerequisite history or ownership failures.
415    pub fn open(
416        workflow_layout: &'a BackupLayoutGuard,
417        expected_workflow: &ArtifactChecksumRecord,
418        sequence: u64,
419        expected_binding: &ArtifactChecksumRecord,
420    ) -> Result<Self, ExecutionWorkflowPersistenceError> {
421        let workflow = read_execution_workflow(workflow_layout, expected_workflow)?;
422        workflow
423            .stage(sequence)
424            .map_err(|_| ExecutionWorkflowError::UnknownStage(sequence))?;
425        let stage_layout = acquire_stage(workflow_layout, sequence)?;
426        let (binding, plan) = read_binding(&stage_layout, &workflow, sequence, expected_binding)?;
427        validate_predecessors(workflow_layout, &workflow, &binding)?;
428        workflow_layout.check_root()?;
429        Ok(Self {
430            workflow_layout,
431            stage_layout,
432            binding,
433            plan,
434        })
435    }
436    /// Read the exact bound original child plan; no fresh permission is implied.
437    #[must_use]
438    pub const fn plan(&self) -> &OperationPlanRecord {
439        &self.plan
440    }
441    /// Read the immutable original stage and predecessor commitments.
442    #[must_use]
443    pub const fn binding(&self) -> &ExecutionStageBindingRecord {
444        &self.binding
445    }
446    /// Re-admit both original records and expose the existing journal/layout owner.
447    ///
448    /// Callers must use complete original journal admission for resume. Never
449    /// create a missing journal on resume or substitute a freshly derived plan.
450    /// # Errors
451    /// Rejects changed roots, retained records or any ancestor settlement history.
452    pub fn layout(&self) -> Result<&BackupLayoutGuard, ExecutionWorkflowPersistenceError> {
453        let workflow = read_execution_workflow(self.workflow_layout, self.binding.workflow())?;
454        read_binding(
455            &self.stage_layout,
456            &workflow,
457            self.binding.stage_sequence(),
458            &self.binding.digest(),
459        )?;
460        validate_predecessors(self.workflow_layout, &workflow, &self.binding)?;
461        Ok(&self.stage_layout)
462    }
463}
464
465/// Stage-bound checkpoint publication/replay refusal; originals remain retained.
466#[derive(Debug, Error)]
467pub enum ExecutionStageCheckpointError {
468    /// Original workflow/stage or complete ancestor history cannot be admitted.
469    #[error(transparent)]
470    Stage(#[from] ExecutionWorkflowPersistenceError),
471    /// Complete original journal checkpoint derivation/publication failed.
472    #[error(transparent)]
473    Settlement(#[from] ExecutionSettlementCheckpointError),
474}
475
476/// Failed create-only preparation; retained records never grant repair or extra spending.
477#[derive(Debug, Error)]
478pub enum ExecutionStagePreparationError {
479    /// Original stage records, predecessor settlement or layout were rejected.
480    #[error(transparent)]
481    Stage(#[from] ExecutionWorkflowPersistenceError),
482    /// Complete original child authority derivation failed before allocation.
483    #[error(transparent)]
484    Authority(#[from] OperationPlanError),
485    /// Existing attempt-journal locking or durable create-only publication failed.
486    #[error(transparent)]
487    Journal(#[from] AttemptJournalError),
488}
489
490/// Complete retained stage-resume refusal; original evidence and spending remain unchanged.
491#[derive(Debug, Error)]
492pub enum ExecutionStageResumeError {
493    /// Original stage, ancestor history or layout cannot be admitted.
494    #[error(transparent)]
495    Stage(#[from] ExecutionWorkflowPersistenceError),
496    /// Complete original child journals, accounting or causality cannot be admitted.
497    #[error(transparent)]
498    Progress(#[from] ExecutionProgressPersistenceError),
499}
500fn read_binding(
501    layout: &BackupLayoutGuard,
502    workflow: &ExecutionWorkflowRecord,
503    sequence: u64,
504    expected: &ArtifactChecksumRecord,
505) -> Result<(ExecutionStageBindingRecord, OperationPlanRecord), ExecutionWorkflowPersistenceError> {
506    layout.check_root()?;
507    let binding: ExecutionStageBindingRecord =
508        read_json(&layout.root().join(BINDING_FILE), MAX_EXECUTION_STAGE_BYTES)?;
509    super::json::check_json_size(&binding, MAX_EXECUTION_STAGE_BYTES)?;
510    if binding.stage_sequence() != sequence || &binding.digest() != expected {
511        return Err(ExecutionWorkflowPersistenceError::DigestMismatch);
512    }
513    let plan = read_operation_plan(layout, binding.plan())?;
514    binding.validate(workflow, &plan)?;
515    Ok((binding, plan))
516}
517fn validate_predecessors(
518    layout: &BackupLayoutGuard,
519    workflow: &ExecutionWorkflowRecord,
520    binding: &ExecutionStageBindingRecord,
521) -> Result<(), ExecutionWorkflowPersistenceError> {
522    // Traverse the original acyclic catalog iteratively, admitting each ancestor
523    // once. Keep at most one row per catalog node, not one per dependency path.
524    // Learned evidence may differ between edges; binding/settlement identity cannot.
525    let mut expected: BTreeMap<u64, ExecutionStagePredecessorRecord> = binding
526        .predecessors()
527        .iter()
528        .map(|row| (row.stage_sequence(), row.clone()))
529        .collect();
530    let mut pending: Vec<_> = expected.keys().copied().collect();
531    while let Some(sequence) = pending.pop() {
532        let row = &expected[&sequence];
533        let original = {
534            // Release this layout and every journal lock before admitting another
535            // ancestor, including an ancestor shared by multiple branches.
536            let predecessor = acquire_stage(layout, sequence)?;
537            let (original, _) = read_binding(&predecessor, workflow, sequence, row.binding())?;
538            read_execution_settlement(&predecessor, original.plan(), row.settlement())?;
539            original
540        };
541        for ancestor in original.predecessors() {
542            match expected.entry(ancestor.stage_sequence()) {
543                Entry::Vacant(entry) => {
544                    pending.push(ancestor.stage_sequence());
545                    entry.insert(ancestor.clone());
546                }
547                Entry::Occupied(entry) => {
548                    if entry.get().binding() != ancestor.binding()
549                        || entry.get().settlement() != ancestor.settlement()
550                    {
551                        return Err(ExecutionWorkflowPersistenceError::DigestMismatch);
552                    }
553                }
554            }
555        }
556    }
557    Ok(())
558}
559fn stage_path(layout: &BackupLayoutGuard, sequence: u64) -> PathBuf {
560    layout.root().join(format!("execution-stage-{sequence}"))
561}
562fn acquire_stage(
563    layout: &BackupLayoutGuard,
564    sequence: u64,
565) -> Result<BackupLayoutGuard, ExecutionWorkflowPersistenceError> {
566    layout.check_root()?;
567    let path = stage_path(layout, sequence);
568    let metadata = fs::symlink_metadata(&path).map_err(PersistenceError::from)?;
569    if !metadata.is_dir() {
570        return Err(ExecutionWorkflowPersistenceError::UnsafeStage);
571    }
572    let stage = BackupLayoutGuard::acquire(&path)?;
573    // Unlike an explicitly operator-selected root, a derived stage symlink is
574    // never adopted. Noncooperating namespace stability remains caller-owned.
575    if stage.root() != path {
576        return Err(ExecutionWorkflowPersistenceError::UnsafeStage);
577    }
578    layout.check_root()?;
579    Ok(stage)
580}
581fn create_private_directory(
582    path: &std::path::Path,
583) -> Result<(), ExecutionWorkflowPersistenceError> {
584    #[cfg(unix)]
585    {
586        use std::os::unix::fs::DirBuilderExt;
587        fs::DirBuilder::new()
588            .mode(0o700)
589            .create(path)
590            .map_err(PersistenceError::from)?;
591        Ok(())
592    }
593    #[cfg(not(unix))]
594    {
595        let _ = path;
596        Err(PersistenceError::Io(std::io::Error::from(std::io::ErrorKind::Unsupported)).into())
597    }
598}
599
600/// Typed immutable workflow/stage admission failure; no recovery evidence is removed.
601#[derive(Debug, Error)]
602pub enum ExecutionWorkflowPersistenceError {
603    /// Workflow/stage identity differs from the retained expected digest.
604    #[error("execution workflow/stage digest mismatch")]
605    DigestMismatch,
606    /// A derived stage directory is a symlink or another unsafe entry.
607    #[error("unsafe execution stage directory")]
608    UnsafeStage,
609    /// Original model allocation or learned binding was rejected.
610    #[error(transparent)]
611    Model(#[from] ExecutionWorkflowError),
612    /// Existing immutable operation-plan admission failed.
613    #[error(transparent)]
614    Plan(#[from] OperationPlanPersistenceError),
615    /// Existing complete chronological Applied settlement admission failed.
616    #[error(transparent)]
617    Settlement(#[from] ExecutionSettlementPersistenceError),
618    /// Cooperating layout or journal exclusion failed.
619    #[error(transparent)]
620    Lock(#[from] JournalLockError),
621    /// Existing bounded JSON/durable filesystem owner failed.
622    #[error(transparent)]
623    Persistence(#[from] PersistenceError),
624}
625
626#[cfg(all(test, unix))]
627mod tests;
ic_backup/ops/persistence/execution_workflow/mod.rs

ic_backup/ops/persistence/execution_workflow/
mod.rs
