canic_backup/restore/runner/
execute.rs

1//! Module: restore::runner::execute
2//!
3//! Responsibility: execute ready restore apply journal operations.
4//! Does not own: command rendering, apply journal validation, or restore planning.
5//! Boundary: claims operations, invokes an injected executor, and persists receipts.
6
7use super::{
8    RestoreApplyCommandOutputPair, RestoreApplyJournal, RestoreApplyOperationKind,
9    RestoreApplyOperationReceipt, RestoreApplyRunnerCommand,
10    artifact::{cleanup_upload_staging, stage_upload_artifact},
11    constants::{
12        RESTORE_RUN_COMMAND_EXIT_PREFIX, RESTORE_RUN_MISSING_UPLOADED_SNAPSHOT_ID,
13        RESTORE_RUN_OUTPUT_RECEIPT_LIMIT, RESTORE_RUN_STOPPED_PRECONDITION_FAILED,
14        RESTORE_RUN_VERIFICATION_EVIDENCE_MISMATCH,
15    },
16    io::{read_apply_journal_file, write_apply_journal_file},
17    precondition::{
18        RestoreObservedCanisterStatus, enforce_stopped_canister_precondition,
19        observe_canister_status, parse_canister_status_evidence,
20        stopped_canister_precondition_observation_failure,
21    },
22    status::{
23        enforce_restore_run_command_available, enforce_restore_run_executable,
24        parse_uploaded_snapshot_id, restore_command_unavailable_error,
25        restore_run_max_steps_reached, restore_run_next_action, restore_run_stopped_reason,
26    },
27    types::{
28        RestoreReconciledCommandSuccess, RestoreRunExecutedOperation, RestoreRunOperationReceipt,
29        RestoreRunPreparedOperation, RestoreRunResponse, RestoreRunResponseMode,
30        RestoreRunStepOutcome, RestoreRunnerCommandExecutor, RestoreRunnerConfig,
31        RestoreRunnerError, RestoreRunnerOutcome, RestoreStoppedPreconditionFailure,
32    },
33};
34use crate::{
35    persistence::{CommandLifetimeLock, CommandLifetimeLockError, JournalLock},
36    timestamp::state_updated_at,
37};
38use std::{collections::BTreeSet, ops::ControlFlow, path::Path};
39
40/// Execute ready restore apply journal operations through an injected command executor.
41pub fn restore_run_execute_with_executor(
42    config: &RestoreRunnerConfig,
43    executor: &mut impl RestoreRunnerCommandExecutor,
44) -> Result<RestoreRunResponse, RestoreRunnerError> {
45    let run = restore_run_execute_result_with_executor(config, executor)?;
46    if let Some(error) = run.error {
47        return Err(error);
48    }
49
50    Ok(run.response)
51}
52
53pub fn restore_run_execute_result_with_executor(
54    config: &RestoreRunnerConfig,
55    executor: &mut impl RestoreRunnerCommandExecutor,
56) -> Result<RestoreRunnerOutcome, RestoreRunnerError> {
57    restore_run_execute_result_with_terminal_writer(config, executor, &mut |path, journal| {
58        write_apply_journal_file(path, journal)
59    })
60}
61
62#[cfg(all(test, unix))]
63pub fn restore_run_execute_with_terminal_barriers(
64    config: &RestoreRunnerConfig,
65    executor: &mut impl RestoreRunnerCommandExecutor,
66    mut barriers: impl FnMut(crate::persistence::DurableWriteBarrier),
67) -> Result<RestoreRunResponse, RestoreRunnerError> {
68    let run =
69        restore_run_execute_result_with_terminal_writer(config, executor, &mut |path, journal| {
70            super::io::write_apply_journal_file_at_barriers(path, journal, &mut barriers)
71        })?;
72    if let Some(error) = run.error {
73        return Err(error);
74    }
75    Ok(run.response)
76}
77
78fn restore_run_execute_result_with_terminal_writer(
79    config: &RestoreRunnerConfig,
80    executor: &mut impl RestoreRunnerCommandExecutor,
81    terminal_writer: &mut impl FnMut(&Path, &RestoreApplyJournal) -> Result<(), RestoreRunnerError>,
82) -> Result<RestoreRunnerOutcome, RestoreRunnerError> {
83    let _lock = JournalLock::acquire(&config.journal)?;
84    let mut journal = read_apply_journal_file(&config.journal)?;
85    let mut executed_operations = Vec::new();
86    let mut operation_receipts = Vec::new();
87
88    loop {
89        let report = journal.report();
90        let max_steps_reached =
91            restore_run_max_steps_reached(config, executed_operations.len(), &report);
92        if report.complete || max_steps_reached {
93            return Ok(RestoreRunnerOutcome::ok(restore_run_execute_summary(
94                &journal,
95                executed_operations,
96                operation_receipts,
97                max_steps_reached,
98                config.updated_at.as_ref(),
99            )));
100        }
101
102        enforce_restore_run_executable(&journal, &report)?;
103        let prepared = restore_run_prepare_next_operation(config, executor, &mut journal)?;
104        let sequence = prepared.sequence;
105        match restore_run_execute_prepared_operation(
106            config,
107            executor,
108            &mut journal,
109            prepared,
110            terminal_writer,
111        )? {
112            RestoreRunStepOutcome::Completed {
113                executed_operation,
114                operation_receipt,
115            } => {
116                executed_operations.push(executed_operation);
117                operation_receipts.push(operation_receipt);
118            }
119            RestoreRunStepOutcome::Failed {
120                executed_operation,
121                operation_receipt,
122                status,
123            } => {
124                executed_operations.push(executed_operation);
125                operation_receipts.push(operation_receipt);
126                let response = restore_run_execute_summary(
127                    &journal,
128                    executed_operations,
129                    operation_receipts,
130                    false,
131                    config.updated_at.as_ref(),
132                );
133                return Ok(RestoreRunnerOutcome {
134                    response,
135                    error: Some(RestoreRunnerError::CommandFailed { sequence, status }),
136                });
137            }
138        }
139    }
140}
141
142fn restore_run_prepare_next_operation(
143    config: &RestoreRunnerConfig,
144    executor: &mut impl RestoreRunnerCommandExecutor,
145    journal: &mut RestoreApplyJournal,
146) -> Result<RestoreRunPreparedOperation, RestoreRunnerError> {
147    let preview = journal.next_command_preview_with_config(&config.command);
148    enforce_restore_run_command_available(&preview)?;
149
150    let operation = preview
151        .operation
152        .clone()
153        .ok_or_else(|| restore_command_unavailable_error(&preview))?;
154    let mut command = preview
155        .command
156        .clone()
157        .ok_or_else(|| restore_command_unavailable_error(&preview))?;
158    let sequence = operation.sequence;
159    let command_lock = restore_command_lock(config, &operation)?;
160    let attempt = journal
161        .operation_receipts
162        .iter()
163        .filter(|receipt| receipt.sequence == sequence)
164        .count()
165        + 1;
166    let reconciled_success = reconcile_restore_operation(
167        config,
168        executor,
169        &operation,
170        command_lock.as_ref().map(CommandLifetimeLock::handle),
171    )?;
172    if let Some(reconciled_success) = reconciled_success {
173        return Ok(RestoreRunPreparedOperation {
174            operation,
175            command: reconciled_success.command.clone(),
176            sequence,
177            attempt,
178            command_lock,
179            _staged_artifact: None,
180            reconciled_success: Some(reconciled_success),
181        });
182    }
183    let staged_artifact = stage_upload_artifact(config, journal, &operation)?;
184    if let Some(staged) = &staged_artifact {
185        command = crate::restore::RestoreApplyRunnerCommand::from_operation_with_artifact_path(
186            &operation,
187            journal,
188            &config.command,
189            Some(staged.artifact_path()),
190        )
191        .ok_or_else(|| restore_command_unavailable_error(&preview))?;
192    }
193    let upload_inventory_before = if operation.operation
194        == RestoreApplyOperationKind::UploadSnapshot
195        && operation.snapshot_ids_before.is_none()
196    {
197        Some(
198            observe_snapshot_inventory(
199                config,
200                executor,
201                &operation,
202                command_lock.as_ref().map(CommandLifetimeLock::handle),
203            )?
204            .snapshot_ids,
205        )
206    } else {
207        None
208    };
209
210    enforce_apply_claim_sequence(sequence, journal)?;
211    let pending_at = Some(state_updated_at(config.updated_at.as_ref()));
212    if operation.operation == RestoreApplyOperationKind::UploadSnapshot {
213        let snapshot_ids_before = operation
214            .snapshot_ids_before
215            .clone()
216            .or(upload_inventory_before)
217            .ok_or(RestoreRunnerError::InvalidSnapshotInventory { sequence })?;
218        journal.mark_upload_snapshot_pending_at(sequence, pending_at, snapshot_ids_before)?;
219    } else {
220        journal.mark_operation_pending_at(sequence, pending_at)?;
221    }
222    write_apply_journal_file(&config.journal, journal)?;
223
224    Ok(RestoreRunPreparedOperation {
225        operation,
226        command,
227        sequence,
228        attempt,
229        command_lock,
230        _staged_artifact: staged_artifact,
231        reconciled_success: None,
232    })
233}
234
235fn restore_run_execute_prepared_operation(
236    config: &RestoreRunnerConfig,
237    executor: &mut impl RestoreRunnerCommandExecutor,
238    journal: &mut RestoreApplyJournal,
239    prepared: RestoreRunPreparedOperation,
240    terminal_writer: &mut impl FnMut(&Path, &RestoreApplyJournal) -> Result<(), RestoreRunnerError>,
241) -> Result<RestoreRunStepOutcome, RestoreRunnerError> {
242    if prepared.reconciled_success.is_some() {
243        return restore_run_commit_reconciled_success(config, journal, prepared, terminal_writer);
244    }
245    let mut prepared = match restore_run_enforce_stopped_precondition(
246        config,
247        executor,
248        journal,
249        prepared,
250        terminal_writer,
251    )? {
252        ControlFlow::Continue(prepared) => prepared,
253        ControlFlow::Break(outcome) => return Ok(outcome),
254    };
255    let command_lifetime = prepared
256        .command_lock
257        .as_ref()
258        .map(CommandLifetimeLock::handle);
259
260    let output = executor.execute(&prepared.command, command_lifetime);
261    finish_restore_command_lock(&mut prepared)?;
262    let output = output?;
263    cleanup_upload_staging(config, &prepared.operation)?;
264    let status_label = output.status;
265    let output_pair = RestoreApplyCommandOutputPair::from_bytes(
266        &output.stdout,
267        &output.stderr,
268        RESTORE_RUN_OUTPUT_RECEIPT_LIMIT,
269    );
270
271    if output.success {
272        if matches!(
273            prepared.operation.operation,
274            RestoreApplyOperationKind::VerifyMember | RestoreApplyOperationKind::VerifyDeployment
275        ) && !verification_output_matches(&output_pair, &prepared.operation)
276        {
277            return restore_run_commit_command_failure(
278                config,
279                journal,
280                prepared,
281                RESTORE_RUN_VERIFICATION_EVIDENCE_MISMATCH.to_string(),
282                output_pair,
283                terminal_writer,
284            );
285        }
286        let is_upload_snapshot =
287            prepared.operation.operation == RestoreApplyOperationKind::UploadSnapshot;
288        let uploaded_snapshot_id = is_upload_snapshot
289            .then(|| parse_uploaded_snapshot_id(&String::from_utf8_lossy(&output.stdout)))
290            .flatten();
291        if is_upload_snapshot && uploaded_snapshot_id.is_none() {
292            return restore_run_commit_missing_uploaded_snapshot_id(
293                config,
294                journal,
295                prepared,
296                output_pair,
297                terminal_writer,
298            );
299        }
300
301        return restore_run_commit_command_success(
302            config,
303            journal,
304            prepared,
305            status_label,
306            output_pair,
307            uploaded_snapshot_id,
308            terminal_writer,
309        );
310    }
311
312    restore_run_commit_command_failure(
313        config,
314        journal,
315        prepared,
316        status_label,
317        output_pair,
318        terminal_writer,
319    )
320}
321
322fn restore_run_enforce_stopped_precondition(
323    config: &RestoreRunnerConfig,
324    executor: &mut impl RestoreRunnerCommandExecutor,
325    journal: &mut RestoreApplyJournal,
326    mut prepared: RestoreRunPreparedOperation,
327    terminal_writer: &mut impl FnMut(&Path, &RestoreApplyJournal) -> Result<(), RestoreRunnerError>,
328) -> Result<ControlFlow<RestoreRunStepOutcome, RestoreRunPreparedOperation>, RestoreRunnerError> {
329    if !prepared.command.requires_stopped_canister {
330        return Ok(ControlFlow::Continue(prepared));
331    }
332
333    let precondition = enforce_stopped_canister_precondition(
334        config,
335        executor,
336        &prepared.operation,
337        prepared.attempt,
338        config.updated_at.as_ref(),
339        prepared
340            .command_lock
341            .as_ref()
342            .map(CommandLifetimeLock::handle),
343    );
344    match precondition {
345        Ok(Some(outcome)) => {
346            finish_restore_command_lock(&mut prepared)?;
347            let outcome = restore_run_commit_precondition_failure(
348                config,
349                journal,
350                prepared,
351                outcome,
352                terminal_writer,
353            )?;
354            Ok(ControlFlow::Break(outcome))
355        }
356        Ok(None) => Ok(ControlFlow::Continue(prepared)),
357        Err(error) => {
358            let outcome = stopped_canister_precondition_observation_failure(
359                config,
360                &prepared.operation,
361                prepared.attempt,
362                config.updated_at.as_ref(),
363            );
364            finish_restore_command_lock(&mut prepared)?;
365            let _ = restore_run_commit_precondition_failure(
366                config,
367                journal,
368                prepared,
369                outcome,
370                terminal_writer,
371            )?;
372            Err(error)
373        }
374    }
375}
376
377fn restore_run_commit_reconciled_success(
378    config: &RestoreRunnerConfig,
379    journal: &mut RestoreApplyJournal,
380    mut prepared: RestoreRunPreparedOperation,
381    terminal_writer: &mut impl FnMut(&Path, &RestoreApplyJournal) -> Result<(), RestoreRunnerError>,
382) -> Result<RestoreRunStepOutcome, RestoreRunnerError> {
383    let reconciled = prepared
384        .reconciled_success
385        .take()
386        .expect("reconciled restore operation");
387    prepared.command = reconciled.command;
388    cleanup_upload_staging(config, &prepared.operation)?;
389    finish_restore_command_lock(&mut prepared)?;
390    restore_run_commit_command_success(
391        config,
392        journal,
393        prepared,
394        reconciled.status,
395        reconciled.output,
396        reconciled.uploaded_snapshot_id,
397        terminal_writer,
398    )
399}
400
401fn verification_output_matches(
402    output: &RestoreApplyCommandOutputPair,
403    operation: &crate::restore::RestoreApplyJournalOperation,
404) -> bool {
405    verification_json_matches(
406        &output.stdout.text,
407        operation.expected_module_hash.as_deref(),
408    ) || verification_json_matches(
409        &output.stderr.text,
410        operation.expected_module_hash.as_deref(),
411    )
412}
413
414fn verification_json_matches(output: &str, expected_module_hash: Option<&str>) -> bool {
415    let Some(evidence) = parse_canister_status_evidence(output) else {
416        return false;
417    };
418    if evidence.status != RestoreObservedCanisterStatus::Running {
419        return false;
420    }
421    let Some(expected_module_hash) = expected_module_hash else {
422        return true;
423    };
424    evidence.module_hash.as_deref().is_some_and(|actual| {
425        normalize_module_hash(actual) == normalize_module_hash(expected_module_hash)
426    })
427}
428
429fn normalize_module_hash(module_hash: &str) -> String {
430    module_hash
431        .trim()
432        .strip_prefix("0x")
433        .unwrap_or_else(|| module_hash.trim())
434        .to_ascii_lowercase()
435}
436
437#[cfg(test)]
438mod verification_tests {
439    use super::*;
440
441    #[test]
442    fn verification_requires_running_status_and_matching_module_hash() {
443        assert!(verification_json_matches(
444            r#"{"status":"Running","module_hash":"0xABCD"}"#,
445            Some("abcd")
446        ));
447        assert!(!verification_json_matches(
448            r#"{"status":"Running","module_hash":"0xDEAD"}"#,
449            Some("abcd")
450        ));
451        assert!(!verification_json_matches(
452            r#"{"status":"Stopped","module_hash":"0xABCD"}"#,
453            Some("abcd")
454        ));
455        assert!(!verification_json_matches("not json", Some("abcd")));
456    }
457}
458
459fn reconcile_restore_operation(
460    config: &RestoreRunnerConfig,
461    executor: &mut impl RestoreRunnerCommandExecutor,
462    operation: &crate::restore::RestoreApplyJournalOperation,
463    command_lifetime: Option<crate::persistence::CommandLifetimeHandle>,
464) -> Result<Option<RestoreReconciledCommandSuccess>, RestoreRunnerError> {
465    let recovering = operation.state == crate::restore::RestoreApplyOperationState::Pending
466        || (operation.operation == RestoreApplyOperationKind::UploadSnapshot
467            && operation.snapshot_ids_before.is_some());
468    if !recovering {
469        return Ok(None);
470    }
471
472    match operation.operation {
473        RestoreApplyOperationKind::StopCanister => reconcile_lifecycle_operation(
474            config,
475            executor,
476            operation,
477            command_lifetime,
478            RestoreObservedCanisterStatus::Stopped,
479        ),
480        RestoreApplyOperationKind::StartCanister => reconcile_lifecycle_operation(
481            config,
482            executor,
483            operation,
484            command_lifetime,
485            RestoreObservedCanisterStatus::Running,
486        ),
487        RestoreApplyOperationKind::UploadSnapshot => {
488            reconcile_upload_operation(config, executor, operation, command_lifetime)
489        }
490        RestoreApplyOperationKind::LoadSnapshot
491        | RestoreApplyOperationKind::VerifyMember
492        | RestoreApplyOperationKind::VerifyDeployment => Ok(None),
493    }
494}
495
496fn reconcile_lifecycle_operation(
497    config: &RestoreRunnerConfig,
498    executor: &mut impl RestoreRunnerCommandExecutor,
499    operation: &crate::restore::RestoreApplyJournalOperation,
500    command_lifetime: Option<crate::persistence::CommandLifetimeHandle>,
501    completed_status: RestoreObservedCanisterStatus,
502) -> Result<Option<RestoreReconciledCommandSuccess>, RestoreRunnerError> {
503    let observation = observe_canister_status(config, executor, operation, command_lifetime)?;
504    if !observation.success {
505        return Err(RestoreRunnerError::CanisterStatusObservationFailed {
506            sequence: operation.sequence,
507            status: observation.status_label,
508        });
509    }
510    let status = observation
511        .status
512        .ok_or(RestoreRunnerError::CanisterStatusUnknown {
513            sequence: operation.sequence,
514        })?;
515    if status == RestoreObservedCanisterStatus::Stopping {
516        return Err(RestoreRunnerError::CanisterStatusUnsettled {
517            sequence: operation.sequence,
518        });
519    }
520    if status != completed_status {
521        return Ok(None);
522    }
523
524    Ok(Some(RestoreReconciledCommandSuccess {
525        command: observation.command,
526        status: observation.status_label,
527        output: observation.output,
528        uploaded_snapshot_id: None,
529    }))
530}
531
532fn reconcile_upload_operation(
533    config: &RestoreRunnerConfig,
534    executor: &mut impl RestoreRunnerCommandExecutor,
535    operation: &crate::restore::RestoreApplyJournalOperation,
536    command_lifetime: Option<crate::persistence::CommandLifetimeHandle>,
537) -> Result<Option<RestoreReconciledCommandSuccess>, RestoreRunnerError> {
538    let baseline = operation.snapshot_ids_before.as_ref().ok_or(
539        RestoreRunnerError::InvalidSnapshotInventory {
540            sequence: operation.sequence,
541        },
542    )?;
543    let observation = observe_snapshot_inventory(config, executor, operation, command_lifetime)?;
544    let baseline = baseline.iter().map(String::as_str).collect::<BTreeSet<_>>();
545    let observed = observation
546        .snapshot_ids
547        .iter()
548        .map(String::as_str)
549        .collect::<BTreeSet<_>>();
550    let missing = baseline
551        .difference(&observed)
552        .map(|snapshot_id| (*snapshot_id).to_string())
553        .collect::<Vec<_>>();
554    if !missing.is_empty() {
555        return Err(RestoreRunnerError::SnapshotInventoryLostBaseline {
556            sequence: operation.sequence,
557            snapshot_ids: missing,
558        });
559    }
560    let created = observed
561        .difference(&baseline)
562        .map(|snapshot_id| (*snapshot_id).to_string())
563        .collect::<Vec<_>>();
564    match created.as_slice() {
565        [] => Ok(None),
566        [snapshot_id] => Ok(Some(RestoreReconciledCommandSuccess {
567            command: observation.command,
568            status: observation.status,
569            output: observation.output,
570            uploaded_snapshot_id: Some(snapshot_id.clone()),
571        })),
572        _ => Err(RestoreRunnerError::UploadedSnapshotIdentityAmbiguous {
573            sequence: operation.sequence,
574            snapshot_ids: created,
575        }),
576    }
577}
578
579struct RestoreSnapshotInventoryObservation {
580    command: RestoreApplyRunnerCommand,
581    status: String,
582    output: RestoreApplyCommandOutputPair,
583    snapshot_ids: Vec<String>,
584}
585
586#[derive(serde::Deserialize)]
587#[serde(deny_unknown_fields)]
588struct RestoreSnapshotInventoryResponse {
589    snapshots: Vec<RestoreSnapshotInventoryEntry>,
590}
591
592#[derive(serde::Deserialize)]
593#[serde(deny_unknown_fields)]
594struct RestoreSnapshotInventoryEntry {
595    snapshot_id: String,
596    #[serde(rename = "taken_at_timestamp")]
597    _taken_at_timestamp: Option<u64>,
598    #[serde(rename = "total_size_bytes")]
599    _total_size_bytes: Option<u64>,
600}
601
602fn observe_snapshot_inventory(
603    config: &RestoreRunnerConfig,
604    executor: &mut impl RestoreRunnerCommandExecutor,
605    operation: &crate::restore::RestoreApplyJournalOperation,
606    command_lifetime: Option<crate::persistence::CommandLifetimeHandle>,
607) -> Result<RestoreSnapshotInventoryObservation, RestoreRunnerError> {
608    let mut args = vec![
609        "canister".to_string(),
610        "snapshot".to_string(),
611        "list".to_string(),
612        operation.target_canister.clone(),
613        "--json".to_string(),
614    ];
615    if let Some(environment) = &config.command.environment {
616        args.push("-e".to_string());
617        args.push(environment.clone());
618    }
619    let command = RestoreApplyRunnerCommand {
620        program: config.command.program.clone(),
621        args,
622        mutates: false,
623        requires_stopped_canister: false,
624        note: "observes target snapshot inventory for upload reconciliation".to_string(),
625    };
626    let raw = executor.execute(&command, command_lifetime)?;
627    let output = RestoreApplyCommandOutputPair::from_bytes(
628        &raw.stdout,
629        &raw.stderr,
630        RESTORE_RUN_OUTPUT_RECEIPT_LIMIT,
631    );
632    if !raw.success {
633        return Err(RestoreRunnerError::SnapshotInventoryObservationFailed {
634            sequence: operation.sequence,
635            status: raw.status,
636        });
637    }
638    let inventory = serde_json::from_slice::<RestoreSnapshotInventoryResponse>(&raw.stdout)
639        .map_err(|_| RestoreRunnerError::InvalidSnapshotInventory {
640            sequence: operation.sequence,
641        })?;
642    let mut identities = BTreeSet::new();
643    for snapshot in inventory.snapshots {
644        let snapshot_id = (!snapshot.snapshot_id.trim().is_empty())
645            .then(|| snapshot.snapshot_id.trim())
646            .ok_or(RestoreRunnerError::InvalidSnapshotInventory {
647                sequence: operation.sequence,
648            })?;
649        if !identities.insert(snapshot_id.to_string()) {
650            return Err(RestoreRunnerError::InvalidSnapshotInventory {
651                sequence: operation.sequence,
652            });
653        }
654    }
655    Ok(RestoreSnapshotInventoryObservation {
656        command,
657        status: raw.status,
658        output,
659        snapshot_ids: identities.into_iter().collect(),
660    })
661}
662
663fn restore_command_lock(
664    config: &RestoreRunnerConfig,
665    operation: &crate::restore::RestoreApplyJournalOperation,
666) -> Result<Option<CommandLifetimeLock>, RestoreRunnerError> {
667    if !matches!(
668        operation.operation,
669        RestoreApplyOperationKind::UploadSnapshot
670            | RestoreApplyOperationKind::StopCanister
671            | RestoreApplyOperationKind::LoadSnapshot
672            | RestoreApplyOperationKind::StartCanister
673    ) {
674        return Ok(None);
675    }
676
677    CommandLifetimeLock::acquire(&config.journal, operation.sequence)
678        .map(Some)
679        .map_err(|error| restore_command_lock_error(operation, error))
680}
681
682fn finish_restore_command_lock(
683    prepared: &mut RestoreRunPreparedOperation,
684) -> Result<(), RestoreRunnerError> {
685    let Some(command_lock) = prepared.command_lock.take() else {
686        return Ok(());
687    };
688    command_lock
689        .finish()
690        .map_err(|error| restore_command_lock_error(&prepared.operation, error))
691}
692
693fn restore_command_lock_error(
694    operation: &crate::restore::RestoreApplyJournalOperation,
695    error: CommandLifetimeLockError,
696) -> RestoreRunnerError {
697    match error {
698        CommandLifetimeLockError::InFlight { lock_path } => RestoreRunnerError::CommandInFlight {
699            sequence: operation.sequence,
700            operation: operation.operation.clone(),
701            lock_path,
702        },
703        CommandLifetimeLockError::UnsafeEntry { lock_path, kind } => {
704            RestoreRunnerError::CommandLockUnsafeEntry {
705                sequence: operation.sequence,
706                operation: operation.operation.clone(),
707                lock_path,
708                kind,
709            }
710        }
711        CommandLifetimeLockError::Io(error) => RestoreRunnerError::Io(error),
712    }
713}
714
715fn restore_run_commit_missing_uploaded_snapshot_id(
716    config: &RestoreRunnerConfig,
717    journal: &mut RestoreApplyJournal,
718    prepared: RestoreRunPreparedOperation,
719    output_pair: RestoreApplyCommandOutputPair,
720    terminal_writer: &mut impl FnMut(&Path, &RestoreApplyJournal) -> Result<(), RestoreRunnerError>,
721) -> Result<RestoreRunStepOutcome, RestoreRunnerError> {
722    let failed_updated_at = state_updated_at(config.updated_at.as_ref());
723    let status = RESTORE_RUN_MISSING_UPLOADED_SNAPSHOT_ID.to_string();
724    journal.mark_operation_failed_at(
725        prepared.sequence,
726        status.clone(),
727        Some(failed_updated_at.clone()),
728    )?;
729    journal.record_operation_receipt(RestoreApplyOperationReceipt::command_failed(
730        &prepared.operation,
731        prepared.command.clone(),
732        status.clone(),
733        Some(failed_updated_at.clone()),
734        output_pair,
735        prepared.attempt,
736        status.clone(),
737    ))?;
738    terminal_writer(&config.journal, journal)?;
739
740    Ok(RestoreRunStepOutcome::Failed {
741        executed_operation: RestoreRunExecutedOperation::failed(
742            prepared.operation.clone(),
743            prepared.command.clone(),
744            RESTORE_RUN_MISSING_UPLOADED_SNAPSHOT_ID.to_string(),
745        ),
746        operation_receipt: RestoreRunOperationReceipt::failed(
747            prepared.operation,
748            prepared.command,
749            status.clone(),
750            Some(failed_updated_at),
751        ),
752        status,
753    })
754}
755
756fn restore_run_commit_precondition_failure(
757    config: &RestoreRunnerConfig,
758    journal: &mut RestoreApplyJournal,
759    prepared: RestoreRunPreparedOperation,
760    outcome: RestoreStoppedPreconditionFailure,
761    terminal_writer: &mut impl FnMut(&Path, &RestoreApplyJournal) -> Result<(), RestoreRunnerError>,
762) -> Result<RestoreRunStepOutcome, RestoreRunnerError> {
763    let failed_updated_at = state_updated_at(config.updated_at.as_ref());
764    journal.mark_operation_failed_at(
765        prepared.sequence,
766        outcome.failure_reason.clone(),
767        Some(failed_updated_at.clone()),
768    )?;
769    journal.record_operation_receipt(RestoreApplyOperationReceipt::command_failed(
770        &prepared.operation,
771        outcome.command.clone(),
772        outcome.status_label.clone(),
773        Some(failed_updated_at.clone()),
774        outcome.output,
775        prepared.attempt,
776        outcome.failure_reason,
777    ))?;
778    terminal_writer(&config.journal, journal)?;
779
780    Ok(RestoreRunStepOutcome::Failed {
781        executed_operation: RestoreRunExecutedOperation::failed(
782            prepared.operation.clone(),
783            outcome.command.clone(),
784            outcome.status_label.clone(),
785        ),
786        operation_receipt: RestoreRunOperationReceipt::failed(
787            prepared.operation,
788            outcome.command,
789            outcome.status_label,
790            Some(failed_updated_at),
791        ),
792        status: RESTORE_RUN_STOPPED_PRECONDITION_FAILED.to_string(),
793    })
794}
795
796fn restore_run_commit_command_success(
797    config: &RestoreRunnerConfig,
798    journal: &mut RestoreApplyJournal,
799    prepared: RestoreRunPreparedOperation,
800    status_label: String,
801    output_pair: RestoreApplyCommandOutputPair,
802    uploaded_snapshot_id: Option<String>,
803    terminal_writer: &mut impl FnMut(&Path, &RestoreApplyJournal) -> Result<(), RestoreRunnerError>,
804) -> Result<RestoreRunStepOutcome, RestoreRunnerError> {
805    let completed_updated_at = state_updated_at(config.updated_at.as_ref());
806    journal.mark_operation_completed_at(prepared.sequence, Some(completed_updated_at.clone()))?;
807    journal.record_operation_receipt(RestoreApplyOperationReceipt::command_completed(
808        &prepared.operation,
809        prepared.command.clone(),
810        status_label.clone(),
811        Some(completed_updated_at.clone()),
812        output_pair,
813        prepared.attempt,
814        uploaded_snapshot_id,
815    ))?;
816    terminal_writer(&config.journal, journal)?;
817
818    Ok(RestoreRunStepOutcome::Completed {
819        executed_operation: RestoreRunExecutedOperation::completed(
820            prepared.operation.clone(),
821            prepared.command.clone(),
822            status_label.clone(),
823        ),
824        operation_receipt: RestoreRunOperationReceipt::completed(
825            prepared.operation,
826            prepared.command,
827            status_label,
828            Some(completed_updated_at),
829        ),
830    })
831}
832
833fn restore_run_commit_command_failure(
834    config: &RestoreRunnerConfig,
835    journal: &mut RestoreApplyJournal,
836    prepared: RestoreRunPreparedOperation,
837    status_label: String,
838    output_pair: RestoreApplyCommandOutputPair,
839    terminal_writer: &mut impl FnMut(&Path, &RestoreApplyJournal) -> Result<(), RestoreRunnerError>,
840) -> Result<RestoreRunStepOutcome, RestoreRunnerError> {
841    let failed_updated_at = state_updated_at(config.updated_at.as_ref());
842    let failure_reason = format!("{RESTORE_RUN_COMMAND_EXIT_PREFIX}-{status_label}");
843    journal.mark_operation_failed_at(
844        prepared.sequence,
845        failure_reason.clone(),
846        Some(failed_updated_at.clone()),
847    )?;
848    journal.record_operation_receipt(RestoreApplyOperationReceipt::command_failed(
849        &prepared.operation,
850        prepared.command.clone(),
851        status_label.clone(),
852        Some(failed_updated_at.clone()),
853        output_pair,
854        prepared.attempt,
855        failure_reason,
856    ))?;
857    terminal_writer(&config.journal, journal)?;
858
859    Ok(RestoreRunStepOutcome::Failed {
860        executed_operation: RestoreRunExecutedOperation::failed(
861            prepared.operation.clone(),
862            prepared.command.clone(),
863            status_label.clone(),
864        ),
865        operation_receipt: RestoreRunOperationReceipt::failed(
866            prepared.operation,
867            prepared.command,
868            status_label.clone(),
869            Some(failed_updated_at),
870        ),
871        status: status_label,
872    })
873}
874
875fn restore_run_execute_summary(
876    journal: &RestoreApplyJournal,
877    executed_operations: Vec<RestoreRunExecutedOperation>,
878    operation_receipts: Vec<RestoreRunOperationReceipt>,
879    max_steps_reached: bool,
880    requested_state_updated_at: Option<&String>,
881) -> RestoreRunResponse {
882    let report = journal.report();
883    let executed_operation_count = executed_operations.len();
884    let stopped_reason = restore_run_stopped_reason(&report, max_steps_reached, true);
885    let next_action = restore_run_next_action(&report);
886
887    let mut response = RestoreRunResponse::from_report(
888        journal.backup_id.clone(),
889        report,
890        RestoreRunResponseMode::execute(stopped_reason, next_action),
891    );
892    response.set_requested_state_updated_at(requested_state_updated_at);
893    response.max_steps_reached = Some(max_steps_reached);
894    response.executed_operation_count = Some(executed_operation_count);
895    response.executed_operations = executed_operations;
896    response.set_operation_receipts(operation_receipts);
897    response
898}
899
900fn enforce_apply_claim_sequence(
901    expected: usize,
902    journal: &RestoreApplyJournal,
903) -> Result<(), RestoreRunnerError> {
904    let actual = journal
905        .next_transition_operation()
906        .map(|operation| operation.sequence);
907
908    if actual == Some(expected) {
909        return Ok(());
910    }
911
912    Err(RestoreRunnerError::ClaimSequenceMismatch { expected, actual })
913}
canic_backup/restore/runner/execute.rs

canic_backup/restore/runner/
execute.rs
