ic_backup/ops/persistence/execution_workflow/
mod.rs

1//! Fixed immutable stage admission using the existing layout, plan and settlement owners.
2
3mod ancestry;
4mod progress;
5mod stage;
6use ancestry::{
7    acquire_stage, read_admitted_binding, read_binding, stage_path, validate_predecessors,
8};
9pub use progress::{ExecutionWorkflowProgressError, read_execution_workflow_progress};
10pub use stage::ExecutionStageGuard;
11#[cfg(test)]
12use stage::StagePreparationBarrier;
13
14use super::{
15    AttemptJournalError, AttemptJournalGuard, BackupLayoutGuard, ExecutionProgressPersistenceError,
16    ExecutionSettlementCheckpointError, ExecutionSettlementPersistenceError, JournalLock,
17    JournalLockError, OperationPlanPersistenceError, PersistenceError, read_execution_progress,
18    read_execution_settlement, read_json, read_operation_plan, reconcile_execution_settlement,
19    replay_execution_settlement,
20};
21use crate::model::{
22    artifacts::ArtifactChecksumRecord,
23    execution_workflow::{
24        ExecutionStageBindingRecord, ExecutionStagePredecessorRecord, ExecutionWorkflowError,
25        ExecutionWorkflowRecord, MAX_EXECUTION_STAGE_BYTES, MAX_EXECUTION_WORKFLOW_BYTES,
26    },
27    operation_plan::{OperationPlanError, OperationPlanRecord},
28};
29use crate::policy::execution_progress::ExecutionProgressView;
30use std::{
31    collections::{BTreeMap, BTreeSet, btree_map::Entry},
32    fs,
33    path::PathBuf,
34};
35use thiserror::Error;
36
37const WORKFLOW_FILE: &str = "execution-workflow.json";
38const BINDING_FILE: &str = "stage-binding.json";
39
40/// Exact local identities of every original stage's retained checkpoint.
41///
42/// This is a sequential local projection, not an immutable product completion
43/// receipt. It authenticates no learned evidence, remote effect, application
44/// safety or command/byte custody and grants no terminal or reference release.
45#[derive(Clone, Debug, Eq, PartialEq)]
46pub struct ExecutionWorkflowSettlementView {
47    workflow: ArtifactChecksumRecord,
48    stages: Vec<ExecutionStageSettlementView>,
49}
50impl ExecutionWorkflowSettlementView {
51    /// Read the exact original workflow declaration identity.
52    #[must_use]
53    pub const fn workflow(&self) -> &ArtifactChecksumRecord {
54        &self.workflow
55    }
56    /// Read one row per original stage in canonical allocation order.
57    #[must_use]
58    pub fn stages(&self) -> &[ExecutionStageSettlementView] {
59        &self.stages
60    }
61}
62
63/// One original stage's bound plan and complete chronological local checkpoint.
64#[derive(Clone, Debug, Eq, PartialEq)]
65pub struct ExecutionStageSettlementView {
66    stage_sequence: u64,
67    binding: ArtifactChecksumRecord,
68    settlement: ArtifactChecksumRecord,
69}
70impl ExecutionStageSettlementView {
71    /// Read the original opaque stage identity.
72    #[must_use]
73    pub const fn stage_sequence(&self) -> u64 {
74        self.stage_sequence
75    }
76    /// Read the exact retained binding identity.
77    #[must_use]
78    pub const fn binding(&self) -> &ArtifactChecksumRecord {
79        &self.binding
80    }
81    /// Read the complete original journal-history checkpoint identity.
82    #[must_use]
83    pub const fn settlement(&self) -> &ArtifactChecksumRecord {
84        &self.settlement
85    }
86}
87
88/// Durably create the full original workflow allocation without replacement.
89/// # Errors
90/// Rejects occupied/unsafe paths, bounds, ownership and publication failures.
91pub fn create_execution_workflow(
92    layout: &BackupLayoutGuard,
93    record: &ExecutionWorkflowRecord,
94) -> Result<(), ExecutionWorkflowPersistenceError> {
95    layout.check_root()?;
96    let bytes = super::json::encode_json(record, MAX_EXECUTION_WORKFLOW_BYTES)?;
97    super::immutable_record::create_encoded_record(layout, WORKFLOW_FILE, &bytes)
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    super::immutable_record::read(
107        layout,
108        WORKFLOW_FILE,
109        MAX_EXECUTION_WORKFLOW_BYTES,
110        |record: &ExecutionWorkflowRecord| &record.digest() == expected,
111        || ExecutionWorkflowPersistenceError::DigestMismatch,
112    )
113}
114
115/// Read which allocated stage paths are already occupied under the held layout.
116///
117/// This is a namespace preflight for a retained run, not evidence that an absent
118/// stage has never dispatched. An occupied partial stage remains occupied and is
119/// never repaired. Derived symlinks and non-directories refuse admission.
120/// # Errors
121/// Rejects changed workflow/layout, unsafe stage entries and IO failures.
122pub(crate) fn read_execution_stage_occupancy(
123    layout: &BackupLayoutGuard,
124    expected: &ArtifactChecksumRecord,
125) -> Result<BTreeSet<u64>, ExecutionWorkflowPersistenceError> {
126    let workflow = read_execution_workflow(layout, expected)?;
127    let mut occupied = BTreeSet::new();
128    for stage in workflow.stages() {
129        let sequence = stage.operation_sequence();
130        match fs::symlink_metadata(stage_path(layout, sequence)) {
131            Ok(metadata) if metadata.is_dir() => {
132                occupied.insert(sequence);
133            }
134            Ok(_) => return Err(ExecutionWorkflowPersistenceError::UnsafeStage),
135            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
136            Err(error) => return Err(PersistenceError::from(error).into()),
137        }
138    }
139    read_execution_workflow(layout, expected)?;
140    Ok(occupied)
141}
142
143/// Replay every allocated stage's exact retained binding and all-Applied checkpoint.
144///
145/// The original workflow determines the complete stage set. Every fixed stage must
146/// exist with a valid child plan, immutable binding, complete original journal set
147/// and exact published settlement. Every binding's predecessor identities must
148/// match this complete original set. Stage locks are held one at a time under
149/// the caller's workflow layout guard. No absent stage means zero spending, and no
150/// occupied checkpoint is repaired or republished. The resulting digest rows
151/// are local evidence only; integration-owned authenticity, complete transfer,
152/// application safety, command quiescence and terminal/release proof remain separate.
153/// # Errors
154/// Rejects missing, unsafe, changed, held or unsettled original evidence without effects.
155pub fn read_complete_execution_workflow(
156    layout: &BackupLayoutGuard,
157    expected: &ArtifactChecksumRecord,
158) -> Result<ExecutionWorkflowSettlementView, ExecutionStageCheckpointError> {
159    read_complete_with(layout, expected, || {
160        read_execution_workflow(layout, expected)
161    })
162}
163
164/// Replay complete original settlement while retaining the exact workflow journal lock.
165///
166/// Terminal-evidence providers may use this instead of reacquiring their held journal
167/// lock. The guard must protect this layout's original `execution-workflow.json`.
168/// Both opening and closing reads retain the ordinary bounds, strict decoding and
169/// expected original digest checks; child locks and checkpoint admission are unchanged.
170/// This supplies local settlement only, never application or reference-release authority.
171/// # Errors
172/// Rejects a guard for another journal, changed originals, or incomplete settlement.
173pub fn read_complete_execution_workflow_locked(
174    layout: &BackupLayoutGuard,
175    journal: &JournalLock,
176    expected: &ArtifactChecksumRecord,
177) -> Result<ExecutionWorkflowSettlementView, ExecutionStageCheckpointError> {
178    if journal.journal() != layout.root().join(WORKFLOW_FILE) {
179        return Err(ExecutionWorkflowPersistenceError::DigestMismatch.into());
180    }
181    read_complete_with(layout, expected, || {
182        layout.check_root()?;
183        let workflow: ExecutionWorkflowRecord =
184            read_json(journal.journal(), MAX_EXECUTION_WORKFLOW_BYTES)?;
185        super::json::check_json_size(&workflow, MAX_EXECUTION_WORKFLOW_BYTES)?;
186        if &workflow.digest() != expected {
187            return Err(ExecutionWorkflowPersistenceError::DigestMismatch);
188        }
189        Ok(workflow)
190    })
191}
192
193fn read_complete_with(
194    layout: &BackupLayoutGuard,
195    expected: &ArtifactChecksumRecord,
196    read_original: impl Fn() -> Result<ExecutionWorkflowRecord, ExecutionWorkflowPersistenceError>,
197) -> Result<ExecutionWorkflowSettlementView, ExecutionStageCheckpointError> {
198    let workflow = read_original()?;
199    let mut stages = Vec::with_capacity(workflow.stages().len());
200    for allocated in workflow.stages() {
201        let sequence = allocated.operation_sequence();
202        let stage = acquire_stage(layout, sequence)?;
203        let binding: ExecutionStageBindingRecord =
204            read_json(&stage.root().join(BINDING_FILE), MAX_EXECUTION_STAGE_BYTES)
205                .map_err(ExecutionWorkflowPersistenceError::from)?;
206        let (binding, plan) = read_binding(&stage, &workflow, sequence, &binding.digest())?;
207        let settlement = replay_execution_settlement(&stage, &plan.digest())?;
208        read_binding(&stage, &workflow, sequence, &binding.digest())?;
209        let view = ExecutionStageSettlementView {
210            stage_sequence: sequence,
211            binding: binding.digest(),
212            settlement: settlement.digest(),
213        };
214        stages.push((binding, view));
215    }
216    for (binding, _) in &stages {
217        for predecessor in binding.predecessors() {
218            // The original catalog supplies canonical sequence order.
219            let index = stages
220                .binary_search_by_key(&predecessor.stage_sequence(), |(_, view)| {
221                    view.stage_sequence()
222                })
223                .map_err(|_| ExecutionWorkflowPersistenceError::DigestMismatch)?;
224            let actual = &stages[index].1;
225            if predecessor.binding() != actual.binding()
226                || predecessor.settlement() != actual.settlement()
227            {
228                return Err(ExecutionWorkflowPersistenceError::DigestMismatch.into());
229            }
230        }
231    }
232    read_original()?;
233    Ok(ExecutionWorkflowSettlementView {
234        workflow: expected.clone(),
235        stages: stages.into_iter().map(|(_, view)| view).collect(),
236    })
237}
238
239/// Stage-bound checkpoint publication/replay refusal; originals remain retained.
240#[derive(Debug, Error)]
241pub enum ExecutionStageCheckpointError {
242    /// Original workflow/stage or complete ancestor history cannot be admitted.
243    #[error(transparent)]
244    Stage(#[from] ExecutionWorkflowPersistenceError),
245    /// Complete original journal checkpoint derivation/publication failed.
246    #[error(transparent)]
247    Settlement(#[from] ExecutionSettlementCheckpointError),
248}
249
250/// Failed create-only preparation; retained records never grant repair or extra spending.
251#[derive(Debug, Error)]
252pub enum ExecutionStagePreparationError {
253    /// Original stage records, predecessor settlement or layout were rejected.
254    #[error(transparent)]
255    Stage(#[from] ExecutionWorkflowPersistenceError),
256    /// Complete original child authority derivation failed before allocation.
257    #[error(transparent)]
258    Authority(#[from] OperationPlanError),
259    /// Existing attempt-journal locking or durable create-only publication failed.
260    #[error(transparent)]
261    Journal(#[from] AttemptJournalError),
262}
263
264/// Complete retained stage-resume refusal; original evidence and spending remain unchanged.
265#[derive(Debug, Error)]
266pub enum ExecutionStageResumeError {
267    /// Original stage, ancestor history or layout cannot be admitted.
268    #[error(transparent)]
269    Stage(#[from] ExecutionWorkflowPersistenceError),
270    /// Complete original child journals, accounting or causality cannot be admitted.
271    #[error(transparent)]
272    Progress(#[from] ExecutionProgressPersistenceError),
273}
274/// Typed immutable workflow/stage admission failure; no recovery evidence is removed.
275#[derive(Debug, Error)]
276pub enum ExecutionWorkflowPersistenceError {
277    /// Workflow/stage identity differs from the retained expected digest.
278    #[error("execution workflow/stage digest mismatch")]
279    DigestMismatch,
280    /// A derived stage directory is a symlink or another unsafe entry.
281    #[error("unsafe execution stage directory")]
282    UnsafeStage,
283    /// Original model allocation or learned binding was rejected.
284    #[error(transparent)]
285    Model(#[from] ExecutionWorkflowError),
286    /// Existing immutable operation-plan admission failed.
287    #[error(transparent)]
288    Plan(#[from] OperationPlanPersistenceError),
289    /// Existing complete chronological Applied settlement admission failed.
290    #[error(transparent)]
291    Settlement(#[from] ExecutionSettlementPersistenceError),
292    /// Cooperating layout or journal exclusion failed.
293    #[error(transparent)]
294    Lock(#[from] JournalLockError),
295    /// Existing bounded JSON/durable filesystem owner failed.
296    #[error(transparent)]
297    Persistence(#[from] PersistenceError),
298}
299
300#[cfg(all(test, unix))]
301mod tests;
ic_backup/ops/persistence/execution_workflow/mod.rs

ic_backup/ops/persistence/execution_workflow/
mod.rs
