1mod manifest;
2mod operations;
3mod types;
4
5pub use types::*;
6
7use crate::{
8 execution::{
9 BackupExecutionJournal, BackupExecutionOperationReceipt,
10 BackupExecutionOperationReceiptOutcome, BackupExecutionOperationState,
11 },
12 persistence::{BackupLayout, CommandLifetimeLock, CommandLifetimeLockError, JournalLock},
13 plan::{BackupOperationKind, BackupPlan},
14 timestamp::{current_timestamp_marker, state_updated_at, timestamp_marker, timestamp_seconds},
15};
16use operations::{
17 execute_operation_receipt, operation_target, persist_created_snapshot,
18 reconcile_pending_download, recorded_snapshot_receipt,
19};
20use std::collections::{BTreeMap, BTreeSet};
21
22const PREFLIGHT_TTL_SECONDS: u64 = 300;
23
24pub fn backup_run_execute_with_executor(
26 config: &BackupRunnerConfig,
27 executor: &mut impl BackupRunnerExecutor,
28) -> Result<BackupRunResponse, BackupRunnerError> {
29 let layout = BackupLayout::new(config.out.clone());
30 let _lock = JournalLock::acquire(&layout.execution_journal_path())?;
31 let mut plan = layout.read_backup_plan()?;
32 let mut journal = if layout.execution_journal_path().is_file() {
33 layout.read_execution_journal()?
34 } else {
35 let journal = BackupExecutionJournal::from_plan(&plan)?;
36 layout.write_execution_journal(&journal)?;
37 journal
38 };
39 layout.verify_execution_integrity()?;
40
41 accept_preflight_if_needed(config, executor, &layout, &mut plan, &mut journal)?;
42 execute_ready_operations(config, executor, &layout, &plan, &mut journal)
43}
44
45fn accept_preflight_if_needed(
46 config: &BackupRunnerConfig,
47 executor: &mut impl BackupRunnerExecutor,
48 layout: &BackupLayout,
49 plan: &mut BackupPlan,
50 journal: &mut BackupExecutionJournal,
51) -> Result<(), BackupRunnerError> {
52 if journal.preflight_accepted {
53 return Ok(());
54 }
55
56 let validated_at = state_updated_at(config.updated_at.as_ref());
57 let expires_at = timestamp_marker(timestamp_seconds(&validated_at) + PREFLIGHT_TTL_SECONDS);
58 let preflight_id = format!("preflight-{}", plan.run_id);
59 let receipts = executor
60 .preflight_receipts(plan, &preflight_id, &validated_at, &expires_at)
61 .map_err(|error| BackupRunnerError::PreflightFailed {
62 status: error.status,
63 message: error.message,
64 })?;
65 plan.apply_execution_preflight_receipts(&receipts, &validated_at)?;
66 layout.write_backup_plan(plan)?;
67 journal.accept_preflight_receipts_at(&receipts, Some(validated_at))?;
68 layout.write_execution_journal(journal)?;
69 Ok(())
70}
71
72fn execute_ready_operations(
73 config: &BackupRunnerConfig,
74 executor: &mut impl BackupRunnerExecutor,
75 layout: &BackupLayout,
76 plan: &BackupPlan,
77 journal: &mut BackupExecutionJournal,
78) -> Result<BackupRunResponse, BackupRunnerError> {
79 let mut executed = Vec::new();
80
81 loop {
82 let summary = journal.resume_summary();
83 if summary.completed_operations + summary.skipped_operations == summary.total_operations {
84 return Ok(run_response(plan, journal, executed, false));
85 }
86 if config
87 .max_steps
88 .is_some_and(|max_steps| executed.len() >= max_steps)
89 {
90 return Ok(run_response(plan, journal, executed, true));
91 }
92
93 let operation = journal
94 .next_ready_operation()
95 .cloned()
96 .ok_or(BackupRunnerError::NoReadyOperation)?;
97 if operation.state == BackupExecutionOperationState::Blocked {
98 return Err(BackupRunnerError::Blocked {
99 reasons: operation.blocking_reasons,
100 });
101 }
102
103 let mut command_lock = backup_command_lock(layout, &operation)?;
104 let reconciled_receipt = if operation.state == BackupExecutionOperationState::Pending {
105 if let Some(completed_status) = lifecycle_completed_status(&operation.kind) {
106 reconcile_pending_lifecycle(executor, journal, &operation, completed_status)?
107 } else if operation.kind == BackupOperationKind::CreateSnapshot {
108 reconcile_pending_snapshot_create(executor, layout, plan, journal, &operation)?
109 } else if operation.kind == BackupOperationKind::DownloadSnapshot {
110 reconcile_pending_download(layout, journal, &operation)?
111 } else {
112 reject_unknown_backup_command_outcome(&operation, command_lock.take())?;
113 None
114 }
115 } else {
116 if let Some(completed_status) = lifecycle_completed_status(&operation.kind) {
117 prepare_lifecycle_attempt(
118 config,
119 executor,
120 layout,
121 journal,
122 &operation,
123 completed_status,
124 )?
125 } else if operation.kind == BackupOperationKind::CreateSnapshot {
126 prepare_snapshot_create_attempt(
127 config, executor, layout, plan, journal, &operation,
128 )?
129 } else {
130 journal.mark_operation_pending_at(
131 operation.sequence,
132 Some(state_updated_at(config.updated_at.as_ref())),
133 )?;
134 layout.write_execution_journal(journal)?;
135 None
136 }
137 };
138 if let Some(receipt) = reconciled_receipt {
139 finish_reconciled_command_lock(&operation, command_lock.take())?;
140 journal.record_operation_receipt(receipt)?;
141 layout.write_execution_journal(journal)?;
142 executed.push(BackupRunExecutedOperation::completed(&operation));
143 continue;
144 }
145
146 let operation_result = execute_operation_receipt(
147 config,
148 executor,
149 layout,
150 plan,
151 journal,
152 &operation,
153 command_lock.as_ref().map(CommandLifetimeLock::handle),
154 );
155 if let Some(command_lock) = command_lock {
156 command_lock
157 .finish()
158 .map_err(|error| backup_command_lock_error(&operation, error))?;
159 }
160
161 match operation_result {
162 Ok(receipt) => {
163 journal.record_operation_receipt(receipt)?;
164 layout.write_execution_journal(journal)?;
165 executed.push(BackupRunExecutedOperation::completed(&operation));
166 }
167 Err(error) => {
168 let receipt = crate::execution::BackupExecutionOperationReceipt::failed(
169 journal,
170 &operation,
171 Some(state_updated_at(config.updated_at.as_ref())),
172 error.to_string(),
173 );
174 journal.record_operation_receipt(receipt)?;
175 layout.write_execution_journal(journal)?;
176 executed.push(BackupRunExecutedOperation::failed(&operation));
177 return Err(error);
178 }
179 }
180 }
181}
182
183fn prepare_lifecycle_attempt(
184 config: &BackupRunnerConfig,
185 executor: &mut impl BackupRunnerExecutor,
186 layout: &BackupLayout,
187 journal: &mut BackupExecutionJournal,
188 operation: &crate::execution::BackupExecutionJournalOperation,
189 completed_status: BackupRunnerCanisterStatus,
190) -> Result<Option<BackupExecutionOperationReceipt>, BackupRunnerError> {
191 let reconcile_previous_attempt = operation.state == BackupExecutionOperationState::Failed
192 || journal.operation_receipts.iter().any(|receipt| {
193 receipt.sequence == operation.sequence
194 && receipt.outcome == BackupExecutionOperationReceiptOutcome::Failed
195 });
196 journal.mark_operation_pending_at(
197 operation.sequence,
198 Some(state_updated_at(config.updated_at.as_ref())),
199 )?;
200 layout.write_execution_journal(journal)?;
201 if !reconcile_previous_attempt {
202 return Ok(None);
203 }
204 let pending_operation = journal
205 .operations
206 .iter()
207 .find(|candidate| candidate.sequence == operation.sequence)
208 .cloned()
209 .ok_or(
210 crate::execution::BackupExecutionJournalError::OperationNotFound(operation.sequence),
211 )?;
212 reconcile_pending_lifecycle(executor, journal, &pending_operation, completed_status)
213}
214
215fn prepare_snapshot_create_attempt(
216 config: &BackupRunnerConfig,
217 executor: &mut impl BackupRunnerExecutor,
218 layout: &BackupLayout,
219 plan: &BackupPlan,
220 journal: &mut BackupExecutionJournal,
221 operation: &crate::execution::BackupExecutionJournalOperation,
222) -> Result<Option<BackupExecutionOperationReceipt>, BackupRunnerError> {
223 let recovering_previous_attempt = operation.snapshot_ids_before.is_some();
224 let snapshot_ids_before = if recovering_previous_attempt {
225 operation.snapshot_ids_before.clone().ok_or(
226 crate::execution::BackupExecutionJournalError::MissingField(
227 "operations[].snapshot_ids_before",
228 ),
229 )?
230 } else {
231 let target = operation_target(operation)?;
232 let snapshots = observe_snapshot_inventory(executor, operation, &target)?;
233 let mut snapshot_ids = snapshots
234 .into_iter()
235 .map(|snapshot| snapshot.snapshot_id)
236 .collect::<Vec<_>>();
237 snapshot_ids.sort();
238 snapshot_ids
239 };
240 journal.mark_snapshot_create_pending_at(
241 operation.sequence,
242 Some(state_updated_at(config.updated_at.as_ref())),
243 snapshot_ids_before,
244 )?;
245 layout.write_execution_journal(journal)?;
246 if !recovering_previous_attempt {
247 return Ok(None);
248 }
249 let pending_operation = journal
250 .operations
251 .iter()
252 .find(|candidate| candidate.sequence == operation.sequence)
253 .cloned()
254 .ok_or(
255 crate::execution::BackupExecutionJournalError::OperationNotFound(operation.sequence),
256 )?;
257 reconcile_pending_snapshot_create(executor, layout, plan, journal, &pending_operation)
258}
259
260fn reconcile_pending_snapshot_create(
261 executor: &mut impl BackupRunnerExecutor,
262 layout: &BackupLayout,
263 plan: &BackupPlan,
264 journal: &BackupExecutionJournal,
265 operation: &crate::execution::BackupExecutionJournalOperation,
266) -> Result<Option<BackupExecutionOperationReceipt>, BackupRunnerError> {
267 let target = operation_target(operation)?;
268 if let Some(receipt) = recorded_snapshot_receipt(layout, plan, journal, operation, &target)? {
269 return Ok(Some(receipt));
270 }
271 let baseline = operation.snapshot_ids_before.as_ref().ok_or(
272 crate::execution::BackupExecutionJournalError::MissingField(
273 "operations[].snapshot_ids_before",
274 ),
275 )?;
276 let baseline = baseline.iter().map(String::as_str).collect::<BTreeSet<_>>();
277 let observed = observe_snapshot_inventory(executor, operation, &target)?
278 .into_iter()
279 .map(|snapshot| (snapshot.snapshot_id.clone(), snapshot))
280 .collect::<BTreeMap<_, _>>();
281 let missing = baseline
282 .iter()
283 .filter(|snapshot_id| !observed.contains_key(**snapshot_id))
284 .map(|snapshot_id| (*snapshot_id).to_string())
285 .collect::<Vec<_>>();
286 if !missing.is_empty() {
287 return Err(BackupRunnerError::SnapshotInventoryLostBaseline {
288 sequence: operation.sequence,
289 operation_id: operation.operation_id.clone(),
290 snapshot_ids: missing,
291 });
292 }
293 let mut created = observed
294 .into_iter()
295 .filter(|(snapshot_id, _)| !baseline.contains(snapshot_id.as_str()))
296 .collect::<Vec<_>>();
297 match created.len() {
298 0 => Ok(None),
299 1 => {
300 let snapshot = created.pop().expect("one snapshot candidate").1;
301 persist_created_snapshot(layout, plan, journal, operation, &target, snapshot).map(Some)
302 }
303 _ => Err(BackupRunnerError::SnapshotIdentityAmbiguous {
304 sequence: operation.sequence,
305 operation_id: operation.operation_id.clone(),
306 snapshot_ids: created
307 .into_iter()
308 .map(|(snapshot_id, _)| snapshot_id)
309 .collect(),
310 }),
311 }
312}
313
314fn observe_snapshot_inventory(
315 executor: &mut impl BackupRunnerExecutor,
316 operation: &crate::execution::BackupExecutionJournalOperation,
317 target: &str,
318) -> Result<Vec<BackupRunnerSnapshot>, BackupRunnerError> {
319 let snapshots = executor.snapshot_inventory(target).map_err(|error| {
320 BackupRunnerError::SnapshotInventoryFailed {
321 sequence: operation.sequence,
322 status: error.status,
323 message: error.message,
324 }
325 })?;
326 let mut identities = BTreeSet::new();
327 for snapshot in &snapshots {
328 if snapshot.snapshot_id.trim().is_empty() {
329 return Err(BackupRunnerError::InvalidSnapshotIdentity {
330 sequence: operation.sequence,
331 operation_id: operation.operation_id.clone(),
332 });
333 }
334 if !identities.insert(snapshot.snapshot_id.as_str()) {
335 return Err(BackupRunnerError::DuplicateSnapshotIdentity {
336 sequence: operation.sequence,
337 operation_id: operation.operation_id.clone(),
338 snapshot_id: snapshot.snapshot_id.clone(),
339 });
340 }
341 }
342 Ok(snapshots)
343}
344
345fn reconcile_pending_lifecycle(
346 executor: &mut impl BackupRunnerExecutor,
347 journal: &BackupExecutionJournal,
348 operation: &crate::execution::BackupExecutionJournalOperation,
349 completed_status: BackupRunnerCanisterStatus,
350) -> Result<Option<BackupExecutionOperationReceipt>, BackupRunnerError> {
351 let target = operation.target_canister_id.as_deref().ok_or(
352 BackupRunnerError::MissingOperationTarget {
353 sequence: operation.sequence,
354 },
355 )?;
356 let status = executor.canister_status(target).map_err(|error| {
357 BackupRunnerError::CanisterStatusFailed {
358 sequence: operation.sequence,
359 status: error.status,
360 message: error.message,
361 }
362 })?;
363 if status == completed_status {
364 return Ok(Some(BackupExecutionOperationReceipt::completed(
365 journal,
366 operation,
367 Some(current_timestamp_marker()),
368 )));
369 }
370 if status == BackupRunnerCanisterStatus::Stopping {
371 return Err(BackupRunnerError::CanisterStatusUnsettled {
372 sequence: operation.sequence,
373 operation_id: operation.operation_id.clone(),
374 status: status.label(),
375 });
376 }
377 Ok(None)
378}
379
380const fn lifecycle_completed_status(
381 kind: &BackupOperationKind,
382) -> Option<BackupRunnerCanisterStatus> {
383 match kind {
384 BackupOperationKind::Stop => Some(BackupRunnerCanisterStatus::Stopped),
385 BackupOperationKind::Start => Some(BackupRunnerCanisterStatus::Running),
386 _ => None,
387 }
388}
389
390fn finish_reconciled_command_lock(
391 operation: &crate::execution::BackupExecutionJournalOperation,
392 command_lock: Option<CommandLifetimeLock>,
393) -> Result<(), BackupRunnerError> {
394 command_lock
395 .ok_or_else(|| BackupRunnerError::MissingCommandLifetime {
396 sequence: operation.sequence,
397 operation_id: operation.operation_id.clone(),
398 })?
399 .finish()
400 .map_err(|error| backup_command_lock_error(operation, error))
401}
402
403fn reject_unknown_backup_command_outcome(
404 operation: &crate::execution::BackupExecutionJournalOperation,
405 command_lock: Option<CommandLifetimeLock>,
406) -> Result<(), BackupRunnerError> {
407 let Some(command_lock) = command_lock else {
408 return Ok(());
409 };
410 let lock_path = command_lock.path().to_string_lossy().to_string();
411 command_lock
412 .finish()
413 .map_err(|error| backup_command_lock_error(operation, error))?;
414 Err(BackupRunnerError::CommandOutcomeUnknown {
415 sequence: operation.sequence,
416 operation_id: operation.operation_id.clone(),
417 lock_path,
418 })
419}
420
421fn backup_command_lock(
422 layout: &BackupLayout,
423 operation: &crate::execution::BackupExecutionJournalOperation,
424) -> Result<Option<CommandLifetimeLock>, BackupRunnerError> {
425 if !matches!(
426 operation.kind,
427 crate::plan::BackupOperationKind::Stop
428 | crate::plan::BackupOperationKind::CreateSnapshot
429 | crate::plan::BackupOperationKind::Start
430 | crate::plan::BackupOperationKind::DownloadSnapshot
431 ) {
432 return Ok(None);
433 }
434
435 CommandLifetimeLock::acquire(&layout.execution_journal_path(), operation.sequence)
436 .map(Some)
437 .map_err(|error| backup_command_lock_error(operation, error))
438}
439
440fn backup_command_lock_error(
441 operation: &crate::execution::BackupExecutionJournalOperation,
442 error: CommandLifetimeLockError,
443) -> BackupRunnerError {
444 match error {
445 CommandLifetimeLockError::InFlight { lock_path } => BackupRunnerError::CommandInFlight {
446 sequence: operation.sequence,
447 operation_id: operation.operation_id.clone(),
448 lock_path,
449 },
450 CommandLifetimeLockError::UnsafeEntry { lock_path, kind } => {
451 BackupRunnerError::CommandLockUnsafeEntry {
452 sequence: operation.sequence,
453 operation_id: operation.operation_id.clone(),
454 lock_path,
455 kind,
456 }
457 }
458 CommandLifetimeLockError::Io(error) => BackupRunnerError::Io(error),
459 }
460}
461
462fn run_response(
463 plan: &BackupPlan,
464 journal: &BackupExecutionJournal,
465 executed: Vec<BackupRunExecutedOperation>,
466 max_steps_reached: bool,
467) -> BackupRunResponse {
468 let execution = journal.resume_summary();
469 BackupRunResponse {
470 run_id: plan.run_id.clone(),
471 plan_id: plan.plan_id.clone(),
472 backup_id: plan.run_id.clone(),
473 complete: execution.completed_operations + execution.skipped_operations
474 == execution.total_operations,
475 max_steps_reached,
476 executed_operation_count: executed.len(),
477 executed_operations: executed,
478 execution,
479 }
480}
481
482#[cfg(test)]
483mod tests;