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

ic_backup/ops/persistence/execution_workflow/
mod.rs
