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