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    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, 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#[derive(Clone, Copy, Debug, Eq, PartialEq)]
31enum StagePreparationBarrier {
32    Plan,
33    Binding,
34    Journal(u64),
35}
36
37/// Durably create the full original workflow allocation without replacement.
38/// # Errors
39/// Rejects occupied/unsafe paths, bounds, ownership and publication failures.
40pub fn create_execution_workflow(
41    layout: &BackupLayoutGuard,
42    record: &ExecutionWorkflowRecord,
43) -> Result<(), ExecutionWorkflowPersistenceError> {
44    layout.check_root()?;
45    super::json::check_json_size(record, MAX_EXECUTION_WORKFLOW_BYTES)?;
46    let path = layout.root().join(WORKFLOW_FILE);
47    let _lock = JournalLock::acquire(&path)?;
48    create_json_durable(&path, record)?;
49    Ok(())
50}
51/// Read exact original workflow allocation; missing evidence never supplies a new budget.
52/// # Errors
53/// Rejects missing/changed/unsafe/oversized records and ownership failures.
54pub fn read_execution_workflow(
55    layout: &BackupLayoutGuard,
56    expected: &ArtifactChecksumRecord,
57) -> Result<ExecutionWorkflowRecord, ExecutionWorkflowPersistenceError> {
58    layout.check_root()?;
59    let path = layout.root().join(WORKFLOW_FILE);
60    let _lock = JournalLock::acquire(&path)?;
61    let record: ExecutionWorkflowRecord = read_json(&path, MAX_EXECUTION_WORKFLOW_BYTES)?;
62    super::json::check_json_size(&record, MAX_EXECUTION_WORKFLOW_BYTES)?;
63    if &record.digest() != expected {
64        return Err(ExecutionWorkflowPersistenceError::DigestMismatch);
65    }
66    Ok(record)
67}
68
69/// One original stage's fixed layout, retained under both workflow and child exclusion.
70///
71/// Existing attempt journals remain the only spending owner. `create` supplies no
72/// journals; `prepare` creates the complete original set. Reopening never creates,
73/// resets or reconstructs journals. Integrations
74/// retain exact learned inputs, authenticate their relation to settled originals,
75/// and qualify fresh authority, command custody and proof of no previous dispatch.
76#[derive(Debug)]
77pub struct ExecutionStageGuard<'a> {
78    workflow_layout: &'a BackupLayoutGuard,
79    stage_layout: BackupLayoutGuard,
80    binding: ExecutionStageBindingRecord,
81    plan: OperationPlanRecord,
82}
83impl<'a> ExecutionStageGuard<'a> {
84    /// Reconstruct an exact predecessor after a lost checkpoint response.
85    ///
86    /// Derives the settlement identity from all retained original journals and
87    /// reads the already published checkpoint. It never creates a checkpoint,
88    /// journal or receipt. The supplied learned evidence remains integration-owned
89    /// and must be independently qualified before a successor is dispatched.
90    /// # Errors
91    /// Rejects missing, held, unsettled or changed originals and checkpoint drift.
92    pub fn replay_checkpoint(
93        &self,
94        learned_evidence: ArtifactChecksumRecord,
95    ) -> Result<ExecutionStagePredecessorRecord, ExecutionStageCheckpointError> {
96        let settlement = replay_execution_settlement(self.layout()?, &self.plan.digest())?;
97        self.layout()?;
98        Ok(ExecutionStagePredecessorRecord::new(
99            self.binding.stage_sequence(),
100            self.binding.digest(),
101            settlement.digest(),
102            learned_evidence,
103        ))
104    }
105    /// Publish the original all-Applied checkpoint and derive its exact predecessor commitment.
106    ///
107    /// Re-admits the workflow, stage and complete ancestor histories before and after
108    /// delegating to the canonical original-journal checkpoint owner. Hold no attempt
109    /// guards. Failed post-publication admission retains the checkpoint and all originals;
110    /// an occupied checkpoint is never replaced. Identity-bound local replay remains
111    /// owned by `read_execution_settlement`.
112    ///
113    /// The supplied learned evidence stays integration-owned: this authenticates no
114    /// learned input, receipt or effect, and grants no dispatch or terminal/release proof.
115    /// # Errors
116    /// Rejects changed/missing/held originals, unsettled journals and publication failures.
117    pub fn checkpoint(
118        &self,
119        learned_evidence: ArtifactChecksumRecord,
120    ) -> Result<ExecutionStagePredecessorRecord, ExecutionStageCheckpointError> {
121        self.checkpoint_with(learned_evidence, || {})
122    }
123    fn checkpoint_with(
124        &self,
125        learned_evidence: ArtifactChecksumRecord,
126        after_publication: impl FnOnce(),
127    ) -> Result<ExecutionStagePredecessorRecord, ExecutionStageCheckpointError> {
128        let settlement = checkpoint_execution_settlement(self.layout()?, &self.plan.digest())?;
129        after_publication();
130        self.layout()?;
131        Ok(ExecutionStagePredecessorRecord::new(
132            self.binding.stage_sequence(),
133            self.binding.digest(),
134            settlement.digest(),
135            learned_evidence,
136        ))
137    }
138    /// Resume an exact retained stage with its complete original journal progress.
139    ///
140    /// Reuses record-only `open` and canonical complete-journal admission, then
141    /// rechecks the original stage and ancestor histories before returning. Hold
142    /// no attempt guards. Reads only retained local records, without creating
143    /// journals, reading artifact trees or calling providers. Pending and exhausted
144    /// spending is retained.
145    /// The returned progress is a sequential local projection, not atomic custody,
146    /// fresh permission, dispatch, authenticated receipt or terminal/release proof.
147    /// # Errors
148    /// Rejects missing/changed/unsafe originals, journal contention and invalid
149    /// retained accounting/causality without repair or replenishing allowance.
150    pub fn resume(
151        workflow_layout: &'a BackupLayoutGuard,
152        expected_workflow: &ArtifactChecksumRecord,
153        sequence: u64,
154        expected_binding: &ArtifactChecksumRecord,
155    ) -> Result<(Self, ExecutionProgressView), ExecutionStageResumeError> {
156        let stage = Self::open(
157            workflow_layout,
158            expected_workflow,
159            sequence,
160            expected_binding,
161        )?;
162        let progress = read_execution_progress(&stage.stage_layout, &stage.plan.digest())?;
163        stage.layout()?;
164        Ok((stage, progress))
165    }
166    /// Durably prepare a new stage and its complete original attempt-journal set.
167    ///
168    /// Reuses create-only stage admission and the existing journal owner, taking
169    /// one journal lock at a time. Every original authority is derived before
170    /// stage allocation. No reservation, receipt, backend call or fresh dispatch
171    /// authority follows. Hold no other attempt guards during preparation.
172    ///
173    /// Partial preparation remains occupied and is never repaired by retrying.
174    /// Reopen only retained originals and require complete execution admission;
175    /// missing journals never mean unused allowance. A lost successful response
176    /// can reopen the complete exact stage without creating another journal.
177    /// # Errors
178    /// Rejects invalid originals, occupied stages, journal contention and failed
179    /// durable publication, preserving all partial records and original limits.
180    pub fn prepare(
181        workflow_layout: &'a BackupLayoutGuard,
182        binding: ExecutionStageBindingRecord,
183        plan: OperationPlanRecord,
184    ) -> Result<Self, ExecutionStagePreparationError> {
185        Self::prepare_with(workflow_layout, binding, plan, |_| {})
186    }
187    fn prepare_with(
188        workflow_layout: &'a BackupLayoutGuard,
189        binding: ExecutionStageBindingRecord,
190        plan: OperationPlanRecord,
191        mut barrier: impl FnMut(StagePreparationBarrier),
192    ) -> Result<Self, ExecutionStagePreparationError> {
193        let authorities = plan.attempt_authorities()?;
194        let stage = Self::create_with(workflow_layout, binding, plan, &mut barrier)?;
195        let layout = stage.layout()?;
196        for authority in authorities {
197            let sequence = authority.binding().operation_sequence();
198            drop(AttemptJournalGuard::create(layout, authority)?);
199            barrier(StagePreparationBarrier::Journal(sequence));
200        }
201        stage.layout()?;
202        Ok(stage)
203    }
204    /// Create a fixed private stage directory, exact child plan and immutable binding.
205    ///
206    /// Every original predecessor and ancestor needs its exact retained binding, plan and
207    /// complete chronological Applied settlement. The directory is create-only:
208    /// occupied or interrupted preparation is retained and cannot be recreated
209    /// with another plan. No journal or backend effect occurs before return.
210    /// # Errors
211    /// Rejects changed allocation, unavailable prerequisites, occupied stages and IO failures.
212    pub fn create(
213        workflow_layout: &'a BackupLayoutGuard,
214        binding: ExecutionStageBindingRecord,
215        plan: OperationPlanRecord,
216    ) -> Result<Self, ExecutionWorkflowPersistenceError> {
217        Self::create_with(workflow_layout, binding, plan, |_| {})
218    }
219    fn create_with(
220        workflow_layout: &'a BackupLayoutGuard,
221        binding: ExecutionStageBindingRecord,
222        plan: OperationPlanRecord,
223        mut barrier: impl FnMut(StagePreparationBarrier),
224    ) -> Result<Self, ExecutionWorkflowPersistenceError> {
225        let workflow = read_execution_workflow(workflow_layout, binding.workflow())?;
226        binding.validate(&workflow, &plan)?;
227        super::json::check_json_size(&binding, MAX_EXECUTION_STAGE_BYTES)?;
228        // Validate both documents before allocating any stage paths.
229        super::json::check_json_size(
230            &plan,
231            crate::model::operation_plan::MAX_OPERATION_PLAN_BYTES,
232        )?;
233        validate_predecessors(workflow_layout, &workflow, &binding)?;
234        let path = stage_path(workflow_layout, binding.stage_sequence());
235        create_private_directory(&path)?;
236        fs::File::open(workflow_layout.root())
237            .and_then(|directory| directory.sync_all())
238            .map_err(PersistenceError::from)?;
239        let stage_layout = acquire_stage(workflow_layout, binding.stage_sequence())?;
240        create_operation_plan(&stage_layout, &plan)?;
241        barrier(StagePreparationBarrier::Plan);
242        create_json_durable(&stage_layout.root().join(BINDING_FILE), &binding)?;
243        barrier(StagePreparationBarrier::Binding);
244        workflow_layout.check_root()?;
245        Ok(Self {
246            workflow_layout,
247            stage_layout,
248            binding,
249            plan,
250        })
251    }
252    /// Reopen an exact fixed stage without allocation, repair or journal creation.
253    ///
254    /// Lost successful creation responses reconcile here. A partially prepared
255    /// directory without both immutable records rejects and remains retained.
256    /// Missing pending journals cannot be treated as unused allowance by the
257    /// existing execution-progress or settlement owners.
258    /// # Errors
259    /// Rejects missing/changed original records, prerequisite history or ownership failures.
260    pub fn open(
261        workflow_layout: &'a BackupLayoutGuard,
262        expected_workflow: &ArtifactChecksumRecord,
263        sequence: u64,
264        expected_binding: &ArtifactChecksumRecord,
265    ) -> Result<Self, ExecutionWorkflowPersistenceError> {
266        let workflow = read_execution_workflow(workflow_layout, expected_workflow)?;
267        workflow
268            .stage(sequence)
269            .map_err(|_| ExecutionWorkflowError::UnknownStage(sequence))?;
270        let stage_layout = acquire_stage(workflow_layout, sequence)?;
271        let (binding, plan) = read_binding(&stage_layout, &workflow, sequence, expected_binding)?;
272        validate_predecessors(workflow_layout, &workflow, &binding)?;
273        workflow_layout.check_root()?;
274        Ok(Self {
275            workflow_layout,
276            stage_layout,
277            binding,
278            plan,
279        })
280    }
281    /// Read the exact bound original child plan; no fresh permission is implied.
282    #[must_use]
283    pub const fn plan(&self) -> &OperationPlanRecord {
284        &self.plan
285    }
286    /// Read the immutable original stage and predecessor commitments.
287    #[must_use]
288    pub const fn binding(&self) -> &ExecutionStageBindingRecord {
289        &self.binding
290    }
291    /// Re-admit both original records and expose the existing journal/layout owner.
292    ///
293    /// Callers must use complete original journal admission for resume. Never
294    /// create a missing journal on resume or substitute a freshly derived plan.
295    /// # Errors
296    /// Rejects changed roots, retained records or any ancestor settlement history.
297    pub fn layout(&self) -> Result<&BackupLayoutGuard, ExecutionWorkflowPersistenceError> {
298        let workflow = read_execution_workflow(self.workflow_layout, self.binding.workflow())?;
299        read_binding(
300            &self.stage_layout,
301            &workflow,
302            self.binding.stage_sequence(),
303            &self.binding.digest(),
304        )?;
305        validate_predecessors(self.workflow_layout, &workflow, &self.binding)?;
306        Ok(&self.stage_layout)
307    }
308}
309
310/// Stage-bound checkpoint publication/replay refusal; originals remain retained.
311#[derive(Debug, Error)]
312pub enum ExecutionStageCheckpointError {
313    /// Original workflow/stage or complete ancestor history cannot be admitted.
314    #[error(transparent)]
315    Stage(#[from] ExecutionWorkflowPersistenceError),
316    /// Complete original journal checkpoint derivation/publication failed.
317    #[error(transparent)]
318    Settlement(#[from] ExecutionSettlementCheckpointError),
319}
320
321/// Failed create-only preparation; retained records never grant repair or extra spending.
322#[derive(Debug, Error)]
323pub enum ExecutionStagePreparationError {
324    /// Original stage records, predecessor settlement or layout were rejected.
325    #[error(transparent)]
326    Stage(#[from] ExecutionWorkflowPersistenceError),
327    /// Complete original child authority derivation failed before allocation.
328    #[error(transparent)]
329    Authority(#[from] OperationPlanError),
330    /// Existing attempt-journal locking or durable create-only publication failed.
331    #[error(transparent)]
332    Journal(#[from] AttemptJournalError),
333}
334
335/// Complete retained stage-resume refusal; original evidence and spending remain unchanged.
336#[derive(Debug, Error)]
337pub enum ExecutionStageResumeError {
338    /// Original stage, ancestor history or layout cannot be admitted.
339    #[error(transparent)]
340    Stage(#[from] ExecutionWorkflowPersistenceError),
341    /// Complete original child journals, accounting or causality cannot be admitted.
342    #[error(transparent)]
343    Progress(#[from] ExecutionProgressPersistenceError),
344}
345fn read_binding(
346    layout: &BackupLayoutGuard,
347    workflow: &ExecutionWorkflowRecord,
348    sequence: u64,
349    expected: &ArtifactChecksumRecord,
350) -> Result<(ExecutionStageBindingRecord, OperationPlanRecord), ExecutionWorkflowPersistenceError> {
351    layout.check_root()?;
352    let binding: ExecutionStageBindingRecord =
353        read_json(&layout.root().join(BINDING_FILE), MAX_EXECUTION_STAGE_BYTES)?;
354    super::json::check_json_size(&binding, MAX_EXECUTION_STAGE_BYTES)?;
355    if binding.stage_sequence() != sequence || &binding.digest() != expected {
356        return Err(ExecutionWorkflowPersistenceError::DigestMismatch);
357    }
358    let plan = read_operation_plan(layout, binding.plan())?;
359    binding.validate(workflow, &plan)?;
360    Ok((binding, plan))
361}
362fn validate_predecessors(
363    layout: &BackupLayoutGuard,
364    workflow: &ExecutionWorkflowRecord,
365    binding: &ExecutionStageBindingRecord,
366) -> Result<(), ExecutionWorkflowPersistenceError> {
367    // Traverse the original acyclic catalog iteratively, admitting each ancestor
368    // once. Keep at most one row per catalog node, not one per dependency path.
369    // Learned evidence may differ between edges; binding/settlement identity cannot.
370    let mut expected: BTreeMap<u64, ExecutionStagePredecessorRecord> = binding
371        .predecessors()
372        .iter()
373        .map(|row| (row.stage_sequence(), row.clone()))
374        .collect();
375    let mut pending: Vec<_> = expected.keys().copied().collect();
376    while let Some(sequence) = pending.pop() {
377        let row = &expected[&sequence];
378        let original = {
379            // Release this layout and every journal lock before admitting another
380            // ancestor, including an ancestor shared by multiple branches.
381            let predecessor = acquire_stage(layout, sequence)?;
382            let (original, _) = read_binding(&predecessor, workflow, sequence, row.binding())?;
383            read_execution_settlement(&predecessor, original.plan(), row.settlement())?;
384            original
385        };
386        for ancestor in original.predecessors() {
387            match expected.entry(ancestor.stage_sequence()) {
388                Entry::Vacant(entry) => {
389                    pending.push(ancestor.stage_sequence());
390                    entry.insert(ancestor.clone());
391                }
392                Entry::Occupied(entry) => {
393                    if entry.get().binding() != ancestor.binding()
394                        || entry.get().settlement() != ancestor.settlement()
395                    {
396                        return Err(ExecutionWorkflowPersistenceError::DigestMismatch);
397                    }
398                }
399            }
400        }
401    }
402    Ok(())
403}
404fn stage_path(layout: &BackupLayoutGuard, sequence: u64) -> PathBuf {
405    layout.root().join(format!("execution-stage-{sequence}"))
406}
407fn acquire_stage(
408    layout: &BackupLayoutGuard,
409    sequence: u64,
410) -> Result<BackupLayoutGuard, ExecutionWorkflowPersistenceError> {
411    layout.check_root()?;
412    let path = stage_path(layout, sequence);
413    let metadata = fs::symlink_metadata(&path).map_err(PersistenceError::from)?;
414    if !metadata.is_dir() {
415        return Err(ExecutionWorkflowPersistenceError::UnsafeStage);
416    }
417    let stage = BackupLayoutGuard::acquire(&path)?;
418    // Unlike an explicitly operator-selected root, a derived stage symlink is
419    // never adopted. Noncooperating namespace stability remains caller-owned.
420    if stage.root() != path {
421        return Err(ExecutionWorkflowPersistenceError::UnsafeStage);
422    }
423    layout.check_root()?;
424    Ok(stage)
425}
426fn create_private_directory(
427    path: &std::path::Path,
428) -> Result<(), ExecutionWorkflowPersistenceError> {
429    #[cfg(unix)]
430    {
431        use std::os::unix::fs::DirBuilderExt;
432        fs::DirBuilder::new()
433            .mode(0o700)
434            .create(path)
435            .map_err(PersistenceError::from)?;
436        Ok(())
437    }
438    #[cfg(not(unix))]
439    {
440        let _ = path;
441        Err(PersistenceError::Io(std::io::Error::from(std::io::ErrorKind::Unsupported)).into())
442    }
443}
444
445/// Typed immutable workflow/stage admission failure; no recovery evidence is removed.
446#[derive(Debug, Error)]
447pub enum ExecutionWorkflowPersistenceError {
448    /// Workflow/stage identity differs from the retained expected digest.
449    #[error("execution workflow/stage digest mismatch")]
450    DigestMismatch,
451    /// A derived stage directory is a symlink or another unsafe entry.
452    #[error("unsafe execution stage directory")]
453    UnsafeStage,
454    /// Original model allocation or learned binding was rejected.
455    #[error(transparent)]
456    Model(#[from] ExecutionWorkflowError),
457    /// Existing immutable operation-plan admission failed.
458    #[error(transparent)]
459    Plan(#[from] OperationPlanPersistenceError),
460    /// Existing complete chronological Applied settlement admission failed.
461    #[error(transparent)]
462    Settlement(#[from] ExecutionSettlementPersistenceError),
463    /// Cooperating layout or journal exclusion failed.
464    #[error(transparent)]
465    Lock(#[from] JournalLockError),
466    /// Existing bounded JSON/durable filesystem owner failed.
467    #[error(transparent)]
468    Persistence(#[from] PersistenceError),
469}
470
471#[cfg(all(test, unix))]
472mod tests;
ic_backup/ops/persistence/execution_workflow/mod.rs

ic_backup/ops/persistence/execution_workflow/
mod.rs
