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 reconciled_receipt = if operation.state == BackupExecutionOperationState::Pending {
176 if let Some(completed_status) = lifecycle_completed_status(&operation.kind) {
177 reconcile_pending_lifecycle(executor, journal, &operation, completed_status)?
178 } else if operation.kind == BackupOperationKind::CreateSnapshot {
179 reconcile_pending_snapshot_create(executor, layout, plan, journal, &operation)?
180 } else if operation.kind == BackupOperationKind::DownloadSnapshot {
181 reconcile_pending_download(layout, journal, &operation)?
182 } else if operation.kind == BackupOperationKind::VerifyArtifact {
183 reconcile_pending_artifact_verification(layout, journal, &operation)?
184 } else {
185 reject_unknown_backup_command_outcome(&operation, command_lock.take())?;
186 None
187 }
188 } else {
189 if let Some(completed_status) = lifecycle_completed_status(&operation.kind) {
190 prepare_lifecycle_attempt(
191 config,
192 executor,
193 layout,
194 journal,
195 &operation,
196 completed_status,
197 )?
198 } else if operation.kind == BackupOperationKind::CreateSnapshot {
199 prepare_snapshot_create_attempt(
200 config, executor, layout, plan, journal, &operation,
201 )?
202 } else {
203 journal.mark_operation_pending_at(
204 operation.sequence,
205 Some(state_updated_at(config.updated_at.as_ref())),
206 )?;
207 layout.write_execution_journal(journal)?;
208 None
209 }
210 };
211 if let Some(receipt) = reconciled_receipt {
212 finish_reconciled_command_lock(&operation, command_lock.take())?;
213 journal.record_operation_receipt(receipt)?;
214 terminal_writer(layout, journal)?;
215 executed.push(BackupRunExecutedOperation::completed(&operation));
216 continue;
217 }
218
219 let operation_result = execute_operation_receipt(
220 config,
221 executor,
222 layout,
223 plan,
224 journal,
225 &operation,
226 command_lock.as_ref().map(CommandLifetimeLock::handle),
227 );
228 if let Some(command_lock) = command_lock {
229 command_lock
230 .finish()
231 .map_err(|error| backup_command_lock_error(&operation, error))?;
232 }
233
234 match operation_result {
235 Ok(receipt) => {
236 journal.record_operation_receipt(receipt)?;
237 terminal_writer(layout, journal)?;
238 executed.push(BackupRunExecutedOperation::completed(&operation));
239 }
240 Err(error) => {
241 let receipt = crate::execution::BackupExecutionOperationReceipt::failed(
242 journal,
243 &operation,
244 Some(state_updated_at(config.updated_at.as_ref())),
245 error.to_string(),
246 );
247 journal.record_operation_receipt(receipt)?;
248 terminal_writer(layout, journal)?;
249 executed.push(BackupRunExecutedOperation::failed(&operation));
250 return Err(error);
251 }
252 }
253 }
254}
255
256fn prepare_lifecycle_attempt(
257 config: &BackupRunnerConfig,
258 executor: &mut impl BackupRunnerExecutor,
259 layout: &BackupLayout,
260 journal: &mut BackupExecutionJournal,
261 operation: &crate::execution::BackupExecutionJournalOperation,
262 completed_status: BackupRunnerCanisterStatus,
263) -> Result<Option<BackupExecutionOperationReceipt>, BackupRunnerError> {
264 let reconcile_previous_attempt = operation.state == BackupExecutionOperationState::Failed
265 || journal.operation_receipts.iter().any(|receipt| {
266 receipt.sequence == operation.sequence
267 && receipt.outcome == BackupExecutionOperationReceiptOutcome::Failed
268 });
269 journal.mark_operation_pending_at(
270 operation.sequence,
271 Some(state_updated_at(config.updated_at.as_ref())),
272 )?;
273 layout.write_execution_journal(journal)?;
274 if !reconcile_previous_attempt {
275 return Ok(None);
276 }
277 let pending_operation = journal
278 .operations
279 .iter()
280 .find(|candidate| candidate.sequence == operation.sequence)
281 .cloned()
282 .ok_or(
283 crate::execution::BackupExecutionJournalError::OperationNotFound(operation.sequence),
284 )?;
285 reconcile_pending_lifecycle(executor, journal, &pending_operation, completed_status)
286}
287
288fn prepare_snapshot_create_attempt(
289 config: &BackupRunnerConfig,
290 executor: &mut impl BackupRunnerExecutor,
291 layout: &BackupLayout,
292 plan: &BackupPlan,
293 journal: &mut BackupExecutionJournal,
294 operation: &crate::execution::BackupExecutionJournalOperation,
295) -> Result<Option<BackupExecutionOperationReceipt>, BackupRunnerError> {
296 let recovering_previous_attempt = operation.snapshot_ids_before.is_some();
297 let snapshot_ids_before = if recovering_previous_attempt {
298 operation.snapshot_ids_before.clone().ok_or(
299 crate::execution::BackupExecutionJournalError::MissingField(
300 "operations[].snapshot_ids_before",
301 ),
302 )?
303 } else {
304 let target = operation_target(operation)?;
305 let snapshots = observe_snapshot_inventory(executor, operation, &target)?;
306 let mut snapshot_ids = snapshots
307 .into_iter()
308 .map(|snapshot| snapshot.snapshot_id)
309 .collect::<Vec<_>>();
310 snapshot_ids.sort();
311 snapshot_ids
312 };
313 journal.mark_snapshot_create_pending_at(
314 operation.sequence,
315 Some(state_updated_at(config.updated_at.as_ref())),
316 snapshot_ids_before,
317 )?;
318 layout.write_execution_journal(journal)?;
319 if !recovering_previous_attempt {
320 return Ok(None);
321 }
322 let pending_operation = journal
323 .operations
324 .iter()
325 .find(|candidate| candidate.sequence == operation.sequence)
326 .cloned()
327 .ok_or(
328 crate::execution::BackupExecutionJournalError::OperationNotFound(operation.sequence),
329 )?;
330 reconcile_pending_snapshot_create(executor, layout, plan, journal, &pending_operation)
331}
332
333fn reconcile_pending_snapshot_create(
334 executor: &mut impl BackupRunnerExecutor,
335 layout: &BackupLayout,
336 plan: &BackupPlan,
337 journal: &BackupExecutionJournal,
338 operation: &crate::execution::BackupExecutionJournalOperation,
339) -> Result<Option<BackupExecutionOperationReceipt>, BackupRunnerError> {
340 let target = operation_target(operation)?;
341 if let Some(receipt) = recorded_snapshot_receipt(layout, plan, journal, operation, &target)? {
342 return Ok(Some(receipt));
343 }
344 let baseline = operation.snapshot_ids_before.as_ref().ok_or(
345 crate::execution::BackupExecutionJournalError::MissingField(
346 "operations[].snapshot_ids_before",
347 ),
348 )?;
349 let baseline = baseline.iter().map(String::as_str).collect::<BTreeSet<_>>();
350 let observed = observe_snapshot_inventory(executor, operation, &target)?
351 .into_iter()
352 .map(|snapshot| (snapshot.snapshot_id.clone(), snapshot))
353 .collect::<BTreeMap<_, _>>();
354 let missing = baseline
355 .iter()
356 .filter(|snapshot_id| !observed.contains_key(**snapshot_id))
357 .map(|snapshot_id| (*snapshot_id).to_string())
358 .collect::<Vec<_>>();
359 if !missing.is_empty() {
360 return Err(BackupRunnerError::SnapshotInventoryLostBaseline {
361 sequence: operation.sequence,
362 operation_id: operation.operation_id.clone(),
363 snapshot_ids: missing,
364 });
365 }
366 let mut created = observed
367 .into_iter()
368 .filter(|(snapshot_id, _)| !baseline.contains(snapshot_id.as_str()))
369 .collect::<Vec<_>>();
370 match created.len() {
371 0 => Ok(None),
372 1 => {
373 let snapshot = created.pop().expect("one snapshot candidate").1;
374 persist_created_snapshot(layout, plan, journal, operation, &target, snapshot).map(Some)
375 }
376 _ => Err(BackupRunnerError::SnapshotIdentityAmbiguous {
377 sequence: operation.sequence,
378 operation_id: operation.operation_id.clone(),
379 snapshot_ids: created
380 .into_iter()
381 .map(|(snapshot_id, _)| snapshot_id)
382 .collect(),
383 }),
384 }
385}
386
387fn observe_snapshot_inventory(
388 executor: &mut impl BackupRunnerExecutor,
389 operation: &crate::execution::BackupExecutionJournalOperation,
390 target: &str,
391) -> Result<Vec<BackupRunnerSnapshot>, BackupRunnerError> {
392 let snapshots = executor.snapshot_inventory(target).map_err(|error| {
393 BackupRunnerError::SnapshotInventoryFailed {
394 sequence: operation.sequence,
395 status: error.status,
396 message: error.message,
397 }
398 })?;
399 let mut identities = BTreeSet::new();
400 for snapshot in &snapshots {
401 if snapshot.snapshot_id.trim().is_empty() {
402 return Err(BackupRunnerError::InvalidSnapshotIdentity {
403 sequence: operation.sequence,
404 operation_id: operation.operation_id.clone(),
405 });
406 }
407 if !identities.insert(snapshot.snapshot_id.as_str()) {
408 return Err(BackupRunnerError::DuplicateSnapshotIdentity {
409 sequence: operation.sequence,
410 operation_id: operation.operation_id.clone(),
411 snapshot_id: snapshot.snapshot_id.clone(),
412 });
413 }
414 }
415 Ok(snapshots)
416}
417
418fn reconcile_pending_lifecycle(
419 executor: &mut impl BackupRunnerExecutor,
420 journal: &BackupExecutionJournal,
421 operation: &crate::execution::BackupExecutionJournalOperation,
422 completed_status: BackupRunnerCanisterStatus,
423) -> Result<Option<BackupExecutionOperationReceipt>, BackupRunnerError> {
424 let target = operation.target_canister_id.as_deref().ok_or(
425 BackupRunnerError::MissingOperationTarget {
426 sequence: operation.sequence,
427 },
428 )?;
429 let status = executor.canister_status(target).map_err(|error| {
430 BackupRunnerError::CanisterStatusFailed {
431 sequence: operation.sequence,
432 status: error.status,
433 message: error.message,
434 }
435 })?;
436 if status == completed_status {
437 return Ok(Some(BackupExecutionOperationReceipt::completed(
438 journal,
439 operation,
440 Some(current_timestamp_marker()),
441 )));
442 }
443 if status == BackupRunnerCanisterStatus::Stopping {
444 return Err(BackupRunnerError::CanisterStatusUnsettled {
445 sequence: operation.sequence,
446 operation_id: operation.operation_id.clone(),
447 status: status.label(),
448 });
449 }
450 Ok(None)
451}
452
453const fn lifecycle_completed_status(
454 kind: &BackupOperationKind,
455) -> Option<BackupRunnerCanisterStatus> {
456 match kind {
457 BackupOperationKind::Stop => Some(BackupRunnerCanisterStatus::Stopped),
458 BackupOperationKind::Start => Some(BackupRunnerCanisterStatus::Running),
459 _ => None,
460 }
461}
462
463fn finish_reconciled_command_lock(
464 operation: &crate::execution::BackupExecutionJournalOperation,
465 command_lock: Option<CommandLifetimeLock>,
466) -> Result<(), BackupRunnerError> {
467 let Some(command_lock) = command_lock else {
468 if backup_operation_uses_command_lock(&operation.kind) {
469 return Err(BackupRunnerError::MissingCommandLifetime {
470 sequence: operation.sequence,
471 operation_id: operation.operation_id.clone(),
472 });
473 }
474 return Ok(());
475 };
476 command_lock
477 .finish()
478 .map_err(|error| backup_command_lock_error(operation, error))
479}
480
481const fn backup_operation_uses_command_lock(kind: &BackupOperationKind) -> bool {
482 matches!(
483 kind,
484 BackupOperationKind::Stop
485 | BackupOperationKind::CreateSnapshot
486 | BackupOperationKind::Start
487 | BackupOperationKind::DownloadSnapshot
488 )
489}
490
491fn reject_unknown_backup_command_outcome(
492 operation: &crate::execution::BackupExecutionJournalOperation,
493 command_lock: Option<CommandLifetimeLock>,
494) -> Result<(), BackupRunnerError> {
495 let Some(command_lock) = command_lock else {
496 return Ok(());
497 };
498 let lock_path = command_lock.path().to_string_lossy().to_string();
499 command_lock
500 .finish()
501 .map_err(|error| backup_command_lock_error(operation, error))?;
502 Err(BackupRunnerError::CommandOutcomeUnknown {
503 sequence: operation.sequence,
504 operation_id: operation.operation_id.clone(),
505 lock_path,
506 })
507}
508
509fn backup_command_lock(
510 layout: &BackupLayout,
511 operation: &crate::execution::BackupExecutionJournalOperation,
512) -> Result<Option<CommandLifetimeLock>, BackupRunnerError> {
513 if !backup_operation_uses_command_lock(&operation.kind) {
514 return Ok(None);
515 }
516
517 CommandLifetimeLock::acquire(&layout.execution_journal_path(), operation.sequence)
518 .map(Some)
519 .map_err(|error| backup_command_lock_error(operation, error))
520}
521
522fn backup_command_lock_error(
523 operation: &crate::execution::BackupExecutionJournalOperation,
524 error: CommandLifetimeLockError,
525) -> BackupRunnerError {
526 match error {
527 CommandLifetimeLockError::InFlight { lock_path } => BackupRunnerError::CommandInFlight {
528 sequence: operation.sequence,
529 operation_id: operation.operation_id.clone(),
530 lock_path,
531 },
532 CommandLifetimeLockError::UnsafeEntry { lock_path, kind } => {
533 BackupRunnerError::CommandLockUnsafeEntry {
534 sequence: operation.sequence,
535 operation_id: operation.operation_id.clone(),
536 lock_path,
537 kind,
538 }
539 }
540 CommandLifetimeLockError::Io(error) => BackupRunnerError::Io(error),
541 }
542}
543
544fn run_response(
545 plan: &BackupPlan,
546 journal: &BackupExecutionJournal,
547 executed: Vec<BackupRunExecutedOperation>,
548 max_steps_reached: bool,
549) -> BackupRunResponse {
550 let execution = journal.resume_summary();
551 BackupRunResponse {
552 run_id: plan.run_id.clone(),
553 plan_id: plan.plan_id.clone(),
554 backup_id: plan.run_id.clone(),
555 complete: execution.completed_operations + execution.skipped_operations
556 == execution.total_operations,
557 max_steps_reached,
558 executed_operation_count: executed.len(),
559 executed_operations: executed,
560 execution,
561 }
562}
563
564#[cfg(test)]
565mod tests;