ic_backup/policy/execution_progress/
mod.rs1use 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#[derive(Clone, Debug)]
15pub struct ExecutionProgressRequest<'a> {
16 pub plan: &'a OperationPlanRecord,
18 pub journals: &'a [&'a AttemptJournalRecord],
20}
21
22#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
24#[serde(rename_all = "snake_case")]
25pub enum OperationProgressState {
26 AwaitingDependencies,
28 MutationAvailable,
30 MutationExhausted,
32 MutationUnresolved,
34 ObservationUnresolved,
36 ReconciliationExhausted,
38 Applied,
40}
41
42#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
44pub struct OperationProgressView {
45 pub operation_sequence: u64,
47 pub state: OperationProgressState,
49 pub attempts: AttemptJournalView,
51}
52
53#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
55pub struct ExecutionAttemptTotalsView {
56 pub mutations_used: u32,
58 pub observations_used: u32,
60 pub mutations_remaining: u32,
62 pub observations_remaining: u32,
64}
65
66#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
71pub struct ExecutionProgressView {
72 pub intent: ArtifactChecksumRecord,
74 pub graph: ArtifactChecksumRecord,
76 pub applied_operations: usize,
78 pub attempts: ExecutionAttemptTotalsView,
80 pub operations: Vec<OperationProgressView>,
82}
83
84pub fn progress(
95 request: &ExecutionProgressRequest<'_>,
96) -> Result<ExecutionProgressView, ExecutionProgressError> {
97 Ok(progress_with_journals(request)?.0)
98}
99
100pub(super) fn progress_with_journals<'a>(
103 request: &ExecutionProgressRequest<'a>,
104) -> Result<
105 (
106 ExecutionProgressView,
107 BTreeMap<u64, &'a AttemptJournalRecord>,
108 ),
109 ExecutionProgressError,
110> {
111 if request.journals.len() > MAX_EFFECT_OPERATIONS {
112 return Err(ExecutionProgressError::TooManyJournals);
113 }
114 let plan = request.plan;
115 let intent = plan.digest();
116 let mut journals = BTreeMap::new();
117 for journal in request.journals {
118 let sequence = journal.authority().binding().operation_sequence();
119 let operation = plan
120 .operation(sequence)
121 .map_err(|_| ExecutionProgressError::UnknownOperation(sequence))?;
122 if journals.contains_key(&sequence) {
123 return Err(ExecutionProgressError::DuplicateJournal(sequence));
124 }
125 if !plan.matches_authority(operation, &intent, journal.authority()) {
126 return Err(ExecutionProgressError::AuthorityMismatch(sequence));
127 }
128 journals.insert(sequence, *journal);
129 }
130 for operation in plan.operations() {
131 if !journals.contains_key(&operation.operation_sequence()) {
132 return Err(ExecutionProgressError::MissingJournal(
133 operation.operation_sequence(),
134 ));
135 }
136 }
137 let is_applied = |sequence: &u64| {
138 journals
139 .get(sequence)
140 .is_some_and(|journal| journal.applied_receipt().is_some())
141 };
142 let mut totals = ExecutionAttemptTotalsView {
143 mutations_used: 0,
144 observations_used: 0,
145 mutations_remaining: 0,
146 observations_remaining: 0,
147 };
148 let mut operations = Vec::with_capacity(journals.len());
149 for node in plan.graph().ordered_nodes() {
150 let sequence = node.operation_sequence();
151 let attempts = journals
153 .get(&sequence)
154 .ok_or(ExecutionProgressError::MissingJournal(sequence))?
155 .view();
156 let unmet = node
157 .depends_on()
158 .iter()
159 .find(|dependency| !is_applied(dependency));
160 if attempts.mutations_used != 0
161 && let Some(prerequisite) = unmet
162 {
163 return Err(ExecutionProgressError::PrematureAttempt {
164 operation_sequence: sequence,
165 prerequisite: *prerequisite,
166 });
167 }
168 totals.add(&attempts)?;
169 operations.push(OperationProgressView {
170 operation_sequence: sequence,
171 state: condition(&attempts, unmet.is_some()),
172 attempts,
173 });
174 }
175 Ok((
176 ExecutionProgressView {
177 intent,
178 graph: plan.graph().digest(),
179 applied_operations: journals
180 .keys()
181 .filter(|sequence| is_applied(sequence))
182 .count(),
183 attempts: totals,
184 operations,
185 },
186 journals,
187 ))
188}
189impl ExecutionAttemptTotalsView {
190 fn add(&mut self, attempts: &AttemptJournalView) -> Result<(), ExecutionProgressError> {
191 self.mutations_used = sum(self.mutations_used, attempts.mutations_used)?;
192 self.observations_used = sum(self.observations_used, attempts.observations_used)?;
193 self.mutations_remaining = sum(self.mutations_remaining, attempts.mutations_remaining)?;
194 self.observations_remaining =
195 sum(self.observations_remaining, attempts.observations_remaining)?;
196 Ok(())
197 }
198}
199fn sum(left: u32, right: u32) -> Result<u32, ExecutionProgressError> {
200 left.checked_add(right)
201 .ok_or(ExecutionProgressError::AccountingOverflow)
202}
203fn condition(attempts: &AttemptJournalView, unmet_dependencies: bool) -> OperationProgressState {
204 if attempts.applied {
205 OperationProgressState::Applied
206 } else if attempts.pending_observation.is_some() {
207 OperationProgressState::ObservationUnresolved
208 } else if attempts.pending_mutation.is_some() {
209 if attempts.observations_remaining == 0 {
210 OperationProgressState::ReconciliationExhausted
211 } else {
212 OperationProgressState::MutationUnresolved
213 }
214 } else if unmet_dependencies {
215 OperationProgressState::AwaitingDependencies
216 } else if attempts.mutations_remaining == 0 {
217 OperationProgressState::MutationExhausted
218 } else {
219 OperationProgressState::MutationAvailable
220 }
221}
222
223#[derive(Debug, Error, Eq, PartialEq)]
225pub enum ExecutionProgressError {
226 #[error("execution progress exceeds {MAX_EFFECT_OPERATIONS} journals")]
228 TooManyJournals,
229 #[error("duplicate journal for operation {0}")]
231 DuplicateJournal(u64),
232 #[error("journal operation {0} is absent from the original plan")]
234 UnknownOperation(u64),
235 #[error("missing original journal for operation {0}")]
237 MissingJournal(u64),
238 #[error("original journal authority mismatch for operation {0}")]
240 AuthorityMismatch(u64),
241 #[error(
243 "operation {operation_sequence} was attempted without applied prerequisite {prerequisite}"
244 )]
245 PrematureAttempt {
246 operation_sequence: u64,
248 prerequisite: u64,
250 },
251 #[error("execution attempt accounting overflow")]
253 AccountingOverflow,
254}
255
256#[cfg(test)]
257mod tests;