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

ic_backup/workflow/execution_workflow/
mod.rs
