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