ic_backup/policy/execution_progress/
mod.rs

1//! Pure original-plan progress from exact retained journals; no scheduling or dispatch.
2
3use crate::model::{
4    artifacts::ArtifactChecksumRecord,
5    attempt_journal::{AttemptJournalRecord, AttemptJournalView},
6    effect_graph::MAX_EFFECT_OPERATIONS,
7    operation_plan::OperationPlanRecord,
8};
9use serde::Serialize;
10use std::collections::BTreeMap;
11use thiserror::Error;
12
13/// Passive complete journal set supplied by the local evidence-owning caller.
14#[derive(Clone, Debug)]
15pub struct ExecutionProgressRequest<'a> {
16    /// Original immutable declaration; its hash binds every journal's intent.
17    pub plan: &'a OperationPlanRecord,
18    /// Exactly one validated retained journal per declared operation, in any order.
19    pub journals: &'a [&'a AttemptJournalRecord],
20}
21
22/// Derived local attempt condition, without fresh permission or backend qualification.
23#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
24#[serde(rename_all = "snake_case")]
25pub enum OperationProgressState {
26    /// At least one declared prerequisite lacks retained Applied evidence.
27    AwaitingDependencies,
28    /// Prerequisites are Applied and original mutation allowance remains.
29    MutationAvailable,
30    /// No unresolved mutation or Applied receipt, and mutation allowance is exhausted.
31    MutationExhausted,
32    /// A mutation is unresolved and original reconciliation allowance remains.
33    MutationUnresolved,
34    /// An observation reservation remains unresolved, even if its allowance is exhausted.
35    ObservationUnresolved,
36    /// A mutation remains unresolved with no pending observation or remaining observation allowance.
37    ReconciliationExhausted,
38    /// This operation has retained integration-qualified Applied evidence.
39    Applied,
40}
41
42/// Read-only local progress for one exact operation in graph planning order.
43#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
44pub struct OperationProgressView {
45    /// Original opaque operation identity, never an array index.
46    pub operation_sequence: u64,
47    /// Derived local condition; none of these values authorize dispatch.
48    pub state: OperationProgressState,
49    /// Exact retained accounting, authority digest and unresolved attempt identities.
50    pub attempts: AttemptJournalView,
51}
52
53/// Sums over original assigned allowances only; unused plan headroom is excluded.
54#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
55pub struct ExecutionAttemptTotalsView {
56    /// Mutation reservations consumed across all original journals, without refunds.
57    pub mutations_used: u32,
58    /// Observation reservations consumed across all original journals, without refunds.
59    pub observations_used: u32,
60    /// Sum of remaining originally assigned mutation allowances.
61    pub mutations_remaining: u32,
62    /// Sum of remaining originally assigned observation allowances.
63    pub observations_remaining: u32,
64}
65
66/// Read-only projection bound to the complete original declaration and retained journals.
67///
68/// Applied evidence is caller-qualified and does not prove full run completion,
69/// current IC state, artifact durability, application safety or reference release.
70#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
71pub struct ExecutionProgressView {
72    /// Full canonical original plan intent.
73    pub intent: ArtifactChecksumRecord,
74    /// Original explicit dependency graph digest.
75    pub graph: ArtifactChecksumRecord,
76    /// Number of operations with retained Applied receipts, not a terminal proof.
77    pub applied_operations: usize,
78    /// Derived totals over the original per-operation ceilings.
79    pub attempts: ExecutionAttemptTotalsView,
80    /// Exact journal conditions in deterministic graph planning order.
81    pub operations: Vec<OperationProgressView>,
82}
83
84impl ExecutionProgressView {
85    pub(crate) const fn is_unattempted(&self) -> bool {
86        self.attempts.mutations_used == 0 && self.attempts.observations_used == 0
87    }
88
89    pub(crate) fn all_applied(&self) -> bool {
90        self.applied_operations == self.operations.len()
91    }
92}
93
94/// Admit exact complete journal coverage and project causal retained local progress.
95///
96/// Missing journals are errors, never fresh zero-consumption declarations. Any
97/// attempted operation requires Applied evidence for every declared prerequisite.
98/// This validates retained causality, not cross-journal dispatch chronology: no
99/// remote observations, authority probes, writes or scheduling occur.
100///
101/// # Errors
102/// Rejects excessive, duplicate, unknown, missing or mismatched journals, premature
103/// attempts and accounting overflow. Unresolved/exhausted states remain visible.
104pub fn progress(
105    request: &ExecutionProgressRequest<'_>,
106) -> Result<ExecutionProgressView, ExecutionProgressError> {
107    Ok(progress_with_journals(request.plan, request.journals.iter().copied())?.0)
108}
109
110// Retain the admitted original index for exact-history settlement, so that it
111// consumes the same coverage/authority proof as the public progress projection.
112pub(crate) fn progress_with_journals<'a>(
113    plan: &OperationPlanRecord,
114    originals: impl ExactSizeIterator<Item = &'a AttemptJournalRecord>,
115) -> Result<
116    (
117        ExecutionProgressView,
118        BTreeMap<u64, &'a AttemptJournalRecord>,
119    ),
120    ExecutionProgressError,
121> {
122    let (intent, journals) = admit_journals(plan, originals)?;
123    let is_applied = |sequence: &u64| {
124        journals
125            .get(sequence)
126            .is_some_and(|journal| journal.applied_receipt().is_some())
127    };
128    let mut totals = ExecutionAttemptTotalsView {
129        mutations_used: 0,
130        observations_used: 0,
131        mutations_remaining: 0,
132        observations_remaining: 0,
133    };
134    let mut operations = Vec::with_capacity(journals.len());
135    for node in plan.graph().ordered_nodes() {
136        let sequence = node.operation_sequence();
137        // Closed graph/table identity and complete coverage were admitted above.
138        let attempts = journals
139            .get(&sequence)
140            .ok_or(ExecutionProgressError::MissingJournal(sequence))?
141            .view();
142        let unmet = node
143            .depends_on()
144            .iter()
145            .find(|dependency| !is_applied(dependency));
146        if attempts.mutations_used != 0
147            && let Some(prerequisite) = unmet
148        {
149            return Err(ExecutionProgressError::PrematureAttempt {
150                operation_sequence: sequence,
151                prerequisite: *prerequisite,
152            });
153        }
154        totals.add(&attempts)?;
155        operations.push(OperationProgressView {
156            operation_sequence: sequence,
157            state: condition(&attempts, unmet.is_some()),
158            attempts,
159        });
160    }
161    let applied_operations = operations
162        .iter()
163        .filter(|operation| operation.attempts.applied)
164        .count();
165    Ok((
166        ExecutionProgressView {
167            intent,
168            graph: plan.graph().digest(),
169            applied_operations,
170            attempts: totals,
171            operations,
172        },
173        journals,
174    ))
175}
176// Admit complete original coverage before deriving any causal state or totals.
177fn admit_journals<'a>(
178    plan: &OperationPlanRecord,
179    originals: impl ExactSizeIterator<Item = &'a AttemptJournalRecord>,
180) -> Result<
181    (
182        ArtifactChecksumRecord,
183        BTreeMap<u64, &'a AttemptJournalRecord>,
184    ),
185    ExecutionProgressError,
186> {
187    if originals.len() > MAX_EFFECT_OPERATIONS {
188        return Err(ExecutionProgressError::TooManyJournals);
189    }
190    let intent = plan.digest();
191    let mut journals = BTreeMap::new();
192    for journal in originals {
193        let sequence = journal.authority().binding().operation_sequence();
194        let operation = plan
195            .operation(sequence)
196            .map_err(|_| ExecutionProgressError::UnknownOperation(sequence))?;
197        if journals.contains_key(&sequence) {
198            return Err(ExecutionProgressError::DuplicateJournal(sequence));
199        }
200        if !plan.matches_authority(operation, &intent, journal.authority()) {
201            return Err(ExecutionProgressError::AuthorityMismatch(sequence));
202        }
203        journals.insert(sequence, journal);
204    }
205    for operation in plan.operations() {
206        if !journals.contains_key(&operation.operation_sequence()) {
207            return Err(ExecutionProgressError::MissingJournal(
208                operation.operation_sequence(),
209            ));
210        }
211    }
212    Ok((intent, journals))
213}
214
215impl ExecutionAttemptTotalsView {
216    fn add(&mut self, attempts: &AttemptJournalView) -> Result<(), ExecutionProgressError> {
217        self.mutations_used = sum(self.mutations_used, attempts.mutations_used)?;
218        self.observations_used = sum(self.observations_used, attempts.observations_used)?;
219        self.mutations_remaining = sum(self.mutations_remaining, attempts.mutations_remaining)?;
220        self.observations_remaining =
221            sum(self.observations_remaining, attempts.observations_remaining)?;
222        Ok(())
223    }
224}
225fn sum(left: u32, right: u32) -> Result<u32, ExecutionProgressError> {
226    left.checked_add(right)
227        .ok_or(ExecutionProgressError::AccountingOverflow)
228}
229fn condition(attempts: &AttemptJournalView, unmet_dependencies: bool) -> OperationProgressState {
230    if attempts.applied {
231        OperationProgressState::Applied
232    } else if attempts.pending_observation.is_some() {
233        OperationProgressState::ObservationUnresolved
234    } else if attempts.pending_mutation.is_some() {
235        if attempts.observations_remaining == 0 {
236            OperationProgressState::ReconciliationExhausted
237        } else {
238            OperationProgressState::MutationUnresolved
239        }
240    } else if unmet_dependencies {
241        OperationProgressState::AwaitingDependencies
242    } else if attempts.mutations_remaining == 0 {
243        OperationProgressState::MutationExhausted
244    } else {
245        OperationProgressState::MutationAvailable
246    }
247}
248
249/// Typed original-plan journal admission, retained causality or accounting failure.
250#[derive(Debug, Error, Eq, PartialEq)]
251pub enum ExecutionProgressError {
252    /// The journal input exceeds the original bounded operation count.
253    #[error("execution progress exceeds {MAX_EFFECT_OPERATIONS} journals")]
254    TooManyJournals,
255    /// Multiple supplied journals claim the same original operation identity.
256    #[error("duplicate journal for operation {0}")]
257    DuplicateJournal(u64),
258    /// A supplied journal's operation is absent from the original plan.
259    #[error("journal operation {0} is absent from the original plan")]
260    UnknownOperation(u64),
261    /// Retained evidence for an original operation was not supplied.
262    #[error("missing original journal for operation {0}")]
263    MissingJournal(u64),
264    /// Intent, context, target, request or original allowance differs from the plan.
265    #[error("original journal authority mismatch for operation {0}")]
266    AuthorityMismatch(u64),
267    /// A reservation exists while a declared prerequisite lacks Applied evidence.
268    #[error(
269        "operation {operation_sequence} was attempted without applied prerequisite {prerequisite}"
270    )]
271    PrematureAttempt {
272        /// Operation with retained consumed mutation allowance.
273        operation_sequence: u64,
274        /// Exact declared prerequisite lacking retained Applied evidence.
275        prerequisite: u64,
276    },
277    /// Derived accounting could not be represented without overflow.
278    #[error("execution attempt accounting overflow")]
279    AccountingOverflow,
280}
281
282#[cfg(test)]
283mod tests;
ic_backup/policy/execution_progress/mod.rs

ic_backup/policy/execution_progress/
mod.rs
