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, mut completed) = admit_retained_stages(layout, expected_workflow, retained)?;
108    let mut advanced = 0;
109    for node in workflow.allocation().ordered_operations() {
110        let sequence = node.operation_sequence();
111        if completed.contains_key(&sequence) {
112            continue;
113        }
114        if advanced == stage_limit.get() {
115            return Ok(ExecutionWorkflowAdvance::Paused);
116        }
117        let predecessors = node
118            .depends_on()
119            .iter()
120            .map(|dependency| {
121                completed.get(dependency).cloned().ok_or(
122                    ExecutionWorkflowRunError::<D::Error>::MissingPredecessor(*dependency),
123                )
124            })
125            .collect::<Result<Vec<_>, ExecutionWorkflowRunError<D::Error>>>()?;
126        let checkpoint =
127            execute_absent_stage(layout, &workflow, sequence, predecessors, driver).await?;
128        completed.insert(sequence, checkpoint);
129        advanced += 1;
130    }
131    Ok(ExecutionWorkflowAdvance::Complete(
132        read_complete_execution_workflow(layout, expected_workflow)?,
133    ))
134}
135
136fn admit_retained_stages<E: std::error::Error + 'static>(
137    layout: &BackupLayoutGuard,
138    expected_workflow: &ArtifactChecksumRecord,
139    retained: &BTreeMap<u64, RetainedStageEvidence>,
140) -> Result<
141    (
142        ExecutionWorkflowRecord,
143        BTreeMap<u64, ExecutionStagePredecessorRecord>,
144    ),
145    ExecutionWorkflowRunError<E>,
146> {
147    let workflow = read_execution_workflow(layout, expected_workflow)?;
148    let occupied = read_execution_stage_occupancy(layout, expected_workflow)?;
149    if !retained.keys().copied().eq(occupied.iter().copied()) {
150        return Err(ExecutionWorkflowRunError::RetainedStageSetMismatch);
151    }
152    let mut completed = BTreeMap::new();
153    for (&sequence, evidence) in retained {
154        let stage =
155            ExecutionStageGuard::open(layout, expected_workflow, sequence, &evidence.binding)?;
156        for predecessor in stage.binding().predecessors() {
157            let Some(original) = retained.get(&predecessor.stage_sequence()) else {
158                return Err(ExecutionWorkflowRunError::RetainedStageSetMismatch);
159            };
160            if &original.learned_evidence != predecessor.learned_evidence() {
161                return Err(ExecutionWorkflowRunError::RetainedLearnedMismatch(sequence));
162            }
163        }
164        let checkpoint = stage.replay_checkpoint(evidence.learned_evidence.clone())?;
165        completed.insert(sequence, checkpoint);
166    }
167    // Recheck the namespace before invoking any integration callback. Create-only
168    // stage preparation still refuses an occupied path after this preflight.
169    if read_execution_stage_occupancy(layout, expected_workflow)? != occupied {
170        return Err(ExecutionWorkflowRunError::RetainedStageSetMismatch);
171    }
172    Ok((workflow, completed))
173}
174
175async fn execute_absent_stage<D: ExecutionWorkflowStageDriver>(
176    layout: &BackupLayoutGuard,
177    workflow: &ExecutionWorkflowRecord,
178    sequence: u64,
179    predecessors: Vec<ExecutionStagePredecessorRecord>,
180    driver: &mut D,
181) -> Result<ExecutionStagePredecessorRecord, ExecutionWorkflowRunError<D::Error>> {
182    let plan = driver
183        .bind_stage(workflow, sequence, &predecessors)
184        .map_err(ExecutionWorkflowRunError::Driver)?;
185    let binding = ExecutionStageBindingRecord::new(workflow, sequence, &plan, predecessors)?;
186    let stage = ExecutionStageGuard::prepare(layout, binding, plan)?;
187    let learned = driver
188        .execute_stage(&stage)
189        .await
190        .map_err(ExecutionWorkflowRunError::Driver)?;
191    Ok(stage.reconcile_checkpoint(learned)?)
192}
193
194/// Failure of new original-workflow coordination; all originals remain retained.
195#[derive(Debug, Error)]
196pub enum ExecutionWorkflowRunError<E: std::error::Error + 'static> {
197    /// A validated predecessor was not available in the current run.
198    #[error("original workflow predecessor {0} did not complete")]
199    MissingPredecessor(u64),
200    /// Supplied identities do not cover exactly the occupied original stage paths.
201    #[error("retained stage identities do not match occupied original stages")]
202    RetainedStageSetMismatch,
203    /// An occupied successor froze a different learned predecessor commitment,
204    #[error("retained stage {0} has a different learned predecessor commitment")]
205    RetainedLearnedMismatch(u64),
206    /// Original allocation could not be published or was already occupied.
207    #[error(transparent)]
208    Workflow(#[from] ExecutionWorkflowPersistenceError),
209    /// A driver-supplied child violated the fixed original allocation.
210    #[error(transparent)]
211    Binding(#[from] ExecutionWorkflowError),
212    /// Complete child preparation failed without replacing original evidence.
213    #[error(transparent)]
214    Preparation(#[from] ExecutionStagePreparationError),
215    /// Driver-owned planning, qualification, provider or receipt step failed.
216    #[error("original workflow stage driver failed: {0}")]
217    Driver(#[source] E),
218    /// Canonical all-Applied checkpoint or final local replay failed.
219    #[error(transparent)]
220    Checkpoint(#[from] ExecutionStageCheckpointError),
221}
222
223#[cfg(test)]
224mod tests;
ic_backup/workflow/execution_workflow/mod.rs

ic_backup/workflow/execution_workflow/
mod.rs
