1mod manifest;
2mod operations;
3mod types;
4
5pub use types::*;
6
7use crate::{
8 execution::{
9 BackupExecutionJournal, BackupExecutionOperationReceipt, BackupExecutionOperationState,
10 },
11 persistence::{BackupLayout, CommandLifetimeLock, CommandLifetimeLockError, JournalLock},
12 plan::{BackupOperationKind, BackupPlan},
13 timestamp::{current_timestamp_marker, state_updated_at, timestamp_marker, timestamp_seconds},
14};
15use operations::{
16 execute_operation_receipt, operation_target, persist_created_snapshot,
17 recorded_snapshot_receipt,
18};
19use std::collections::{BTreeMap, BTreeSet};
20
21const PREFLIGHT_TTL_SECONDS: u64 = 300;
22
23pub fn backup_run_execute_with_executor(
25 config: &BackupRunnerConfig,
26 executor: &mut impl BackupRunnerExecutor,
27) -> Result<BackupRunResponse, BackupRunnerError> {
28 let layout = BackupLayout::new(config.out.clone());
29 let _lock = JournalLock::acquire(&layout.execution_journal_path())?;
30 let mut plan = layout.read_backup_plan()?;
31 let mut journal = if layout.execution_journal_path().is_file() {
32 layout.read_execution_journal()?
33 } else {
34 let journal = BackupExecutionJournal::from_plan(&plan)?;
35 layout.write_execution_journal(&journal)?;
36 journal
37 };
38 layout.verify_execution_integrity()?;
39
40 accept_preflight_if_needed(config, executor, &layout, &mut plan, &mut journal)?;
41 execute_ready_operations(config, executor, &layout, &plan, &mut journal)
42}
43
44fn accept_preflight_if_needed(
45 config: &BackupRunnerConfig,
46 executor: &mut impl BackupRunnerExecutor,
47 layout: &BackupLayout,
48 plan: &mut BackupPlan,
49 journal: &mut BackupExecutionJournal,
50) -> Result<(), BackupRunnerError> {
51 if journal.preflight_accepted {
52 return Ok(());
53 }
54
55 let validated_at = state_updated_at(config.updated_at.as_ref());
56 let expires_at = timestamp_marker(timestamp_seconds(&validated_at) + PREFLIGHT_TTL_SECONDS);
57 let preflight_id = format!("preflight-{}", plan.run_id);
58 let receipts = executor
59 .preflight_receipts(plan, &preflight_id, &validated_at, &expires_at)
60 .map_err(|error| BackupRunnerError::PreflightFailed {
61 status: error.status,
62 message: error.message,
63 })?;
64 plan.apply_execution_preflight_receipts(&receipts, &validated_at)?;
65 layout.write_backup_plan(plan)?;
66 journal.accept_preflight_receipts_at(&receipts, Some(validated_at))?;
67 layout.write_execution_journal(journal)?;
68 Ok(())
69}
70
71fn execute_ready_operations(
72 config: &BackupRunnerConfig,
73 executor: &mut impl BackupRunnerExecutor,
74 layout: &BackupLayout,
75 plan: &BackupPlan,
76 journal: &mut BackupExecutionJournal,
77) -> Result<BackupRunResponse, BackupRunnerError> {
78 let mut executed = Vec::new();
79
80 loop {
81 let summary = journal.resume_summary();
82 if summary.completed_operations + summary.skipped_operations == summary.total_operations {
83 return Ok(run_response(plan, journal, executed, false));
84 }
85 if config
86 .max_steps
87 .is_some_and(|max_steps| executed.len() >= max_steps)
88 {
89 return Ok(run_response(plan, journal, executed, true));
90 }
91
92 let operation = journal
93 .next_ready_operation()
94 .cloned()
95 .ok_or(BackupRunnerError::NoReadyOperation)?;
96 if operation.state == BackupExecutionOperationState::Blocked {
97 return Err(BackupRunnerError::Blocked {
98 reasons: operation.blocking_reasons,
99 });
100 }
101
102 let mut command_lock = backup_command_lock(layout, &operation)?;
103 if operation.state == BackupExecutionOperationState::Pending {
104 if operation.kind == BackupOperationKind::Stop {
105 if let Some(receipt) = reconcile_pending_stop(executor, journal, &operation)? {
106 finish_reconciled_command_lock(&operation, command_lock.take())?;
107 journal.record_operation_receipt(receipt)?;
108 layout.write_execution_journal(journal)?;
109 executed.push(BackupRunExecutedOperation::completed(&operation));
110 continue;
111 }
112 } else if operation.kind == BackupOperationKind::CreateSnapshot {
113 if let Some(receipt) =
114 reconcile_pending_snapshot_create(executor, layout, plan, journal, &operation)?
115 {
116 finish_reconciled_command_lock(&operation, command_lock.take())?;
117 journal.record_operation_receipt(receipt)?;
118 layout.write_execution_journal(journal)?;
119 executed.push(BackupRunExecutedOperation::completed(&operation));
120 continue;
121 }
122 } else {
123 reject_unknown_backup_command_outcome(&operation, command_lock.take())?;
124 }
125 } else {
126 if operation.kind == BackupOperationKind::CreateSnapshot {
127 if let Some(receipt) = prepare_snapshot_create_attempt(
128 config, executor, layout, plan, journal, &operation,
129 )? {
130 finish_reconciled_command_lock(&operation, command_lock.take())?;
131 journal.record_operation_receipt(receipt)?;
132 layout.write_execution_journal(journal)?;
133 executed.push(BackupRunExecutedOperation::completed(&operation));
134 continue;
135 }
136 } else {
137 journal.mark_operation_pending_at(
138 operation.sequence,
139 Some(state_updated_at(config.updated_at.as_ref())),
140 )?;
141 layout.write_execution_journal(journal)?;
142 }
143 }
144
145 let operation_result = execute_operation_receipt(
146 config,
147 executor,
148 layout,
149 plan,
150 journal,
151 &operation,
152 command_lock.as_ref().map(CommandLifetimeLock::handle),
153 );
154 if let Some(command_lock) = command_lock {
155 command_lock
156 .finish()
157 .map_err(|error| backup_command_lock_error(&operation, error))?;
158 }
159
160 match operation_result {
161 Ok(receipt) => {
162 journal.record_operation_receipt(receipt)?;
163 layout.write_execution_journal(journal)?;
164 executed.push(BackupRunExecutedOperation::completed(&operation));
165 }
166 Err(error) => {
167 let receipt = crate::execution::BackupExecutionOperationReceipt::failed(
168 journal,
169 &operation,
170 Some(state_updated_at(config.updated_at.as_ref())),
171 error.to_string(),
172 );
173 journal.record_operation_receipt(receipt)?;
174 layout.write_execution_journal(journal)?;
175 executed.push(BackupRunExecutedOperation::failed(&operation));
176 return Err(error);
177 }
178 }
179 }
180}
181
182fn prepare_snapshot_create_attempt(
183 config: &BackupRunnerConfig,
184 executor: &mut impl BackupRunnerExecutor,
185 layout: &BackupLayout,
186 plan: &BackupPlan,
187 journal: &mut BackupExecutionJournal,
188 operation: &crate::execution::BackupExecutionJournalOperation,
189) -> Result<Option<BackupExecutionOperationReceipt>, BackupRunnerError> {
190 let recovering_previous_attempt = operation.snapshot_ids_before.is_some();
191 let snapshot_ids_before = if recovering_previous_attempt {
192 operation.snapshot_ids_before.clone().ok_or(
193 crate::execution::BackupExecutionJournalError::MissingField(
194 "operations[].snapshot_ids_before",
195 ),
196 )?
197 } else {
198 let target = operation_target(operation)?;
199 let snapshots = observe_snapshot_inventory(executor, operation, &target)?;
200 let mut snapshot_ids = snapshots
201 .into_iter()
202 .map(|snapshot| snapshot.snapshot_id)
203 .collect::<Vec<_>>();
204 snapshot_ids.sort();
205 snapshot_ids
206 };
207 journal.mark_snapshot_create_pending_at(
208 operation.sequence,
209 Some(state_updated_at(config.updated_at.as_ref())),
210 snapshot_ids_before,
211 )?;
212 layout.write_execution_journal(journal)?;
213 if !recovering_previous_attempt {
214 return Ok(None);
215 }
216 let pending_operation = journal
217 .operations
218 .iter()
219 .find(|candidate| candidate.sequence == operation.sequence)
220 .cloned()
221 .ok_or(
222 crate::execution::BackupExecutionJournalError::OperationNotFound(operation.sequence),
223 )?;
224 reconcile_pending_snapshot_create(executor, layout, plan, journal, &pending_operation)
225}
226
227fn reconcile_pending_snapshot_create(
228 executor: &mut impl BackupRunnerExecutor,
229 layout: &BackupLayout,
230 plan: &BackupPlan,
231 journal: &BackupExecutionJournal,
232 operation: &crate::execution::BackupExecutionJournalOperation,
233) -> Result<Option<BackupExecutionOperationReceipt>, BackupRunnerError> {
234 let target = operation_target(operation)?;
235 if let Some(receipt) = recorded_snapshot_receipt(layout, plan, journal, operation, &target)? {
236 return Ok(Some(receipt));
237 }
238 let baseline = operation.snapshot_ids_before.as_ref().ok_or(
239 crate::execution::BackupExecutionJournalError::MissingField(
240 "operations[].snapshot_ids_before",
241 ),
242 )?;
243 let baseline = baseline.iter().map(String::as_str).collect::<BTreeSet<_>>();
244 let observed = observe_snapshot_inventory(executor, operation, &target)?
245 .into_iter()
246 .map(|snapshot| (snapshot.snapshot_id.clone(), snapshot))
247 .collect::<BTreeMap<_, _>>();
248 let missing = baseline
249 .iter()
250 .filter(|snapshot_id| !observed.contains_key(**snapshot_id))
251 .map(|snapshot_id| (*snapshot_id).to_string())
252 .collect::<Vec<_>>();
253 if !missing.is_empty() {
254 return Err(BackupRunnerError::SnapshotInventoryLostBaseline {
255 sequence: operation.sequence,
256 operation_id: operation.operation_id.clone(),
257 snapshot_ids: missing,
258 });
259 }
260 let mut created = observed
261 .into_iter()
262 .filter(|(snapshot_id, _)| !baseline.contains(snapshot_id.as_str()))
263 .collect::<Vec<_>>();
264 match created.len() {
265 0 => Ok(None),
266 1 => {
267 let snapshot = created.pop().expect("one snapshot candidate").1;
268 persist_created_snapshot(layout, plan, journal, operation, &target, snapshot).map(Some)
269 }
270 _ => Err(BackupRunnerError::SnapshotIdentityAmbiguous {
271 sequence: operation.sequence,
272 operation_id: operation.operation_id.clone(),
273 snapshot_ids: created
274 .into_iter()
275 .map(|(snapshot_id, _)| snapshot_id)
276 .collect(),
277 }),
278 }
279}
280
281fn observe_snapshot_inventory(
282 executor: &mut impl BackupRunnerExecutor,
283 operation: &crate::execution::BackupExecutionJournalOperation,
284 target: &str,
285) -> Result<Vec<BackupRunnerSnapshot>, BackupRunnerError> {
286 let snapshots = executor.snapshot_inventory(target).map_err(|error| {
287 BackupRunnerError::SnapshotInventoryFailed {
288 sequence: operation.sequence,
289 status: error.status,
290 message: error.message,
291 }
292 })?;
293 let mut identities = BTreeSet::new();
294 for snapshot in &snapshots {
295 if snapshot.snapshot_id.trim().is_empty() {
296 return Err(BackupRunnerError::InvalidSnapshotIdentity {
297 sequence: operation.sequence,
298 operation_id: operation.operation_id.clone(),
299 });
300 }
301 if !identities.insert(snapshot.snapshot_id.as_str()) {
302 return Err(BackupRunnerError::DuplicateSnapshotIdentity {
303 sequence: operation.sequence,
304 operation_id: operation.operation_id.clone(),
305 snapshot_id: snapshot.snapshot_id.clone(),
306 });
307 }
308 }
309 Ok(snapshots)
310}
311
312fn reconcile_pending_stop(
313 executor: &mut impl BackupRunnerExecutor,
314 journal: &BackupExecutionJournal,
315 operation: &crate::execution::BackupExecutionJournalOperation,
316) -> Result<Option<BackupExecutionOperationReceipt>, BackupRunnerError> {
317 let target = operation.target_canister_id.as_deref().ok_or(
318 BackupRunnerError::MissingOperationTarget {
319 sequence: operation.sequence,
320 },
321 )?;
322 let status = executor.canister_status(target).map_err(|error| {
323 BackupRunnerError::CanisterStatusFailed {
324 sequence: operation.sequence,
325 status: error.status,
326 message: error.message,
327 }
328 })?;
329 match status {
330 BackupRunnerCanisterStatus::Running => Ok(None),
331 BackupRunnerCanisterStatus::Stopped => {
332 Ok(Some(BackupExecutionOperationReceipt::completed(
333 journal,
334 operation,
335 Some(current_timestamp_marker()),
336 )))
337 }
338 BackupRunnerCanisterStatus::Stopping => Err(BackupRunnerError::CanisterStatusUnsettled {
339 sequence: operation.sequence,
340 operation_id: operation.operation_id.clone(),
341 status: status.label(),
342 }),
343 }
344}
345
346fn finish_reconciled_command_lock(
347 operation: &crate::execution::BackupExecutionJournalOperation,
348 command_lock: Option<CommandLifetimeLock>,
349) -> Result<(), BackupRunnerError> {
350 command_lock
351 .ok_or_else(|| BackupRunnerError::MissingCommandLifetime {
352 sequence: operation.sequence,
353 operation_id: operation.operation_id.clone(),
354 })?
355 .finish()
356 .map_err(|error| backup_command_lock_error(operation, error))
357}
358
359fn reject_unknown_backup_command_outcome(
360 operation: &crate::execution::BackupExecutionJournalOperation,
361 command_lock: Option<CommandLifetimeLock>,
362) -> Result<(), BackupRunnerError> {
363 let Some(command_lock) = command_lock else {
364 return Ok(());
365 };
366 let lock_path = command_lock.path().to_string_lossy().to_string();
367 command_lock
368 .finish()
369 .map_err(|error| backup_command_lock_error(operation, error))?;
370 Err(BackupRunnerError::CommandOutcomeUnknown {
371 sequence: operation.sequence,
372 operation_id: operation.operation_id.clone(),
373 lock_path,
374 })
375}
376
377fn backup_command_lock(
378 layout: &BackupLayout,
379 operation: &crate::execution::BackupExecutionJournalOperation,
380) -> Result<Option<CommandLifetimeLock>, BackupRunnerError> {
381 if !matches!(
382 operation.kind,
383 crate::plan::BackupOperationKind::Stop
384 | crate::plan::BackupOperationKind::CreateSnapshot
385 | crate::plan::BackupOperationKind::Start
386 | crate::plan::BackupOperationKind::DownloadSnapshot
387 ) {
388 return Ok(None);
389 }
390
391 CommandLifetimeLock::acquire(&layout.execution_journal_path(), operation.sequence)
392 .map(Some)
393 .map_err(|error| backup_command_lock_error(operation, error))
394}
395
396fn backup_command_lock_error(
397 operation: &crate::execution::BackupExecutionJournalOperation,
398 error: CommandLifetimeLockError,
399) -> BackupRunnerError {
400 match error {
401 CommandLifetimeLockError::InFlight { lock_path } => BackupRunnerError::CommandInFlight {
402 sequence: operation.sequence,
403 operation_id: operation.operation_id.clone(),
404 lock_path,
405 },
406 CommandLifetimeLockError::UnsafeEntry { lock_path, kind } => {
407 BackupRunnerError::CommandLockUnsafeEntry {
408 sequence: operation.sequence,
409 operation_id: operation.operation_id.clone(),
410 lock_path,
411 kind,
412 }
413 }
414 CommandLifetimeLockError::Io(error) => BackupRunnerError::Io(error),
415 }
416}
417
418fn run_response(
419 plan: &BackupPlan,
420 journal: &BackupExecutionJournal,
421 executed: Vec<BackupRunExecutedOperation>,
422 max_steps_reached: bool,
423) -> BackupRunResponse {
424 let execution = journal.resume_summary();
425 BackupRunResponse {
426 run_id: plan.run_id.clone(),
427 plan_id: plan.plan_id.clone(),
428 backup_id: plan.run_id.clone(),
429 complete: execution.completed_operations + execution.skipped_operations
430 == execution.total_operations,
431 max_steps_reached,
432 executed_operation_count: executed.len(),
433 executed_operations: executed,
434 execution,
435 }
436}
437
438#[cfg(test)]
439mod tests;