1mod manifest;
2mod operations;
3mod types;
4
5pub use types::*;
6
7use crate::{
8 execution::{
9 BackupExecutionJournal, BackupExecutionOperationReceipt, BackupExecutionOperationState,
10 },
11 persistence::{BackupLayout, CommandLifetimeLock, CommandLifetimeLockError, JournalLock},
12 plan::{BackupOperationKind, BackupPlan},
13 timestamp::{current_timestamp_marker, state_updated_at, timestamp_marker, timestamp_seconds},
14};
15use operations::execute_operation_receipt;
16
17const PREFLIGHT_TTL_SECONDS: u64 = 300;
18
19pub fn backup_run_execute_with_executor(
21 config: &BackupRunnerConfig,
22 executor: &mut impl BackupRunnerExecutor,
23) -> Result<BackupRunResponse, BackupRunnerError> {
24 let layout = BackupLayout::new(config.out.clone());
25 let _lock = JournalLock::acquire(&layout.execution_journal_path())?;
26 let mut plan = layout.read_backup_plan()?;
27 let mut journal = if layout.execution_journal_path().is_file() {
28 layout.read_execution_journal()?
29 } else {
30 let journal = BackupExecutionJournal::from_plan(&plan)?;
31 layout.write_execution_journal(&journal)?;
32 journal
33 };
34 layout.verify_execution_integrity()?;
35
36 accept_preflight_if_needed(config, executor, &layout, &mut plan, &mut journal)?;
37 execute_ready_operations(config, executor, &layout, &plan, &mut journal)
38}
39
40fn accept_preflight_if_needed(
41 config: &BackupRunnerConfig,
42 executor: &mut impl BackupRunnerExecutor,
43 layout: &BackupLayout,
44 plan: &mut BackupPlan,
45 journal: &mut BackupExecutionJournal,
46) -> Result<(), BackupRunnerError> {
47 if journal.preflight_accepted {
48 return Ok(());
49 }
50
51 let validated_at = state_updated_at(config.updated_at.as_ref());
52 let expires_at = timestamp_marker(timestamp_seconds(&validated_at) + PREFLIGHT_TTL_SECONDS);
53 let preflight_id = format!("preflight-{}", plan.run_id);
54 let receipts = executor
55 .preflight_receipts(plan, &preflight_id, &validated_at, &expires_at)
56 .map_err(|error| BackupRunnerError::PreflightFailed {
57 status: error.status,
58 message: error.message,
59 })?;
60 plan.apply_execution_preflight_receipts(&receipts, &validated_at)?;
61 layout.write_backup_plan(plan)?;
62 journal.accept_preflight_receipts_at(&receipts, Some(validated_at))?;
63 layout.write_execution_journal(journal)?;
64 Ok(())
65}
66
67fn execute_ready_operations(
68 config: &BackupRunnerConfig,
69 executor: &mut impl BackupRunnerExecutor,
70 layout: &BackupLayout,
71 plan: &BackupPlan,
72 journal: &mut BackupExecutionJournal,
73) -> Result<BackupRunResponse, BackupRunnerError> {
74 let mut executed = Vec::new();
75
76 loop {
77 let summary = journal.resume_summary();
78 if summary.completed_operations + summary.skipped_operations == summary.total_operations {
79 return Ok(run_response(plan, journal, executed, false));
80 }
81 if config
82 .max_steps
83 .is_some_and(|max_steps| executed.len() >= max_steps)
84 {
85 return Ok(run_response(plan, journal, executed, true));
86 }
87
88 let operation = journal
89 .next_ready_operation()
90 .cloned()
91 .ok_or(BackupRunnerError::NoReadyOperation)?;
92 if operation.state == BackupExecutionOperationState::Blocked {
93 return Err(BackupRunnerError::Blocked {
94 reasons: operation.blocking_reasons,
95 });
96 }
97
98 let mut command_lock = backup_command_lock(layout, &operation)?;
99 if operation.state == BackupExecutionOperationState::Pending {
100 if operation.kind == BackupOperationKind::Stop {
101 if let Some(receipt) = reconcile_pending_stop(executor, journal, &operation)? {
102 finish_reconciled_command_lock(&operation, command_lock.take())?;
103 journal.record_operation_receipt(receipt)?;
104 layout.write_execution_journal(journal)?;
105 executed.push(BackupRunExecutedOperation::completed(&operation));
106 continue;
107 }
108 } else {
109 reject_unknown_backup_command_outcome(&operation, command_lock.take())?;
110 }
111 } else {
112 journal.mark_operation_pending_at(
113 operation.sequence,
114 Some(state_updated_at(config.updated_at.as_ref())),
115 )?;
116 layout.write_execution_journal(journal)?;
117 }
118
119 let operation_result = execute_operation_receipt(
120 config,
121 executor,
122 layout,
123 plan,
124 journal,
125 &operation,
126 command_lock.as_ref().map(CommandLifetimeLock::handle),
127 );
128 if let Some(command_lock) = command_lock {
129 command_lock
130 .finish()
131 .map_err(|error| backup_command_lock_error(&operation, error))?;
132 }
133
134 match operation_result {
135 Ok(receipt) => {
136 journal.record_operation_receipt(receipt)?;
137 layout.write_execution_journal(journal)?;
138 executed.push(BackupRunExecutedOperation::completed(&operation));
139 }
140 Err(error) => {
141 let receipt = crate::execution::BackupExecutionOperationReceipt::failed(
142 journal,
143 &operation,
144 Some(state_updated_at(config.updated_at.as_ref())),
145 error.to_string(),
146 );
147 journal.record_operation_receipt(receipt)?;
148 layout.write_execution_journal(journal)?;
149 executed.push(BackupRunExecutedOperation::failed(&operation));
150 return Err(error);
151 }
152 }
153 }
154}
155
156fn reconcile_pending_stop(
157 executor: &mut impl BackupRunnerExecutor,
158 journal: &BackupExecutionJournal,
159 operation: &crate::execution::BackupExecutionJournalOperation,
160) -> Result<Option<BackupExecutionOperationReceipt>, BackupRunnerError> {
161 let target = operation.target_canister_id.as_deref().ok_or(
162 BackupRunnerError::MissingOperationTarget {
163 sequence: operation.sequence,
164 },
165 )?;
166 let status = executor.canister_status(target).map_err(|error| {
167 BackupRunnerError::CanisterStatusFailed {
168 sequence: operation.sequence,
169 status: error.status,
170 message: error.message,
171 }
172 })?;
173 match status {
174 BackupRunnerCanisterStatus::Running => Ok(None),
175 BackupRunnerCanisterStatus::Stopped => {
176 Ok(Some(BackupExecutionOperationReceipt::completed(
177 journal,
178 operation,
179 Some(current_timestamp_marker()),
180 )))
181 }
182 BackupRunnerCanisterStatus::Stopping => Err(BackupRunnerError::CanisterStatusUnsettled {
183 sequence: operation.sequence,
184 operation_id: operation.operation_id.clone(),
185 status: status.label(),
186 }),
187 }
188}
189
190fn finish_reconciled_command_lock(
191 operation: &crate::execution::BackupExecutionJournalOperation,
192 command_lock: Option<CommandLifetimeLock>,
193) -> Result<(), BackupRunnerError> {
194 command_lock
195 .ok_or_else(|| BackupRunnerError::MissingCommandLifetime {
196 sequence: operation.sequence,
197 operation_id: operation.operation_id.clone(),
198 })?
199 .finish()
200 .map_err(|error| backup_command_lock_error(operation, error))
201}
202
203fn reject_unknown_backup_command_outcome(
204 operation: &crate::execution::BackupExecutionJournalOperation,
205 command_lock: Option<CommandLifetimeLock>,
206) -> Result<(), BackupRunnerError> {
207 let Some(command_lock) = command_lock else {
208 return Ok(());
209 };
210 let lock_path = command_lock.path().to_string_lossy().to_string();
211 command_lock
212 .finish()
213 .map_err(|error| backup_command_lock_error(operation, error))?;
214 Err(BackupRunnerError::CommandOutcomeUnknown {
215 sequence: operation.sequence,
216 operation_id: operation.operation_id.clone(),
217 lock_path,
218 })
219}
220
221fn backup_command_lock(
222 layout: &BackupLayout,
223 operation: &crate::execution::BackupExecutionJournalOperation,
224) -> Result<Option<CommandLifetimeLock>, BackupRunnerError> {
225 if !matches!(
226 operation.kind,
227 crate::plan::BackupOperationKind::Stop
228 | crate::plan::BackupOperationKind::CreateSnapshot
229 | crate::plan::BackupOperationKind::Start
230 | crate::plan::BackupOperationKind::DownloadSnapshot
231 ) {
232 return Ok(None);
233 }
234
235 CommandLifetimeLock::acquire(&layout.execution_journal_path(), operation.sequence)
236 .map(Some)
237 .map_err(|error| backup_command_lock_error(operation, error))
238}
239
240fn backup_command_lock_error(
241 operation: &crate::execution::BackupExecutionJournalOperation,
242 error: CommandLifetimeLockError,
243) -> BackupRunnerError {
244 match error {
245 CommandLifetimeLockError::InFlight { lock_path } => BackupRunnerError::CommandInFlight {
246 sequence: operation.sequence,
247 operation_id: operation.operation_id.clone(),
248 lock_path,
249 },
250 CommandLifetimeLockError::UnsafeEntry { lock_path, kind } => {
251 BackupRunnerError::CommandLockUnsafeEntry {
252 sequence: operation.sequence,
253 operation_id: operation.operation_id.clone(),
254 lock_path,
255 kind,
256 }
257 }
258 CommandLifetimeLockError::Io(error) => BackupRunnerError::Io(error),
259 }
260}
261
262fn run_response(
263 plan: &BackupPlan,
264 journal: &BackupExecutionJournal,
265 executed: Vec<BackupRunExecutedOperation>,
266 max_steps_reached: bool,
267) -> BackupRunResponse {
268 let execution = journal.resume_summary();
269 BackupRunResponse {
270 run_id: plan.run_id.clone(),
271 plan_id: plan.plan_id.clone(),
272 backup_id: plan.run_id.clone(),
273 complete: execution.completed_operations + execution.skipped_operations
274 == execution.total_operations,
275 max_steps_reached,
276 executed_operation_count: executed.len(),
277 executed_operations: executed,
278 execution,
279 }
280}
281
282#[cfg(test)]
283mod tests;