ic_backup/workflow/execution_workflow/
mod.rs

1//! Original-workflow stage coordination without a default IC or application adapter.
2
3use crate::{
4    model::{
5        artifacts::ArtifactChecksumRecord,
6        execution_workflow::{
7            ExecutionStageBindingRecord, ExecutionStagePredecessorRecord, ExecutionWorkflowError,
8            ExecutionWorkflowRecord,
9        },
10        operation_plan::OperationPlanRecord,
11    },
12    ops::persistence::{
13        BackupLayoutGuard, ExecutionStageCheckpointError, ExecutionStageGuard,
14        ExecutionStagePreparationError, ExecutionWorkflowPersistenceError,
15        ExecutionWorkflowSettlementView, read_complete_execution_workflow,
16        read_execution_stage_occupancy, read_execution_workflow,
17    },
18};
19use std::{collections::BTreeMap, future::Future, num::NonZeroUsize};
20use thiserror::Error;
21
22/// Integration-owned exact child planning and one-shot execution of each original stage.
23///
24/// `bind_stage` must qualify learned inputs against the supplied original predecessors.
25/// `execute_stage` owns fresh effect-boundary admission, provider calls, authenticated
26/// request/reply retention and explicit original Applied receipts. Its returned digest
27/// is opaque learned evidence; the coordinator does not authenticate it. Neither
28/// callback may treat local checkpoint admission as a dispatch or release permit.
29/// Return learned evidence without publishing a checkpoint: advancement owns that
30/// publication once, after the callback releases its attempt guards.
31pub trait ExecutionWorkflowStageDriver {
32    /// Integration-owned planning, admission or execution error.
33    type Error: std::error::Error + 'static;
34
35    /// Build one exact child plan under the original stage allocation.
36    /// # Errors
37    /// Returns integration-owned learned-input or child-planning rejection.
38    fn bind_stage(
39        &mut self,
40        workflow: &ExecutionWorkflowRecord,
41        stage_sequence: u64,
42        predecessors: &[ExecutionStagePredecessorRecord],
43    ) -> Result<OperationPlanRecord, Self::Error>;
44
45    /// Execute only the already prepared original child; retain original effects.
46    fn execute_stage(
47        &mut self,
48        stage: &ExecutionStageGuard<'_>,
49    ) -> impl Future<Output = Result<ArtifactChecksumRecord, Self::Error>>;
50}
51
52/// Integration-qualified identities for an already settled original stage.
53///
54/// Local replay derives its binding and settlement from retained records. The
55/// learned digest is supplied by the integration, which must authenticate the
56/// original response and its relation to the settled attempt before resuming.
57#[derive(Clone, Debug, Eq, PartialEq)]
58pub struct RetainedStageEvidence {
59    binding: ArtifactChecksumRecord,
60    learned_evidence: ArtifactChecksumRecord,
61}
62impl RetainedStageEvidence {
63    /// Read the qualified learned input for an enclosing stage driver.
64    #[must_use]
65    pub const fn learned_evidence(&self) -> &ArtifactChecksumRecord {
66        &self.learned_evidence
67    }
68
69    /// Supply an original binding identity and independently qualified learned input.
70    #[must_use]
71    pub const fn new(
72        binding: ArtifactChecksumRecord,
73        learned_evidence: ArtifactChecksumRecord,
74    ) -> Self {
75        Self {
76            binding,
77            learned_evidence,
78        }
79    }
80}
81
82/// Bounded operator advancement, without a second progress or spending ledger.
83#[derive(Clone, Debug, Eq, PartialEq)]
84pub enum ExecutionWorkflowAdvance {
85    /// The selected bound was reached between settled stages; original absent stages remain.
86    Paused,
87    /// Every original stage's exact retained checkpoint passed local replay.
88    /// This grants no product terminal or source-reference release authority.
89    Complete(ExecutionWorkflowSettlementView),
90}
91
92/// Advance at most `stage_limit` absent stages of one prepared original workflow.
93///
94/// Prepare once with `create_execution_workflow`, retaining the original declarations
95/// before effects. Every continuation uses the same complete occupied-stage evidence
96/// admission. A pause happens only between settled stages,
97/// never halfway through a transfer. The limit bounds stage callbacks, not each stage's
98/// original finite effect allowance or elapsed time. All admission, custody, durable
99/// reply retention and effect-specific qualification remain mandatory driver duties.
100///
101/// Pending observations/mutations, partial preparation or changed originals refuse
102/// before any callback; the bound cannot reset them or supply resend permission.
103/// A failed/cancelled callback retains original uncertainty. Inspect it through
104/// `read_execution_workflow_progress` and reconcile through the original owners.
105/// Complete replay invokes no driver callbacks, regardless of the selected limit.
106/// # Errors
107/// Retains the existing typed preparation, checkpoint, original-evidence and driver failures.
108pub async fn advance_execution_workflow<D: ExecutionWorkflowStageDriver>(
109    layout: &BackupLayoutGuard,
110    expected_workflow: &ArtifactChecksumRecord,
111    retained: &BTreeMap<u64, RetainedStageEvidence>,
112    stage_limit: NonZeroUsize,
113    driver: &mut D,
114) -> Result<ExecutionWorkflowAdvance, ExecutionWorkflowRunError<D::Error>> {
115    let (workflow, mut completed) = admit_retained_stages(layout, expected_workflow, retained)?;
116    let mut advanced = 0;
117    for node in workflow.allocation().ordered_operations() {
118        let sequence = node.operation_sequence();
119        if completed.contains_key(&sequence) {
120            continue;
121        }
122        if advanced == stage_limit.get() {
123            return Ok(ExecutionWorkflowAdvance::Paused);
124        }
125        let predecessors = node
126            .depends_on()
127            .iter()
128            .map(|dependency| {
129                completed.get(dependency).cloned().ok_or(
130                    ExecutionWorkflowRunError::<D::Error>::MissingPredecessor(*dependency),
131                )
132            })
133            .collect::<Result<Vec<_>, ExecutionWorkflowRunError<D::Error>>>()?;
134        let checkpoint =
135            execute_absent_stage(layout, &workflow, sequence, predecessors, driver).await?;
136        completed.insert(sequence, checkpoint);
137        advanced += 1;
138    }
139    Ok(ExecutionWorkflowAdvance::Complete(
140        read_complete_execution_workflow(layout, expected_workflow)?,
141    ))
142}
143
144fn admit_retained_stages<E: std::error::Error + 'static>(
145    layout: &BackupLayoutGuard,
146    expected_workflow: &ArtifactChecksumRecord,
147    retained: &BTreeMap<u64, RetainedStageEvidence>,
148) -> Result<
149    (
150        ExecutionWorkflowRecord,
151        BTreeMap<u64, ExecutionStagePredecessorRecord>,
152    ),
153    ExecutionWorkflowRunError<E>,
154> {
155    let workflow = read_execution_workflow(layout, expected_workflow)?;
156    let occupied = read_execution_stage_occupancy(layout, expected_workflow)?;
157    if !retained.keys().copied().eq(occupied.iter().copied()) {
158        return Err(ExecutionWorkflowRunError::RetainedStageSetMismatch);
159    }
160    let mut completed = BTreeMap::new();
161    for (&sequence, evidence) in retained {
162        let stage =
163            ExecutionStageGuard::open(layout, expected_workflow, sequence, &evidence.binding)?;
164        for predecessor in stage.binding().predecessors() {
165            let Some(original) = retained.get(&predecessor.stage_sequence()) else {
166                return Err(ExecutionWorkflowRunError::RetainedStageSetMismatch);
167            };
168            if &original.learned_evidence != predecessor.learned_evidence() {
169                return Err(ExecutionWorkflowRunError::RetainedLearnedMismatch(sequence));
170            }
171        }
172        let checkpoint = stage.replay_checkpoint(evidence.learned_evidence.clone())?;
173        completed.insert(sequence, checkpoint);
174    }
175    // Recheck the namespace before invoking any integration callback. Create-only
176    // stage preparation still refuses an occupied path after this preflight.
177    if read_execution_stage_occupancy(layout, expected_workflow)? != occupied {
178        return Err(ExecutionWorkflowRunError::RetainedStageSetMismatch);
179    }
180    Ok((workflow, completed))
181}
182
183async fn execute_absent_stage<D: ExecutionWorkflowStageDriver>(
184    layout: &BackupLayoutGuard,
185    workflow: &ExecutionWorkflowRecord,
186    sequence: u64,
187    predecessors: Vec<ExecutionStagePredecessorRecord>,
188    driver: &mut D,
189) -> Result<ExecutionStagePredecessorRecord, ExecutionWorkflowRunError<D::Error>> {
190    let plan = driver
191        .bind_stage(workflow, sequence, &predecessors)
192        .map_err(ExecutionWorkflowRunError::Driver)?;
193    let binding = ExecutionStageBindingRecord::new(workflow, sequence, &plan, predecessors)?;
194    let stage = ExecutionStageGuard::prepare(layout, binding, plan)?;
195    let learned = driver
196        .execute_stage(&stage)
197        .await
198        .map_err(ExecutionWorkflowRunError::Driver)?;
199    Ok(stage.reconcile_checkpoint(learned)?)
200}
201
202/// Failure of new original-workflow coordination; all originals remain retained.
203#[derive(Debug, Error)]
204pub enum ExecutionWorkflowRunError<E: std::error::Error + 'static> {
205    /// A validated predecessor was not available in the current run.
206    #[error("original workflow predecessor {0} did not complete")]
207    MissingPredecessor(u64),
208    /// Supplied identities do not cover exactly the occupied original stage paths.
209    #[error("retained stage identities do not match occupied original stages")]
210    RetainedStageSetMismatch,
211    /// An occupied successor froze a different learned predecessor commitment,
212    #[error("retained stage {0} has a different learned predecessor commitment")]
213    RetainedLearnedMismatch(u64),
214    /// Original allocation could not be published or was already occupied.
215    #[error(transparent)]
216    Workflow(#[from] ExecutionWorkflowPersistenceError),
217    /// A driver-supplied child violated the fixed original allocation.
218    #[error(transparent)]
219    Binding(#[from] ExecutionWorkflowError),
220    /// Complete child preparation failed without replacing original evidence.
221    #[error(transparent)]
222    Preparation(#[from] ExecutionStagePreparationError),
223    /// Driver-owned planning, qualification, provider or receipt step failed.
224    #[error("original workflow stage driver failed: {0}")]
225    Driver(#[source] E),
226    /// Canonical all-Applied checkpoint or final local replay failed.
227    #[error(transparent)]
228    Checkpoint(#[from] ExecutionStageCheckpointError),
229}
230
231#[cfg(test)]
232mod tests;
ic_backup/workflow/execution_workflow/mod.rs

ic_backup/workflow/execution_workflow/
mod.rs
