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

ic_backup/ops/persistence/execution_workflow/
mod.rs
