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

ic_backup/workflow/execution_workflow/
mod.rs
