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_artifact_verification, reconcile_pending_download, recorded_snapshot_receipt,
19};
20use std::collections::{BTreeMap, BTreeSet};
21
22const PREFLIGHT_TTL_SECONDS: u64 = 300;
23
24#[cfg(test)]
25pub(crate) fn build_manifest_for_test(
26 config: &BackupRunnerConfig,
27 plan: &BackupPlan,
28 journal: &crate::journal::DownloadJournal,
29) -> Result<crate::manifest::DeploymentBackupManifest, BackupRunnerError> {
30 manifest::build_manifest(config, plan, journal)
31}
32
33pub fn backup_run_execute_with_executor(
35 config: &BackupRunnerConfig,
36 executor: &mut impl BackupRunnerExecutor,
37) -> Result<BackupRunResponse, BackupRunnerError> {
38 backup_run_execute_with_terminal_writer(config, executor, &mut |layout, journal| {
39 layout.write_execution_journal(journal)
40 })
41}
42
43#[cfg(all(test, unix))]
44pub(crate) fn backup_run_execute_with_terminal_barriers(
45 config: &BackupRunnerConfig,
46 executor: &mut impl BackupRunnerExecutor,
47 mut barriers: impl FnMut(crate::persistence::DurableWriteBarrier),
48) -> Result<BackupRunResponse, BackupRunnerError> {
49 backup_run_execute_with_terminal_writer(config, executor, &mut |layout, journal| {
50 layout.write_execution_journal_at_barriers(journal, &mut barriers)
51 })
52}
53
54fn backup_run_execute_with_terminal_writer(
55 config: &BackupRunnerConfig,
56 executor: &mut impl BackupRunnerExecutor,
57 terminal_writer: &mut impl FnMut(
58 &BackupLayout,
59 &BackupExecutionJournal,
60 ) -> Result<(), crate::persistence::PersistenceError>,
61) -> Result<BackupRunResponse, BackupRunnerError> {
62 let layout = BackupLayout::new(config.out.clone());
63 let _lock = JournalLock::acquire(&layout.execution_journal_path())?;
64 let mut plan = layout.read_backup_plan()?;
65 let mut journal = if layout.execution_journal_path().is_file() {
66 layout.read_execution_journal()?
67 } else {
68 let journal = BackupExecutionJournal::from_plan(&plan)?;
69 layout.write_execution_journal(&journal)?;
70 journal
71 };
72 layout.verify_execution_integrity()?;
73 reject_premature_manifest(&layout, &journal)?;
74
75 accept_preflight_if_needed(config, executor, &layout, &mut plan, &mut journal)?;
76 execute_ready_operations(
77 config,
78 executor,
79 &layout,
80 &plan,
81 &mut journal,
82 terminal_writer,
83 )
84}
85
86fn reject_premature_manifest(
87 layout: &BackupLayout,
88 journal: &BackupExecutionJournal,
89) -> Result<(), BackupRunnerError> {
90 if !layout.manifest_path().exists() {
91 return Ok(());
92 }
93 let finalize = journal
94 .operations
95 .iter()
96 .find(|operation| operation.kind == BackupOperationKind::FinalizeManifest)
97 .ok_or(BackupRunnerError::NoReadyOperation)?;
98 if matches!(
99 finalize.state,
100 BackupExecutionOperationState::Pending
101 | BackupExecutionOperationState::Failed
102 | BackupExecutionOperationState::Completed
103 ) {
104 return Ok(());
105 }
106 Err(BackupRunnerError::PrematureManifest {
107 sequence: finalize.sequence,
108 state: finalize.state.clone(),
109 })
110}
111
112fn accept_preflight_if_needed(
113 config: &BackupRunnerConfig,
114 executor: &mut impl BackupRunnerExecutor,
115 layout: &BackupLayout,
116 plan: &mut BackupPlan,
117 journal: &mut BackupExecutionJournal,
118) -> Result<(), BackupRunnerError> {
119 if journal.preflight_accepted {
120 return Ok(());
121 }
122
123 let validated_at = state_updated_at(config.updated_at.as_ref());
124 let expires_at = timestamp_marker(timestamp_seconds(&validated_at) + PREFLIGHT_TTL_SECONDS);
125 let preflight_id = format!("preflight-{}", plan.run_id);
126 let receipts = executor
127 .preflight_receipts(plan, &preflight_id, &validated_at, &expires_at)
128 .map_err(|error| BackupRunnerError::PreflightFailed {
129 status: error.status,
130 message: error.message,
131 })?;
132 plan.apply_execution_preflight_receipts(&receipts, &validated_at)?;
133 layout.write_backup_plan(plan)?;
134 journal.accept_preflight_receipts_at(&receipts, Some(validated_at))?;
135 layout.write_execution_journal(journal)?;
136 Ok(())
137}
138
139fn execute_ready_operations(
140 config: &BackupRunnerConfig,
141 executor: &mut impl BackupRunnerExecutor,
142 layout: &BackupLayout,
143 plan: &BackupPlan,
144 journal: &mut BackupExecutionJournal,
145 terminal_writer: &mut impl FnMut(
146 &BackupLayout,
147 &BackupExecutionJournal,
148 ) -> Result<(), crate::persistence::PersistenceError>,
149) -> Result<BackupRunResponse, BackupRunnerError> {
150 let mut executed = Vec::new();
151
152 loop {
153 let summary = journal.resume_summary();
154 if summary.completed_operations + summary.skipped_operations == summary.total_operations {
155 return Ok(run_response(plan, journal, executed, false));
156 }
157 if config
158 .max_steps
159 .is_some_and(|max_steps| executed.len() >= max_steps)
160 {
161 return Ok(run_response(plan, journal, executed, true));
162 }
163
164 let operation = journal
165 .next_ready_operation()
166 .cloned()
167 .ok_or(BackupRunnerError::NoReadyOperation)?;
168 if operation.state == BackupExecutionOperationState::Blocked {
169 return Err(BackupRunnerError::Blocked {
170 reasons: operation.blocking_reasons,
171 });
172 }
173
174 let mut command_lock = backup_command_lock(layout, &operation)?;
175 let preparation = prepare_backup_operation(
176 config,
177 executor,
178 layout,
179 plan,
180 journal,
181 &operation,
182 &mut command_lock,
183 );
184 let reconciled_receipt = match preparation {
185 Ok(receipt) => receipt,
186 Err(error) => {
187 let error = finish_preparation_command_lock(&operation, &mut command_lock, error)?;
188 return Err(failure_after_containment(
189 config,
190 executor,
191 layout,
192 plan,
193 journal,
194 &operation,
195 terminal_writer,
196 error,
197 ));
198 }
199 };
200 if let Some(receipt) = reconciled_receipt {
201 finish_backup_command_lock(&operation, command_lock.take())?;
202 journal.record_operation_receipt(receipt)?;
203 terminal_writer(layout, journal)?;
204 executed.push(BackupRunExecutedOperation::completed(&operation));
205 continue;
206 }
207
208 let operation_result = execute_operation_receipt(
209 config,
210 executor,
211 layout,
212 plan,
213 journal,
214 &operation,
215 command_lock.as_ref().map(CommandLifetimeLock::handle),
216 );
217 finish_backup_command_lock(&operation, command_lock)?;
218
219 match operation_result {
220 Ok(receipt) => {
221 journal.record_operation_receipt(receipt)?;
222 terminal_writer(layout, journal)?;
223 executed.push(BackupRunExecutedOperation::completed(&operation));
224 }
225 Err(error) => {
226 let receipt = crate::execution::BackupExecutionOperationReceipt::failed(
227 journal,
228 &operation,
229 Some(state_updated_at(config.updated_at.as_ref())),
230 error.to_string(),
231 );
232 journal.record_operation_receipt(receipt)?;
233 terminal_writer(layout, journal)?;
234 executed.push(BackupRunExecutedOperation::failed(&operation));
235 return Err(failure_after_containment(
236 config,
237 executor,
238 layout,
239 plan,
240 journal,
241 &operation,
242 terminal_writer,
243 error,
244 ));
245 }
246 }
247 }
248}
249
250fn finish_preparation_command_lock(
251 operation: &crate::execution::BackupExecutionJournalOperation,
252 command_lock: &mut Option<CommandLifetimeLock>,
253 primary: BackupRunnerError,
254) -> Result<BackupRunnerError, BackupRunnerError> {
255 let Some(command_lock) = command_lock.take() else {
256 return Ok(primary);
257 };
258 match command_lock
259 .finish()
260 .map_err(|error| backup_command_lock_error(operation, error))
261 {
262 Ok(()) => Ok(primary),
263 Err(containment) => Err(BackupRunnerError::FailureContainmentFailed {
264 primary: Box::new(primary),
265 containment: Box::new(containment),
266 }),
267 }
268}
269
270fn prepare_backup_operation(
271 config: &BackupRunnerConfig,
272 executor: &mut impl BackupRunnerExecutor,
273 layout: &BackupLayout,
274 plan: &BackupPlan,
275 journal: &mut BackupExecutionJournal,
276 operation: &crate::execution::BackupExecutionJournalOperation,
277 command_lock: &mut Option<CommandLifetimeLock>,
278) -> Result<Option<BackupExecutionOperationReceipt>, BackupRunnerError> {
279 if operation.state == BackupExecutionOperationState::Pending {
280 if let Some(completed_status) = lifecycle_completed_status(&operation.kind) {
281 reconcile_pending_lifecycle(executor, journal, operation, completed_status)
282 } else if operation.kind == BackupOperationKind::CreateSnapshot {
283 reconcile_pending_snapshot_create(executor, layout, plan, journal, operation)
284 } else if operation.kind == BackupOperationKind::DownloadSnapshot {
285 reconcile_pending_download(layout, journal, operation)
286 } else if operation.kind == BackupOperationKind::VerifyArtifact {
287 reconcile_pending_artifact_verification(layout, journal, operation)
288 } else {
289 reject_unknown_backup_command_outcome(operation, command_lock.take())?;
290 Ok(None)
291 }
292 } else if let Some(completed_status) = lifecycle_completed_status(&operation.kind) {
293 prepare_lifecycle_attempt(
294 config,
295 executor,
296 layout,
297 journal,
298 operation,
299 completed_status,
300 )
301 } else if operation.kind == BackupOperationKind::CreateSnapshot {
302 prepare_snapshot_create_attempt(config, executor, layout, plan, journal, operation)
303 } else {
304 journal.mark_operation_pending_at(
305 operation.sequence,
306 Some(state_updated_at(config.updated_at.as_ref())),
307 )?;
308 layout.write_execution_journal(journal)?;
309 Ok(None)
310 }
311}
312
313#[expect(
314 clippy::too_many_arguments,
315 reason = "failure containment needs the existing runner authorities and the primary typed cause"
316)]
317fn failure_after_containment(
318 config: &BackupRunnerConfig,
319 executor: &mut impl BackupRunnerExecutor,
320 layout: &BackupLayout,
321 plan: &BackupPlan,
322 journal: &mut BackupExecutionJournal,
323 primary_operation: &crate::execution::BackupExecutionJournalOperation,
324 terminal_writer: &mut impl FnMut(
325 &BackupLayout,
326 &BackupExecutionJournal,
327 ) -> Result<(), crate::persistence::PersistenceError>,
328 primary: BackupRunnerError,
329) -> BackupRunnerError {
330 if !journal.restart_required && primary_operation.kind != BackupOperationKind::Stop {
331 return primary;
332 }
333 if primary_operation.kind == BackupOperationKind::Stop
334 && matches!(
335 &primary,
336 BackupRunnerError::CanisterStatusFailed { .. }
337 | BackupRunnerError::CanisterStatusUnsettled { .. }
338 )
339 {
340 return primary;
341 }
342 match contain_backup_failure(
343 config,
344 executor,
345 layout,
346 plan,
347 journal,
348 primary_operation,
349 terminal_writer,
350 ) {
351 Ok(()) => primary,
352 Err(containment) => BackupRunnerError::FailureContainmentFailed {
353 primary: Box::new(primary),
354 containment: Box::new(containment),
355 },
356 }
357}
358
359fn contain_backup_failure(
360 config: &BackupRunnerConfig,
361 executor: &mut impl BackupRunnerExecutor,
362 layout: &BackupLayout,
363 plan: &BackupPlan,
364 journal: &mut BackupExecutionJournal,
365 primary_operation: &crate::execution::BackupExecutionJournalOperation,
366 terminal_writer: &mut impl FnMut(
367 &BackupLayout,
368 &BackupExecutionJournal,
369 ) -> Result<(), crate::persistence::PersistenceError>,
370) -> Result<(), BackupRunnerError> {
371 reconcile_failed_stop_before_containment(
372 config,
373 executor,
374 layout,
375 journal,
376 primary_operation,
377 terminal_writer,
378 )?;
379 let rearm_snapshot_phase = matches!(
380 primary_operation.kind,
381 BackupOperationKind::Stop | BackupOperationKind::CreateSnapshot
382 );
383
384 while let Some(operation) = journal.next_failure_containment_start().cloned() {
385 let mut command_lock = backup_command_lock(layout, &operation)?;
386 let reconcile_previous_attempt = operation.state == BackupExecutionOperationState::Pending
387 || operation.state == BackupExecutionOperationState::Failed
388 || journal.operation_receipts.iter().any(|receipt| {
389 receipt.sequence == operation.sequence
390 && receipt.outcome == BackupExecutionOperationReceiptOutcome::Failed
391 });
392 if operation.state != BackupExecutionOperationState::Pending {
393 journal.mark_failure_containment_start_pending_at(
394 operation.sequence,
395 Some(state_updated_at(config.updated_at.as_ref())),
396 )?;
397 layout.write_execution_journal(journal)?;
398 }
399 let pending_operation = journal
400 .operations
401 .iter()
402 .find(|candidate| candidate.sequence == operation.sequence)
403 .cloned()
404 .ok_or(
405 crate::execution::BackupExecutionJournalError::OperationNotFound(
406 operation.sequence,
407 ),
408 )?;
409 let reconciled_receipt = if reconcile_previous_attempt {
410 reconcile_pending_lifecycle(
411 executor,
412 journal,
413 &pending_operation,
414 BackupRunnerCanisterStatus::Running,
415 )?
416 } else {
417 None
418 };
419 let result = if let Some(receipt) = reconciled_receipt {
420 finish_backup_command_lock(&pending_operation, command_lock.take())?;
421 Ok(receipt)
422 } else {
423 let result = execute_operation_receipt(
424 config,
425 executor,
426 layout,
427 plan,
428 journal,
429 &pending_operation,
430 command_lock.as_ref().map(CommandLifetimeLock::handle),
431 );
432 if let Some(command_lock) = command_lock {
433 command_lock
434 .finish()
435 .map_err(|error| backup_command_lock_error(&pending_operation, error))?;
436 }
437 result
438 };
439
440 match result {
441 Ok(receipt) => {
442 journal.record_operation_receipt(receipt)?;
443 if rearm_snapshot_phase {
444 journal.rearm_after_failure_containment(
445 operation.sequence,
446 Some(state_updated_at(config.updated_at.as_ref())),
447 )?;
448 }
449 terminal_writer(layout, journal)?;
450 }
451 Err(error) => {
452 let receipt = BackupExecutionOperationReceipt::failed(
453 journal,
454 &pending_operation,
455 Some(state_updated_at(config.updated_at.as_ref())),
456 error.to_string(),
457 );
458 journal.record_operation_receipt(receipt)?;
459 terminal_writer(layout, journal)?;
460 return Err(error);
461 }
462 }
463 }
464
465 Ok(())
466}
467
468fn reconcile_failed_stop_before_containment(
469 config: &BackupRunnerConfig,
470 executor: &mut impl BackupRunnerExecutor,
471 layout: &BackupLayout,
472 journal: &mut BackupExecutionJournal,
473 primary_operation: &crate::execution::BackupExecutionJournalOperation,
474 terminal_writer: &mut impl FnMut(
475 &BackupLayout,
476 &BackupExecutionJournal,
477 ) -> Result<(), crate::persistence::PersistenceError>,
478) -> Result<(), BackupRunnerError> {
479 if primary_operation.kind != BackupOperationKind::Stop {
480 return Ok(());
481 }
482 let target = operation_target(primary_operation)?;
483 let status = executor.canister_status(&target).map_err(|error| {
484 BackupRunnerError::CanisterStatusFailed {
485 sequence: primary_operation.sequence,
486 status: error.status,
487 message: error.message,
488 }
489 })?;
490 match status {
491 BackupRunnerCanisterStatus::Running => Ok(()),
492 BackupRunnerCanisterStatus::Stopping => Err(BackupRunnerError::CanisterStatusUnsettled {
493 sequence: primary_operation.sequence,
494 operation_id: primary_operation.operation_id.clone(),
495 status: status.label(),
496 }),
497 BackupRunnerCanisterStatus::Stopped => {
498 journal.mark_operation_pending_at(
499 primary_operation.sequence,
500 Some(state_updated_at(config.updated_at.as_ref())),
501 )?;
502 layout.write_execution_journal(journal)?;
503 let pending_operation = journal
504 .operations
505 .iter()
506 .find(|operation| operation.sequence == primary_operation.sequence)
507 .cloned()
508 .ok_or(
509 crate::execution::BackupExecutionJournalError::OperationNotFound(
510 primary_operation.sequence,
511 ),
512 )?;
513 journal.record_operation_receipt(BackupExecutionOperationReceipt::completed(
514 journal,
515 &pending_operation,
516 Some(current_timestamp_marker()),
517 ))?;
518 terminal_writer(layout, journal)?;
519 Ok(())
520 }
521 }
522}
523
524fn prepare_lifecycle_attempt(
525 config: &BackupRunnerConfig,
526 executor: &mut impl BackupRunnerExecutor,
527 layout: &BackupLayout,
528 journal: &mut BackupExecutionJournal,
529 operation: &crate::execution::BackupExecutionJournalOperation,
530 completed_status: BackupRunnerCanisterStatus,
531) -> Result<Option<BackupExecutionOperationReceipt>, BackupRunnerError> {
532 let reconcile_previous_attempt = operation.state == BackupExecutionOperationState::Failed
533 || journal.operation_receipts.iter().any(|receipt| {
534 receipt.sequence == operation.sequence
535 && receipt.outcome == BackupExecutionOperationReceiptOutcome::Failed
536 });
537 journal.mark_operation_pending_at(
538 operation.sequence,
539 Some(state_updated_at(config.updated_at.as_ref())),
540 )?;
541 layout.write_execution_journal(journal)?;
542 if !reconcile_previous_attempt {
543 return Ok(None);
544 }
545 let pending_operation = journal
546 .operations
547 .iter()
548 .find(|candidate| candidate.sequence == operation.sequence)
549 .cloned()
550 .ok_or(
551 crate::execution::BackupExecutionJournalError::OperationNotFound(operation.sequence),
552 )?;
553 reconcile_pending_lifecycle(executor, journal, &pending_operation, completed_status)
554}
555
556fn prepare_snapshot_create_attempt(
557 config: &BackupRunnerConfig,
558 executor: &mut impl BackupRunnerExecutor,
559 layout: &BackupLayout,
560 plan: &BackupPlan,
561 journal: &mut BackupExecutionJournal,
562 operation: &crate::execution::BackupExecutionJournalOperation,
563) -> Result<Option<BackupExecutionOperationReceipt>, BackupRunnerError> {
564 let recovering_previous_attempt = operation.snapshot_ids_before.is_some();
565 let snapshot_ids_before = if recovering_previous_attempt {
566 operation.snapshot_ids_before.clone().ok_or(
567 crate::execution::BackupExecutionJournalError::MissingField(
568 "operations[].snapshot_ids_before",
569 ),
570 )?
571 } else {
572 let target = operation_target(operation)?;
573 let snapshots = observe_snapshot_inventory(executor, operation, &target)?;
574 let mut snapshot_ids = snapshots
575 .into_iter()
576 .map(|snapshot| snapshot.snapshot_id)
577 .collect::<Vec<_>>();
578 snapshot_ids.sort();
579 snapshot_ids
580 };
581 journal.mark_snapshot_create_pending_at(
582 operation.sequence,
583 Some(state_updated_at(config.updated_at.as_ref())),
584 snapshot_ids_before,
585 )?;
586 layout.write_execution_journal(journal)?;
587 if !recovering_previous_attempt {
588 return Ok(None);
589 }
590 let pending_operation = journal
591 .operations
592 .iter()
593 .find(|candidate| candidate.sequence == operation.sequence)
594 .cloned()
595 .ok_or(
596 crate::execution::BackupExecutionJournalError::OperationNotFound(operation.sequence),
597 )?;
598 reconcile_pending_snapshot_create(executor, layout, plan, journal, &pending_operation)
599}
600
601fn reconcile_pending_snapshot_create(
602 executor: &mut impl BackupRunnerExecutor,
603 layout: &BackupLayout,
604 plan: &BackupPlan,
605 journal: &BackupExecutionJournal,
606 operation: &crate::execution::BackupExecutionJournalOperation,
607) -> Result<Option<BackupExecutionOperationReceipt>, BackupRunnerError> {
608 let target = operation_target(operation)?;
609 if let Some(receipt) = recorded_snapshot_receipt(layout, plan, journal, operation, &target)? {
610 return Ok(Some(receipt));
611 }
612 let baseline = operation.snapshot_ids_before.as_ref().ok_or(
613 crate::execution::BackupExecutionJournalError::MissingField(
614 "operations[].snapshot_ids_before",
615 ),
616 )?;
617 let baseline = baseline.iter().map(String::as_str).collect::<BTreeSet<_>>();
618 let observed = observe_snapshot_inventory(executor, operation, &target)?
619 .into_iter()
620 .map(|snapshot| (snapshot.snapshot_id.clone(), snapshot))
621 .collect::<BTreeMap<_, _>>();
622 let missing = baseline
623 .iter()
624 .filter(|snapshot_id| !observed.contains_key(**snapshot_id))
625 .map(|snapshot_id| (*snapshot_id).to_string())
626 .collect::<Vec<_>>();
627 if !missing.is_empty() {
628 return Err(BackupRunnerError::SnapshotInventoryLostBaseline {
629 sequence: operation.sequence,
630 operation_id: operation.operation_id.clone(),
631 snapshot_ids: missing,
632 });
633 }
634 let mut created = observed
635 .into_iter()
636 .filter(|(snapshot_id, _)| !baseline.contains(snapshot_id.as_str()))
637 .collect::<Vec<_>>();
638 match created.len() {
639 0 => Ok(None),
640 1 => {
641 let snapshot = created.pop().expect("one snapshot candidate").1;
642 persist_created_snapshot(layout, plan, journal, operation, &target, snapshot).map(Some)
643 }
644 _ => Err(BackupRunnerError::SnapshotIdentityAmbiguous {
645 sequence: operation.sequence,
646 operation_id: operation.operation_id.clone(),
647 snapshot_ids: created
648 .into_iter()
649 .map(|(snapshot_id, _)| snapshot_id)
650 .collect(),
651 }),
652 }
653}
654
655fn observe_snapshot_inventory(
656 executor: &mut impl BackupRunnerExecutor,
657 operation: &crate::execution::BackupExecutionJournalOperation,
658 target: &str,
659) -> Result<Vec<BackupRunnerSnapshot>, BackupRunnerError> {
660 let snapshots = executor.snapshot_inventory(target).map_err(|error| {
661 BackupRunnerError::SnapshotInventoryFailed {
662 sequence: operation.sequence,
663 status: error.status,
664 message: error.message,
665 }
666 })?;
667 let mut identities = BTreeSet::new();
668 for snapshot in &snapshots {
669 if snapshot.snapshot_id.trim().is_empty() {
670 return Err(BackupRunnerError::InvalidSnapshotIdentity {
671 sequence: operation.sequence,
672 operation_id: operation.operation_id.clone(),
673 });
674 }
675 if !identities.insert(snapshot.snapshot_id.as_str()) {
676 return Err(BackupRunnerError::DuplicateSnapshotIdentity {
677 sequence: operation.sequence,
678 operation_id: operation.operation_id.clone(),
679 snapshot_id: snapshot.snapshot_id.clone(),
680 });
681 }
682 }
683 Ok(snapshots)
684}
685
686fn reconcile_pending_lifecycle(
687 executor: &mut impl BackupRunnerExecutor,
688 journal: &BackupExecutionJournal,
689 operation: &crate::execution::BackupExecutionJournalOperation,
690 completed_status: BackupRunnerCanisterStatus,
691) -> Result<Option<BackupExecutionOperationReceipt>, BackupRunnerError> {
692 let target = operation.target_canister_id.as_deref().ok_or(
693 BackupRunnerError::MissingOperationTarget {
694 sequence: operation.sequence,
695 },
696 )?;
697 let status = executor.canister_status(target).map_err(|error| {
698 BackupRunnerError::CanisterStatusFailed {
699 sequence: operation.sequence,
700 status: error.status,
701 message: error.message,
702 }
703 })?;
704 if status == completed_status {
705 return Ok(Some(BackupExecutionOperationReceipt::completed(
706 journal,
707 operation,
708 Some(current_timestamp_marker()),
709 )));
710 }
711 if status == BackupRunnerCanisterStatus::Stopping {
712 return Err(BackupRunnerError::CanisterStatusUnsettled {
713 sequence: operation.sequence,
714 operation_id: operation.operation_id.clone(),
715 status: status.label(),
716 });
717 }
718 Ok(None)
719}
720
721const fn lifecycle_completed_status(
722 kind: &BackupOperationKind,
723) -> Option<BackupRunnerCanisterStatus> {
724 match kind {
725 BackupOperationKind::Stop => Some(BackupRunnerCanisterStatus::Stopped),
726 BackupOperationKind::Start => Some(BackupRunnerCanisterStatus::Running),
727 _ => None,
728 }
729}
730
731fn finish_backup_command_lock(
732 operation: &crate::execution::BackupExecutionJournalOperation,
733 command_lock: Option<CommandLifetimeLock>,
734) -> Result<(), BackupRunnerError> {
735 let Some(command_lock) = command_lock else {
736 if backup_operation_uses_command_lock(&operation.kind) {
737 return Err(BackupRunnerError::MissingCommandLifetime {
738 sequence: operation.sequence,
739 operation_id: operation.operation_id.clone(),
740 });
741 }
742 return Ok(());
743 };
744 command_lock
745 .finish()
746 .map_err(|error| backup_command_lock_error(operation, error))
747}
748
749const fn backup_operation_uses_command_lock(kind: &BackupOperationKind) -> bool {
750 matches!(
751 kind,
752 BackupOperationKind::Stop
753 | BackupOperationKind::CreateSnapshot
754 | BackupOperationKind::Start
755 | BackupOperationKind::DownloadSnapshot
756 )
757}
758
759fn reject_unknown_backup_command_outcome(
760 operation: &crate::execution::BackupExecutionJournalOperation,
761 command_lock: Option<CommandLifetimeLock>,
762) -> Result<(), BackupRunnerError> {
763 let Some(command_lock) = command_lock else {
764 return Ok(());
765 };
766 let lock_path = command_lock.path().to_string_lossy().to_string();
767 command_lock
768 .finish()
769 .map_err(|error| backup_command_lock_error(operation, error))?;
770 Err(BackupRunnerError::CommandOutcomeUnknown {
771 sequence: operation.sequence,
772 operation_id: operation.operation_id.clone(),
773 lock_path,
774 })
775}
776
777fn backup_command_lock(
778 layout: &BackupLayout,
779 operation: &crate::execution::BackupExecutionJournalOperation,
780) -> Result<Option<CommandLifetimeLock>, BackupRunnerError> {
781 if !backup_operation_uses_command_lock(&operation.kind) {
782 return Ok(None);
783 }
784
785 CommandLifetimeLock::acquire(&layout.execution_journal_path(), operation.sequence)
786 .map(Some)
787 .map_err(|error| backup_command_lock_error(operation, error))
788}
789
790fn backup_command_lock_error(
791 operation: &crate::execution::BackupExecutionJournalOperation,
792 error: CommandLifetimeLockError,
793) -> BackupRunnerError {
794 match error {
795 CommandLifetimeLockError::InFlight { lock_path } => BackupRunnerError::CommandInFlight {
796 sequence: operation.sequence,
797 operation_id: operation.operation_id.clone(),
798 lock_path,
799 },
800 CommandLifetimeLockError::UnsafeEntry { lock_path, kind } => {
801 BackupRunnerError::CommandLockUnsafeEntry {
802 sequence: operation.sequence,
803 operation_id: operation.operation_id.clone(),
804 lock_path,
805 kind,
806 }
807 }
808 CommandLifetimeLockError::Io(error) => BackupRunnerError::Io(error),
809 }
810}
811
812fn run_response(
813 plan: &BackupPlan,
814 journal: &BackupExecutionJournal,
815 executed: Vec<BackupRunExecutedOperation>,
816 max_steps_reached: bool,
817) -> BackupRunResponse {
818 let execution = journal.resume_summary();
819 BackupRunResponse {
820 run_id: plan.run_id.clone(),
821 plan_id: plan.plan_id.clone(),
822 backup_id: plan.run_id.clone(),
823 complete: execution.completed_operations + execution.skipped_operations
824 == execution.total_operations,
825 max_steps_reached,
826 executed_operation_count: executed.len(),
827 executed_operations: executed,
828 execution,
829 }
830}
831
832#[cfg(test)]
833mod tests;