canic_backup/runner/
mod.rs

1mod manifest;
2mod operations;
3mod types;
4
5pub use types::*;
6
7use crate::{
8    execution::{BackupExecutionJournal, BackupExecutionOperationState},
9    persistence::{BackupLayout, JournalLock},
10    plan::BackupPlan,
11    timestamp::{state_updated_at, timestamp_marker, timestamp_seconds},
12};
13use operations::execute_operation_receipt;
14
15const PREFLIGHT_TTL_SECONDS: u64 = 300;
16
17/// Execute a persisted backup plan through an injected host executor.
18pub fn backup_run_execute_with_executor(
19    config: &BackupRunnerConfig,
20    executor: &mut impl BackupRunnerExecutor,
21) -> Result<BackupRunResponse, BackupRunnerError> {
22    let layout = BackupLayout::new(config.out.clone());
23    let _lock = JournalLock::acquire(&layout.execution_journal_path())?;
24    let mut plan = layout.read_backup_plan()?;
25    let mut journal = if layout.execution_journal_path().is_file() {
26        layout.read_execution_journal()?
27    } else {
28        let journal = BackupExecutionJournal::from_plan(&plan)?;
29        layout.write_execution_journal(&journal)?;
30        journal
31    };
32    layout.verify_execution_integrity()?;
33
34    accept_preflight_if_needed(config, executor, &layout, &mut plan, &mut journal)?;
35    execute_ready_operations(config, executor, &layout, &plan, &mut journal)
36}
37
38fn accept_preflight_if_needed(
39    config: &BackupRunnerConfig,
40    executor: &mut impl BackupRunnerExecutor,
41    layout: &BackupLayout,
42    plan: &mut BackupPlan,
43    journal: &mut BackupExecutionJournal,
44) -> Result<(), BackupRunnerError> {
45    if journal.preflight_accepted {
46        return Ok(());
47    }
48
49    let validated_at = state_updated_at(config.updated_at.as_ref());
50    let expires_at = timestamp_marker(timestamp_seconds(&validated_at) + PREFLIGHT_TTL_SECONDS);
51    let preflight_id = format!("preflight-{}", plan.run_id);
52    let receipts = executor
53        .preflight_receipts(plan, &preflight_id, &validated_at, &expires_at)
54        .map_err(|error| BackupRunnerError::PreflightFailed {
55            status: error.status,
56            message: error.message,
57        })?;
58    plan.apply_execution_preflight_receipts(&receipts, &validated_at)?;
59    layout.write_backup_plan(plan)?;
60    journal.accept_preflight_receipts_at(&receipts, Some(validated_at))?;
61    layout.write_execution_journal(journal)?;
62    Ok(())
63}
64
65fn execute_ready_operations(
66    config: &BackupRunnerConfig,
67    executor: &mut impl BackupRunnerExecutor,
68    layout: &BackupLayout,
69    plan: &BackupPlan,
70    journal: &mut BackupExecutionJournal,
71) -> Result<BackupRunResponse, BackupRunnerError> {
72    let mut executed = Vec::new();
73
74    loop {
75        let summary = journal.resume_summary();
76        if summary.completed_operations + summary.skipped_operations == summary.total_operations {
77            return Ok(run_response(plan, journal, executed, false));
78        }
79        if config
80            .max_steps
81            .is_some_and(|max_steps| executed.len() >= max_steps)
82        {
83            return Ok(run_response(plan, journal, executed, true));
84        }
85
86        let operation = journal
87            .next_ready_operation()
88            .cloned()
89            .ok_or(BackupRunnerError::NoReadyOperation)?;
90        if operation.state == BackupExecutionOperationState::Blocked {
91            return Err(BackupRunnerError::Blocked {
92                reasons: operation.blocking_reasons,
93            });
94        }
95
96        if operation.state != BackupExecutionOperationState::Pending {
97            journal.mark_operation_pending_at(
98                operation.sequence,
99                Some(state_updated_at(config.updated_at.as_ref())),
100            )?;
101            layout.write_execution_journal(journal)?;
102        }
103
104        match execute_operation_receipt(config, executor, layout, plan, journal, &operation) {
105            Ok(receipt) => {
106                journal.record_operation_receipt(receipt)?;
107                layout.write_execution_journal(journal)?;
108                executed.push(BackupRunExecutedOperation::completed(&operation));
109            }
110            Err(error) => {
111                let receipt = crate::execution::BackupExecutionOperationReceipt::failed(
112                    journal,
113                    &operation,
114                    Some(state_updated_at(config.updated_at.as_ref())),
115                    error.to_string(),
116                );
117                journal.record_operation_receipt(receipt)?;
118                layout.write_execution_journal(journal)?;
119                executed.push(BackupRunExecutedOperation::failed(&operation));
120                return Err(error);
121            }
122        }
123    }
124}
125
126fn run_response(
127    plan: &BackupPlan,
128    journal: &BackupExecutionJournal,
129    executed: Vec<BackupRunExecutedOperation>,
130    max_steps_reached: bool,
131) -> BackupRunResponse {
132    let execution = journal.resume_summary();
133    BackupRunResponse {
134        run_id: plan.run_id.clone(),
135        plan_id: plan.plan_id.clone(),
136        backup_id: plan.run_id.clone(),
137        complete: execution.completed_operations + execution.skipped_operations
138            == execution.total_operations,
139        max_steps_reached,
140        executed_operation_count: executed.len(),
141        executed_operations: executed,
142        execution,
143    }
144}
145
146#[cfg(test)]
147mod tests;
canic_backup/runner/mod.rs

canic_backup/runner/
mod.rs
