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
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 if operation.kind == BackupOperationKind::VerifyArtifact {
112 reconcile_pending_artifact_verification(layout, journal, &operation)?
113 } else {
114 reject_unknown_backup_command_outcome(&operation, command_lock.take())?;
115 None
116 }
117 } else {
118 if let Some(completed_status) = lifecycle_completed_status(&operation.kind) {
119 prepare_lifecycle_attempt(
120 config,
121 executor,
122 layout,
123 journal,
124 &operation,
125 completed_status,
126 )?
127 } else if operation.kind == BackupOperationKind::CreateSnapshot {
128 prepare_snapshot_create_attempt(
129 config, executor, layout, plan, journal, &operation,
130 )?
131 } else {
132 journal.mark_operation_pending_at(
133 operation.sequence,
134 Some(state_updated_at(config.updated_at.as_ref())),
135 )?;
136 layout.write_execution_journal(journal)?;
137 None
138 }
139 };
140 if let Some(receipt) = reconciled_receipt {
141 finish_reconciled_command_lock(&operation, command_lock.take())?;
142 journal.record_operation_receipt(receipt)?;
143 layout.write_execution_journal(journal)?;
144 executed.push(BackupRunExecutedOperation::completed(&operation));
145 continue;
146 }
147
148 let operation_result = execute_operation_receipt(
149 config,
150 executor,
151 layout,
152 plan,
153 journal,
154 &operation,
155 command_lock.as_ref().map(CommandLifetimeLock::handle),
156 );
157 if let Some(command_lock) = command_lock {
158 command_lock
159 .finish()
160 .map_err(|error| backup_command_lock_error(&operation, error))?;
161 }
162
163 match operation_result {
164 Ok(receipt) => {
165 journal.record_operation_receipt(receipt)?;
166 layout.write_execution_journal(journal)?;
167 executed.push(BackupRunExecutedOperation::completed(&operation));
168 }
169 Err(error) => {
170 let receipt = crate::execution::BackupExecutionOperationReceipt::failed(
171 journal,
172 &operation,
173 Some(state_updated_at(config.updated_at.as_ref())),
174 error.to_string(),
175 );
176 journal.record_operation_receipt(receipt)?;
177 layout.write_execution_journal(journal)?;
178 executed.push(BackupRunExecutedOperation::failed(&operation));
179 return Err(error);
180 }
181 }
182 }
183}
184
185fn prepare_lifecycle_attempt(
186 config: &BackupRunnerConfig,
187 executor: &mut impl BackupRunnerExecutor,
188 layout: &BackupLayout,
189 journal: &mut BackupExecutionJournal,
190 operation: &crate::execution::BackupExecutionJournalOperation,
191 completed_status: BackupRunnerCanisterStatus,
192) -> Result<Option<BackupExecutionOperationReceipt>, BackupRunnerError> {
193 let reconcile_previous_attempt = operation.state == BackupExecutionOperationState::Failed
194 || journal.operation_receipts.iter().any(|receipt| {
195 receipt.sequence == operation.sequence
196 && receipt.outcome == BackupExecutionOperationReceiptOutcome::Failed
197 });
198 journal.mark_operation_pending_at(
199 operation.sequence,
200 Some(state_updated_at(config.updated_at.as_ref())),
201 )?;
202 layout.write_execution_journal(journal)?;
203 if !reconcile_previous_attempt {
204 return Ok(None);
205 }
206 let pending_operation = journal
207 .operations
208 .iter()
209 .find(|candidate| candidate.sequence == operation.sequence)
210 .cloned()
211 .ok_or(
212 crate::execution::BackupExecutionJournalError::OperationNotFound(operation.sequence),
213 )?;
214 reconcile_pending_lifecycle(executor, journal, &pending_operation, completed_status)
215}
216
217fn prepare_snapshot_create_attempt(
218 config: &BackupRunnerConfig,
219 executor: &mut impl BackupRunnerExecutor,
220 layout: &BackupLayout,
221 plan: &BackupPlan,
222 journal: &mut BackupExecutionJournal,
223 operation: &crate::execution::BackupExecutionJournalOperation,
224) -> Result<Option<BackupExecutionOperationReceipt>, BackupRunnerError> {
225 let recovering_previous_attempt = operation.snapshot_ids_before.is_some();
226 let snapshot_ids_before = if recovering_previous_attempt {
227 operation.snapshot_ids_before.clone().ok_or(
228 crate::execution::BackupExecutionJournalError::MissingField(
229 "operations[].snapshot_ids_before",
230 ),
231 )?
232 } else {
233 let target = operation_target(operation)?;
234 let snapshots = observe_snapshot_inventory(executor, operation, &target)?;
235 let mut snapshot_ids = snapshots
236 .into_iter()
237 .map(|snapshot| snapshot.snapshot_id)
238 .collect::<Vec<_>>();
239 snapshot_ids.sort();
240 snapshot_ids
241 };
242 journal.mark_snapshot_create_pending_at(
243 operation.sequence,
244 Some(state_updated_at(config.updated_at.as_ref())),
245 snapshot_ids_before,
246 )?;
247 layout.write_execution_journal(journal)?;
248 if !recovering_previous_attempt {
249 return Ok(None);
250 }
251 let pending_operation = journal
252 .operations
253 .iter()
254 .find(|candidate| candidate.sequence == operation.sequence)
255 .cloned()
256 .ok_or(
257 crate::execution::BackupExecutionJournalError::OperationNotFound(operation.sequence),
258 )?;
259 reconcile_pending_snapshot_create(executor, layout, plan, journal, &pending_operation)
260}
261
262fn reconcile_pending_snapshot_create(
263 executor: &mut impl BackupRunnerExecutor,
264 layout: &BackupLayout,
265 plan: &BackupPlan,
266 journal: &BackupExecutionJournal,
267 operation: &crate::execution::BackupExecutionJournalOperation,
268) -> Result<Option<BackupExecutionOperationReceipt>, BackupRunnerError> {
269 let target = operation_target(operation)?;
270 if let Some(receipt) = recorded_snapshot_receipt(layout, plan, journal, operation, &target)? {
271 return Ok(Some(receipt));
272 }
273 let baseline = operation.snapshot_ids_before.as_ref().ok_or(
274 crate::execution::BackupExecutionJournalError::MissingField(
275 "operations[].snapshot_ids_before",
276 ),
277 )?;
278 let baseline = baseline.iter().map(String::as_str).collect::<BTreeSet<_>>();
279 let observed = observe_snapshot_inventory(executor, operation, &target)?
280 .into_iter()
281 .map(|snapshot| (snapshot.snapshot_id.clone(), snapshot))
282 .collect::<BTreeMap<_, _>>();
283 let missing = baseline
284 .iter()
285 .filter(|snapshot_id| !observed.contains_key(**snapshot_id))
286 .map(|snapshot_id| (*snapshot_id).to_string())
287 .collect::<Vec<_>>();
288 if !missing.is_empty() {
289 return Err(BackupRunnerError::SnapshotInventoryLostBaseline {
290 sequence: operation.sequence,
291 operation_id: operation.operation_id.clone(),
292 snapshot_ids: missing,
293 });
294 }
295 let mut created = observed
296 .into_iter()
297 .filter(|(snapshot_id, _)| !baseline.contains(snapshot_id.as_str()))
298 .collect::<Vec<_>>();
299 match created.len() {
300 0 => Ok(None),
301 1 => {
302 let snapshot = created.pop().expect("one snapshot candidate").1;
303 persist_created_snapshot(layout, plan, journal, operation, &target, snapshot).map(Some)
304 }
305 _ => Err(BackupRunnerError::SnapshotIdentityAmbiguous {
306 sequence: operation.sequence,
307 operation_id: operation.operation_id.clone(),
308 snapshot_ids: created
309 .into_iter()
310 .map(|(snapshot_id, _)| snapshot_id)
311 .collect(),
312 }),
313 }
314}
315
316fn observe_snapshot_inventory(
317 executor: &mut impl BackupRunnerExecutor,
318 operation: &crate::execution::BackupExecutionJournalOperation,
319 target: &str,
320) -> Result<Vec<BackupRunnerSnapshot>, BackupRunnerError> {
321 let snapshots = executor.snapshot_inventory(target).map_err(|error| {
322 BackupRunnerError::SnapshotInventoryFailed {
323 sequence: operation.sequence,
324 status: error.status,
325 message: error.message,
326 }
327 })?;
328 let mut identities = BTreeSet::new();
329 for snapshot in &snapshots {
330 if snapshot.snapshot_id.trim().is_empty() {
331 return Err(BackupRunnerError::InvalidSnapshotIdentity {
332 sequence: operation.sequence,
333 operation_id: operation.operation_id.clone(),
334 });
335 }
336 if !identities.insert(snapshot.snapshot_id.as_str()) {
337 return Err(BackupRunnerError::DuplicateSnapshotIdentity {
338 sequence: operation.sequence,
339 operation_id: operation.operation_id.clone(),
340 snapshot_id: snapshot.snapshot_id.clone(),
341 });
342 }
343 }
344 Ok(snapshots)
345}
346
347fn reconcile_pending_lifecycle(
348 executor: &mut impl BackupRunnerExecutor,
349 journal: &BackupExecutionJournal,
350 operation: &crate::execution::BackupExecutionJournalOperation,
351 completed_status: BackupRunnerCanisterStatus,
352) -> Result<Option<BackupExecutionOperationReceipt>, BackupRunnerError> {
353 let target = operation.target_canister_id.as_deref().ok_or(
354 BackupRunnerError::MissingOperationTarget {
355 sequence: operation.sequence,
356 },
357 )?;
358 let status = executor.canister_status(target).map_err(|error| {
359 BackupRunnerError::CanisterStatusFailed {
360 sequence: operation.sequence,
361 status: error.status,
362 message: error.message,
363 }
364 })?;
365 if status == completed_status {
366 return Ok(Some(BackupExecutionOperationReceipt::completed(
367 journal,
368 operation,
369 Some(current_timestamp_marker()),
370 )));
371 }
372 if status == BackupRunnerCanisterStatus::Stopping {
373 return Err(BackupRunnerError::CanisterStatusUnsettled {
374 sequence: operation.sequence,
375 operation_id: operation.operation_id.clone(),
376 status: status.label(),
377 });
378 }
379 Ok(None)
380}
381
382const fn lifecycle_completed_status(
383 kind: &BackupOperationKind,
384) -> Option<BackupRunnerCanisterStatus> {
385 match kind {
386 BackupOperationKind::Stop => Some(BackupRunnerCanisterStatus::Stopped),
387 BackupOperationKind::Start => Some(BackupRunnerCanisterStatus::Running),
388 _ => None,
389 }
390}
391
392fn finish_reconciled_command_lock(
393 operation: &crate::execution::BackupExecutionJournalOperation,
394 command_lock: Option<CommandLifetimeLock>,
395) -> Result<(), BackupRunnerError> {
396 let Some(command_lock) = command_lock else {
397 if backup_operation_uses_command_lock(&operation.kind) {
398 return Err(BackupRunnerError::MissingCommandLifetime {
399 sequence: operation.sequence,
400 operation_id: operation.operation_id.clone(),
401 });
402 }
403 return Ok(());
404 };
405 command_lock
406 .finish()
407 .map_err(|error| backup_command_lock_error(operation, error))
408}
409
410const fn backup_operation_uses_command_lock(kind: &BackupOperationKind) -> bool {
411 matches!(
412 kind,
413 BackupOperationKind::Stop
414 | BackupOperationKind::CreateSnapshot
415 | BackupOperationKind::Start
416 | BackupOperationKind::DownloadSnapshot
417 )
418}
419
420fn reject_unknown_backup_command_outcome(
421 operation: &crate::execution::BackupExecutionJournalOperation,
422 command_lock: Option<CommandLifetimeLock>,
423) -> Result<(), BackupRunnerError> {
424 let Some(command_lock) = command_lock else {
425 return Ok(());
426 };
427 let lock_path = command_lock.path().to_string_lossy().to_string();
428 command_lock
429 .finish()
430 .map_err(|error| backup_command_lock_error(operation, error))?;
431 Err(BackupRunnerError::CommandOutcomeUnknown {
432 sequence: operation.sequence,
433 operation_id: operation.operation_id.clone(),
434 lock_path,
435 })
436}
437
438fn backup_command_lock(
439 layout: &BackupLayout,
440 operation: &crate::execution::BackupExecutionJournalOperation,
441) -> Result<Option<CommandLifetimeLock>, BackupRunnerError> {
442 if !backup_operation_uses_command_lock(&operation.kind) {
443 return Ok(None);
444 }
445
446 CommandLifetimeLock::acquire(&layout.execution_journal_path(), operation.sequence)
447 .map(Some)
448 .map_err(|error| backup_command_lock_error(operation, error))
449}
450
451fn backup_command_lock_error(
452 operation: &crate::execution::BackupExecutionJournalOperation,
453 error: CommandLifetimeLockError,
454) -> BackupRunnerError {
455 match error {
456 CommandLifetimeLockError::InFlight { lock_path } => BackupRunnerError::CommandInFlight {
457 sequence: operation.sequence,
458 operation_id: operation.operation_id.clone(),
459 lock_path,
460 },
461 CommandLifetimeLockError::UnsafeEntry { lock_path, kind } => {
462 BackupRunnerError::CommandLockUnsafeEntry {
463 sequence: operation.sequence,
464 operation_id: operation.operation_id.clone(),
465 lock_path,
466 kind,
467 }
468 }
469 CommandLifetimeLockError::Io(error) => BackupRunnerError::Io(error),
470 }
471}
472
473fn run_response(
474 plan: &BackupPlan,
475 journal: &BackupExecutionJournal,
476 executed: Vec<BackupRunExecutedOperation>,
477 max_steps_reached: bool,
478) -> BackupRunResponse {
479 let execution = journal.resume_summary();
480 BackupRunResponse {
481 run_id: plan.run_id.clone(),
482 plan_id: plan.plan_id.clone(),
483 backup_id: plan.run_id.clone(),
484 complete: execution.completed_operations + execution.skipped_operations
485 == execution.total_operations,
486 max_steps_reached,
487 executed_operation_count: executed.len(),
488 executed_operations: executed,
489 execution,
490 }
491}
492
493#[cfg(test)]
494mod tests;