canic-backup 0.94.3

//! Module: persistence::tests::operational_readiness::pending_claim
//!
//! Responsibility: execute every backup pending-claim crash case.
//! Does not own: operation-specific effect reconciliation or terminal receipts.
//! Boundary: proves claim publication precedes effects and selects restart policy.

use super::{
    durable_write_barrier, hold_at_acknowledged_barrier, kill_child_at_acknowledged_barrier,
};
use crate::{
    execution::{
        BackupExecutionJournal, BackupExecutionJournalOperation, BackupExecutionOperationState,
    },
    operational_readiness::manifest::{
        assert_case_defined, backup_operation_label, backup_post_preflight_operations,
    },
    persistence::{BackupLayout, write_json_durable_at_barriers},
    plan::BackupOperationKind,
    runner::{BackupRunnerConfig, BackupRunnerError, backup_run_execute_with_executor},
    test_support::{FakeBackupRunnerExecutor, temp_dir},
};

use std::{fs, path::Path, process::Command};

const PENDING_CLAIM_CHILD_ROOT_ENV: &str = "CANIC_TEST_PENDING_CLAIM_ROOT";
const PENDING_CLAIM_CHILD_OPERATION_ENV: &str = "CANIC_TEST_PENDING_CLAIM_OPERATION";
const PENDING_CLAIM_CHILD_BARRIER_ENV: &str = "CANIC_TEST_PENDING_CLAIM_BARRIER";
const PENDING_CLAIM_CHILD_HANDSHAKE_ENV: &str = "CANIC_TEST_PENDING_CLAIM_HANDSHAKE";

#[test]
fn pending_claim_publication_selects_the_exact_restart_policy() {
    let Some(root) = std::env::var_os(PENDING_CLAIM_CHILD_ROOT_ENV) else {
        for (prior_steps, operation) in backup_post_preflight_operations().into_iter().enumerate() {
            let label = backup_operation_label(&operation);
            for barrier_name in ["before-rename", "after-directory-sync"] {
                let side = if barrier_name == "before-rename" {
                    "before-durable-write"
                } else {
                    "after-durable-write"
                };
                assert_case_defined(&format!("CANIC-094-B04/{label}/{side}"));
                prove_pending_claim_barrier(operation.clone(), label, prior_steps, barrier_name);
            }
        }
        return;
    };

    let root = std::path::PathBuf::from(root);
    let operation_label =
        std::env::var(PENDING_CLAIM_CHILD_OPERATION_ENV).expect("pending operation label");
    let barrier_name =
        std::env::var(PENDING_CLAIM_CHILD_BARRIER_ENV).expect("pending claim barrier");
    let handshake_root = std::path::PathBuf::from(
        std::env::var_os(PENDING_CLAIM_CHILD_HANDSHAKE_ENV).expect("pending claim handshake root"),
    );
    let layout = BackupLayout::new(root);
    let mut journal = layout
        .read_execution_journal()
        .expect("read pre-claim execution journal");
    let operation = journal
        .next_ready_operation()
        .cloned()
        .expect("next pending-claim operation");
    assert_eq!(backup_operation_label(&operation.kind), operation_label);
    journal
        .mark_operation_pending_at(operation.sequence, Some("unix:20".to_string()))
        .expect("mark operation pending in crash child");
    let barrier = durable_write_barrier(&barrier_name);
    write_json_durable_at_barriers(&layout.execution_journal_path(), &journal, |observed| {
        if observed == barrier {
            hold_at_acknowledged_barrier(&handshake_root);
        }
    })
    .expect("write pending claim in crash child");
}

fn prove_pending_claim_barrier(
    expected_kind: BackupOperationKind,
    operation_label: &str,
    prior_steps: usize,
    barrier_name: &str,
) {
    let root = prepare_operation(operation_label, prior_steps, &expected_kind);
    let handshake_root = temp_dir(&format!(
        "canic-backup-pending-claim-handshake-{operation_label}-{barrier_name}"
    ));
    fs::create_dir_all(&handshake_root).expect("create pending claim handshake root");
    let layout = BackupLayout::new(root.clone());
    let before_claim = layout
        .read_execution_journal()
        .expect("read execution journal before claim");
    let expected_operation = before_claim
        .next_ready_operation()
        .cloned()
        .expect("next operation before claim");
    let receipt_count = before_claim.operation_receipts.len();
    let expected_command = expected_command(&before_claim, &expected_operation);
    let mut child = Command::new(std::env::current_exe().expect("resolve test executable"))
        .args([
            "--exact",
            "persistence::tests::operational_readiness::pending_claim::pending_claim_publication_selects_the_exact_restart_policy",
            "--nocapture",
        ])
        .env(PENDING_CLAIM_CHILD_ROOT_ENV, &root)
        .env(PENDING_CLAIM_CHILD_OPERATION_ENV, operation_label)
        .env(PENDING_CLAIM_CHILD_BARRIER_ENV, barrier_name)
        .env(PENDING_CLAIM_CHILD_HANDSHAKE_ENV, &handshake_root)
        .spawn()
        .expect("spawn pending claim child");

    kill_child_at_acknowledged_barrier(&mut child, &handshake_root);
    let observed = layout
        .read_execution_journal()
        .expect("read execution journal after claim crash");
    let observed_operation = operation_at(&observed, expected_operation.sequence);

    if barrier_name == "before-rename" {
        assert_eq!(observed, before_claim);
        prove_ready_operation_resumes(&root, &expected_operation, receipt_count, expected_command);
    } else {
        assert_eq!(
            observed_operation.state,
            BackupExecutionOperationState::Pending
        );
        assert_eq!(observed.operation_receipts.len(), receipt_count);
        if expected_kind == BackupOperationKind::Stop {
            prove_pending_stop_observes_then_executes(
                &root,
                &expected_operation,
                receipt_count,
                expected_command.expect("stop command"),
            );
        } else if expected_command.is_some() {
            prove_unknown_external_operation_halts(&root, &expected_operation, &observed);
        } else {
            prove_replay_safe_operation_resumes(&root, &expected_operation, receipt_count);
        }
    }

    fs::remove_dir_all(root).expect("remove pending claim layout");
    fs::remove_dir_all(handshake_root).expect("remove pending claim handshake root");
}

fn prove_pending_stop_observes_then_executes(
    root: &Path,
    expected_operation: &BackupExecutionJournalOperation,
    receipt_count: usize,
    stop_command: String,
) {
    let target = expected_operation
        .target_canister_id
        .as_deref()
        .expect("stop operation target");
    let mut executor = FakeBackupRunnerExecutor::default();
    let response = backup_run_execute_with_executor(
        &runner_config(root.to_path_buf(), Some(1)),
        &mut executor,
    )
    .expect("observe running target and execute pending stop");
    let journal = BackupLayout::new(root.to_path_buf())
        .read_execution_journal()
        .expect("read reconciled stop journal");

    assert_eq!(response.executed_operation_count, 1);
    assert_eq!(
        executor.commands,
        vec![format!("status:{target}"), stop_command]
    );
    assert_operation_completed_once(&journal, expected_operation, receipt_count);
}

fn prepare_operation(
    operation_label: &str,
    prior_steps: usize,
    expected_kind: &BackupOperationKind,
) -> std::path::PathBuf {
    let root = temp_dir(&format!("canic-backup-pending-claim-{operation_label}"));
    let layout = BackupLayout::new(root.clone());
    let plan = super::valid_backup_plan();
    let journal = BackupExecutionJournal::from_plan(&plan).expect("execution journal");
    layout.write_backup_plan(&plan).expect("write backup plan");
    layout
        .write_execution_journal(&journal)
        .expect("write execution journal");

    let mut executor = FakeBackupRunnerExecutor::default();
    let response = backup_run_execute_with_executor(
        &runner_config(root.clone(), Some(prior_steps)),
        &mut executor,
    )
    .expect("prepare pending-claim operation");
    let prepared = layout
        .read_execution_journal()
        .expect("read prepared execution journal");

    assert_eq!(response.executed_operation_count, prior_steps);
    assert_eq!(
        prepared
            .next_ready_operation()
            .map(|operation| &operation.kind),
        Some(expected_kind)
    );
    root
}

fn prove_ready_operation_resumes(
    root: &Path,
    expected_operation: &BackupExecutionJournalOperation,
    receipt_count: usize,
    expected_command: Option<String>,
) {
    let mut executor = FakeBackupRunnerExecutor::default();
    let response = backup_run_execute_with_executor(
        &runner_config(root.to_path_buf(), Some(1)),
        &mut executor,
    )
    .expect("resume operation whose claim did not persist");
    let journal = BackupLayout::new(root.to_path_buf())
        .read_execution_journal()
        .expect("read resumed execution journal");

    assert_eq!(response.executed_operation_count, 1);
    assert_operation_completed_once(&journal, expected_operation, receipt_count);
    assert_eq!(
        executor.commands,
        expected_command.into_iter().collect::<Vec<_>>()
    );
}

fn prove_unknown_external_operation_halts(
    root: &Path,
    expected_operation: &BackupExecutionJournalOperation,
    observed: &BackupExecutionJournal,
) {
    let mut executor = FakeBackupRunnerExecutor::default();
    let error = backup_run_execute_with_executor(
        &runner_config(root.to_path_buf(), Some(1)),
        &mut executor,
    )
    .expect_err("pending external mutation must halt with unknown outcome");
    let persisted = BackupLayout::new(root.to_path_buf())
        .read_execution_journal()
        .expect("read halted execution journal");

    std::assert_matches!(
        error,
        BackupRunnerError::CommandOutcomeUnknown {
            sequence,
            operation_id,
            ..
        } if sequence == expected_operation.sequence
            && operation_id == expected_operation.operation_id
    );
    assert!(executor.commands.is_empty());
    assert_eq!(&persisted, observed);
}

fn prove_replay_safe_operation_resumes(
    root: &Path,
    expected_operation: &BackupExecutionJournalOperation,
    receipt_count: usize,
) {
    let mut executor = FakeBackupRunnerExecutor::default();
    let response = backup_run_execute_with_executor(
        &runner_config(root.to_path_buf(), Some(1)),
        &mut executor,
    )
    .expect("resume replay-safe pending operation");
    let journal = BackupLayout::new(root.to_path_buf())
        .read_execution_journal()
        .expect("read replay-safe execution journal");

    assert_eq!(response.executed_operation_count, 1);
    assert!(executor.commands.is_empty());
    assert_operation_completed_once(&journal, expected_operation, receipt_count);
}

fn assert_operation_completed_once(
    journal: &BackupExecutionJournal,
    expected_operation: &BackupExecutionJournalOperation,
    receipt_count: usize,
) {
    assert_eq!(
        operation_at(journal, expected_operation.sequence).state,
        BackupExecutionOperationState::Completed
    );
    assert_eq!(journal.operation_receipts.len(), receipt_count + 1);
    assert_eq!(
        journal
            .operation_receipts
            .iter()
            .filter(|receipt| receipt.sequence == expected_operation.sequence)
            .count(),
        1
    );
}

fn operation_at(
    journal: &BackupExecutionJournal,
    sequence: usize,
) -> &BackupExecutionJournalOperation {
    journal
        .operations
        .iter()
        .find(|operation| operation.sequence == sequence)
        .expect("operation by sequence")
}

fn expected_command(
    journal: &BackupExecutionJournal,
    operation: &BackupExecutionJournalOperation,
) -> Option<String> {
    let target = operation.target_canister_id.as_deref()?;
    match operation.kind {
        BackupOperationKind::Stop => Some(format!("stop:{target}")),
        BackupOperationKind::CreateSnapshot => Some(format!("snapshot:{target}")),
        BackupOperationKind::Start => Some(format!("start:{target}")),
        BackupOperationKind::DownloadSnapshot => {
            let snapshot_id = journal
                .operation_receipts
                .iter()
                .rev()
                .find_map(|receipt| receipt.snapshot_id.as_deref())
                .expect("snapshot receipt before download");
            Some(format!("download:{target}:{snapshot_id}"))
        }
        BackupOperationKind::VerifyArtifact | BackupOperationKind::FinalizeManifest => None,
        BackupOperationKind::ValidateTopology
        | BackupOperationKind::ValidateControlAuthority
        | BackupOperationKind::ValidateSnapshotReadAuthority
        | BackupOperationKind::ValidateQuiescencePolicy => {
            panic!("preflight operation is outside B04")
        }
    }
}

fn runner_config(out: std::path::PathBuf, max_steps: Option<usize>) -> BackupRunnerConfig {
    BackupRunnerConfig {
        out,
        max_steps,
        updated_at: Some("unix:10".to_string()),
        tool_name: "canic".to_string(),
        tool_version: "test".to_string(),
    }
}