ic-backup 0.60.0

//! Spending, exclusive dispatch and retained failures through the public step.

use super::*;
use crate::policy::ResponseIdentityMismatch;
use crate::tests::support::execution_workflow::assert_pending;
use crate::tests::support::execution_workflow::{prepare_single_operation, workflow};
use crate::tests::support::provider::{Provider, SuspendedProvider};
use crate::tests::support::ready::ready;
use crate::workflow::ic_snapshot_transfer_read::IcSnapshotTransferReadReplyError;
use crate::{
    model::{
        artifacts::ArtifactChecksumRecord,
        attempt_journal::{MutationOutcomeRecord, MutationReceiptRequest},
        execution_workflow::ExecutionStageBindingRecord,
        ic_snapshot_data::IcSnapshotDataRequest,
        ic_snapshot_metadata::{IcSnapshotMetadataReply, IcSnapshotMetadataRequest},
        operation_plan::OperationPlanRecord,
    },
    ops::persistence::BackupLayoutGuard,
};
use ic_platform::candid::contract;
use ic_platform::management_canister_types::{ReadCanisterSnapshotDataResult, SnapshotDataKind};
use std::{convert::Infallible, fs, path::PathBuf};

fn metadata_wire() -> Vec<u8> {
    crate::tests::support::ic_snapshot_metadata::raw("data-source")
}
fn metadata_request() -> IcSnapshotMetadataRequest {
    IcSnapshotMetadataRequest::new(workflow().stage(0).unwrap().target(), &[0, 255, 128]).unwrap()
}
fn prepare(
    payload: IcSnapshotTransferReadPayload<'_, '_>,
) -> (
    PathBuf,
    BackupLayoutGuard,
    ExecutionStageBindingRecord,
    OperationPlanRecord,
) {
    prepare_single_operation(
        "ic-backup-transfer-step",
        payload.target(),
        payload.digest().hash(),
    )
}
fn admit(
    _: &IcSnapshotTransferReadRequest<'_, '_>,
) -> impl std::future::Future<Output = Result<(), Infallible>> {
    std::future::ready(Ok(()))
}
fn provider(
    reply: Result<Vec<u8>, IcObservationProviderError>,
) -> Provider<IcObservationProviderError> {
    let mut provider = Provider::new(reply);
    provider.evidence = Some(ArtifactChecksumRecord::from_bytes(
        b"fixture association only",
    ));
    provider
}

#[test]
fn cancellation_retains_exact_metadata_or_data_reservation_and_denies_reentry() {
    let original = metadata_request();
    let wire = metadata_wire();
    let metadata = IcSnapshotMetadataReply::decode(&original, &wire).unwrap();
    let data = IcSnapshotDataRequest::new(
        &metadata,
        SnapshotDataKind::WasmModule { offset: 0, size: 3 },
    )
    .unwrap();
    for payload in [
        IcSnapshotTransferReadPayload::Metadata(&original),
        IcSnapshotTransferReadPayload::Data(&data),
    ] {
        for suspend_admission in [true, false] {
            let (_root, layout, binding, _plan) = prepare(payload);
            let stage =
                ExecutionStageGuard::open(&layout, binding.workflow(), 0, &binding.digest())
                    .unwrap();
            let mut provider = SuspendedProvider { calls: 0 };
            let future = dispatch_pending(&stage, 42, payload, &mut provider, async |_| {
                if suspend_admission {
                    std::future::pending::<()>().await;
                }
                Ok::<(), Infallible>(())
            });
            let before =
                crate::tests::support::execution_workflow::cancel_pending(&stage, 42, future);
            assert_eq!(provider.calls, usize::from(!suspend_admission));
            assert!(ready(dispatch_pending(&stage, 42, payload, &mut provider, admit)).is_err());
            assert_eq!(provider.calls, usize::from(!suspend_admission));
            assert_eq!(
                fs::read(stage.layout().unwrap().root().join("attempt-42.json")).unwrap(),
                before
            );
        }
    }
}

#[test]
fn one_locked_durable_read_leaves_pending_until_explicit_integration_receipt() {
    let metadata_request = metadata_request();
    let wire = metadata_wire();
    let metadata = IcSnapshotMetadataReply::decode(&metadata_request, &wire).unwrap();
    let data = IcSnapshotDataRequest::new(
        &metadata,
        SnapshotDataKind::WasmModule { offset: 0, size: 3 },
    )
    .unwrap();
    for (payload, raw) in [
        (
            IcSnapshotTransferReadPayload::Metadata(&metadata_request),
            wire.clone(),
        ),
        (
            IcSnapshotTransferReadPayload::Data(&data),
            contract::encode_one(ReadCanisterSnapshotDataResult {
                chunk: vec![7, 8, 9],
            })
            .unwrap(),
        ),
    ] {
        let (root, layout, binding, plan) = prepare(payload);
        let stage =
            ExecutionStageGuard::open(&layout, binding.workflow(), 0, &binding.digest()).unwrap();
        let mut provider = provider(Ok(raw.clone()));
        let response = ready(dispatch_pending(
            &stage,
            42,
            payload,
            &mut provider,
            async |request| {
                let retained: crate::model::attempt_journal::AttemptJournalRecord =
                    serde_json::from_slice(
                        &fs::read(stage.layout().unwrap().root().join("attempt-42.json")).unwrap(),
                    )
                    .unwrap();
                assert_eq!(
                    retained.view().pending_mutation,
                    Some(request.mutation_attempt())
                );
                assert!(
                    AttemptJournalGuard::open(stage.layout().unwrap(), request.authority())
                        .is_err()
                );
                Ok::<(), Infallible>(())
            },
        ))
        .unwrap();
        assert_eq!(response.input().reply, raw);
        assert_eq!(provider.calls, 1);
        assert_pending(&stage, 42);
        assert!(
            ready(dispatch_pending(
                &stage,
                42,
                payload,
                &mut provider,
                async |_| -> Result<(), Infallible> {
                    panic!("pending attempt must reject before fresh admission")
                }
            ))
            .is_err()
        );
        assert_eq!(provider.calls, 1);
        let mut journal = AttemptJournalGuard::open(
            stage.layout().unwrap(),
            &plan.attempt_authority(42).unwrap(),
        )
        .unwrap();
        // Explicit fixture-owned qualification; the workflow never creates this receipt.
        journal
            .record_mutation(MutationReceiptRequest {
                attempt: 1,
                request: payload.digest(),
                outcome: MutationOutcomeRecord::Applied,
                evidence: response.input().evidence.clone(),
            })
            .unwrap();
        drop(journal);
        assert!(ready(dispatch_pending(&stage, 42, payload, &mut provider, admit)).is_err());
        assert_eq!(provider.calls, 1);
        stage
            .reconcile_checkpoint(ArtifactChecksumRecord::from_bytes(&raw))
            .unwrap();
        drop(stage);
        drop(layout);
        fs::remove_dir_all(root).unwrap();
    }
}

#[test]
fn wrong_payload_unknown_operation_missing_and_locked_originals_do_not_spend_or_dispatch() {
    let request = metadata_request();
    let payload = IcSnapshotTransferReadPayload::Metadata(&request);
    let (root, layout, binding, plan) = prepare(payload);
    let stage =
        ExecutionStageGuard::open(&layout, binding.workflow(), 0, &binding.digest()).unwrap();
    let path = stage.layout().unwrap().root().join("attempt-42.json");
    let original = fs::read(&path).unwrap();
    let mut provider = provider(Ok(metadata_wire()));
    let other = IcSnapshotMetadataRequest::new(request.target(), &[1]).unwrap();
    assert!(matches!(
        ready(dispatch_pending(
            &stage,
            42,
            IcSnapshotTransferReadPayload::Metadata(&other),
            &mut provider,
            admit
        )),
        Err(IcSnapshotTransferReadExecutionError::Request(
            IcSnapshotTransferReadError::PayloadMismatch
        ))
    ));
    assert!(matches!(
        ready(dispatch_pending(&stage, 99, payload, &mut provider, admit)),
        Err(IcSnapshotTransferReadExecutionError::Plan(_))
    ));
    let held = AttemptJournalGuard::open(
        stage.layout().unwrap(),
        &plan.attempt_authority(42).unwrap(),
    )
    .unwrap();
    assert!(matches!(
        ready(dispatch_pending(&stage, 42, payload, &mut provider, admit)),
        Err(IcSnapshotTransferReadExecutionError::Journal(_))
    ));
    drop(held);
    assert_eq!(fs::read(&path).unwrap(), original);
    fs::remove_file(&path).unwrap();
    assert!(matches!(
        ready(dispatch_pending(&stage, 42, payload, &mut provider, admit)),
        Err(IcSnapshotTransferReadExecutionError::Journal(_))
    ));
    assert!(!path.exists());
    assert_eq!(provider.calls, 0);
    drop(stage);
    drop(layout);
    fs::remove_dir_all(root).unwrap();
}

#[test]
fn fresh_admission_rejection_retains_spending_without_provider_or_redispatch() {
    let request = metadata_request();
    let payload = IcSnapshotTransferReadPayload::Metadata(&request);
    let (root, layout, binding, _) = prepare(payload);
    let stage =
        ExecutionStageGuard::open(&layout, binding.workflow(), 0, &binding.digest()).unwrap();
    let mut provider = provider(Ok(metadata_wire()));
    assert!(matches!(
        ready(dispatch_pending(
            &stage,
            42,
            payload,
            &mut provider,
            async |_| Err(std::io::Error::other("fixture access refused"))
        )),
        Err(IcSnapshotTransferReadExecutionError::Admission(_))
    ));
    assert_pending(&stage, 42);
    assert!(ready(dispatch_pending(&stage, 42, payload, &mut provider, admit)).is_err());
    assert_eq!(provider.calls, 0);
    drop(stage);
    drop(layout);
    fs::remove_dir_all(root).unwrap();
}

#[test]
fn each_provider_failure_retains_pending_original_across_resume_without_reissue() {
    for failure in [
        IcObservationProviderError::Unavailable,
        IcObservationProviderError::Unsupported,
        IcObservationProviderError::Indeterminate,
    ] {
        let request = metadata_request();
        let payload = IcSnapshotTransferReadPayload::Metadata(&request);
        let (root, layout, binding, _) = prepare(payload);
        let stage =
            ExecutionStageGuard::open(&layout, binding.workflow(), 0, &binding.digest()).unwrap();
        let mut provider = provider(Err(failure));
        assert!(
            matches!(ready(dispatch_pending(&stage, 42, payload, &mut provider, admit)), Err(IcSnapshotTransferReadExecutionError::Provider(actual)) if actual == failure)
        );
        assert_pending(&stage, 42);
        let path = stage.layout().unwrap().root().join("attempt-42.json");
        let original = fs::read(&path).unwrap();
        drop(stage);
        let (stage, progress) =
            ExecutionStageGuard::resume(&layout, binding.workflow(), 0, &binding.digest()).unwrap();
        assert_eq!(progress.attempts.mutations_used, 1);
        assert!(ready(dispatch_pending(&stage, 42, payload, &mut provider, admit)).is_err());
        assert_eq!(fs::read(&path).unwrap(), original);
        assert_eq!(provider.calls, 1);
        drop(stage);
        drop(layout);
        fs::remove_dir_all(root).unwrap();
    }
}

#[test]
fn malformed_reply_retains_returned_bytes_and_original_pending_reservation() {
    let request = metadata_request();
    let payload = IcSnapshotTransferReadPayload::Metadata(&request);
    let (root, layout, binding, _) = prepare(payload);
    let stage =
        ExecutionStageGuard::open(&layout, binding.workflow(), 0, &binding.digest()).unwrap();
    let raw = b"invalid bounded candid".to_vec();
    let mut provider = provider(Ok(raw.clone()));
    let error = ready(dispatch_pending(&stage, 42, payload, &mut provider, admit)).unwrap_err();
    let IcSnapshotTransferReadExecutionError::AfterReply {
        source: IcSnapshotTransferReadReplyError::Association(_),
        response,
    } = error
    else {
        panic!("retained association failure")
    };
    assert_eq!(response.input().reply, raw);
    assert_eq!(provider.calls, 1);
    assert_pending(&stage, 42);
    drop(stage);
    drop(layout);
    fs::remove_dir_all(root).unwrap();
}

#[test]
fn changed_stage_at_admission_stops_dispatch_and_changed_stage_after_reply_retains_bytes() {
    for after_reply in [false, true] {
        let request = metadata_request();
        let payload = IcSnapshotTransferReadPayload::Metadata(&request);
        let (root, layout, binding, _) = prepare(payload);
        let stage =
            ExecutionStageGuard::open(&layout, binding.workflow(), 0, &binding.digest()).unwrap();
        let path = stage.layout().unwrap().root().join("stage-binding.json");
        let original = fs::read(&path).unwrap();
        let raw = metadata_wire();
        let mut provider = provider(Ok(raw.clone()));
        if after_reply {
            provider.change_stage = Some(path.clone());
        }
        let error = ready(dispatch_pending(
            &stage,
            42,
            payload,
            &mut provider,
            async |_| {
                if !after_reply {
                    fs::write(&path, b"{}").unwrap();
                }
                Ok::<(), Infallible>(())
            },
        ))
        .unwrap_err();
        if after_reply {
            let IcSnapshotTransferReadExecutionError::AfterReply {
                source: IcSnapshotTransferReadReplyError::Stage(_),
                response,
            } = error
            else {
                panic!("retained reply after stage drift")
            };
            assert_eq!(response.input().reply, raw);
        } else {
            assert!(matches!(
                error,
                IcSnapshotTransferReadExecutionError::Stage(_)
            ));
        }
        assert_eq!(provider.calls, usize::from(after_reply));
        fs::write(&path, original).unwrap(); // Restore only this fixture's deliberately changed declaration.
        assert_pending(&stage, 42);
        drop(stage);
        drop(layout);
        fs::remove_dir_all(root).unwrap();
    }
}

#[test]
fn wrong_claim_and_replaced_layout_retain_reply_without_outcome_or_repair() {
    for replaced in [false, true] {
        let request = metadata_request();
        let payload = IcSnapshotTransferReadPayload::Metadata(&request);
        let (root, layout, binding, _) = prepare(payload);
        let stage =
            ExecutionStageGuard::open(&layout, binding.workflow(), 0, &binding.digest()).unwrap();
        let stage_root = stage.layout().unwrap().root().to_path_buf();
        let raw = metadata_wire();
        let mut provider = provider(Ok(raw.clone()));
        if replaced {
            provider.replace_layout = Some(stage_root.clone());
        } else {
            provider.target = Some("aaaaa-aa");
        }
        let error = ready(dispatch_pending(&stage, 42, payload, &mut provider, admit)).unwrap_err();
        let response = match error {
            IcSnapshotTransferReadExecutionError::AfterReply {
                source: IcSnapshotTransferReadReplyError::Journal(_),
                response,
            } if replaced => response,
            IcSnapshotTransferReadExecutionError::AfterReply {
                source:
                    IcSnapshotTransferReadReplyError::Association(
                        IcSnapshotTransferReadAssociationError::Identity(
                            ResponseIdentityMismatch::Target,
                        ),
                    ),
                response,
            } if !replaced => response,
            other => panic!("unexpected retained rejection: {other:?}"),
        };
        assert_eq!(response.input().reply, raw);
        let retained = if replaced {
            stage_root.with_extension("retained-original")
        } else {
            stage_root
        };
        let journal: crate::model::attempt_journal::AttemptJournalRecord =
            serde_json::from_slice(&fs::read(retained.join("attempt-42.json")).unwrap()).unwrap();
        assert_eq!(journal.view().pending_mutation, Some(1));
        assert!(!journal.view().applied);
        assert_eq!(provider.calls, 1);
        drop(stage);
        drop(layout);
        fs::remove_dir_all(root).unwrap();
    }
}

// Exercise shared dispatch without a receipt; production stage coordinators supply qualification.
async fn dispatch_pending<E: std::error::Error + 'static>(
    stage: &ExecutionStageGuard<'_>,
    operation_sequence: u64,
    payload: IcSnapshotTransferReadPayload<'_, '_>,
    provider: &mut impl IcSnapshotTransferReadProvider,
    admit: impl AsyncFnOnce(&IcSnapshotTransferReadRequest<'_, '_>) -> Result<(), E>,
) -> Result<IcSnapshotTransferReadResponse, IcSnapshotTransferReadExecutionError<E>> {
    let (response, ()) = read_snapshot_with(
        stage,
        operation_sequence,
        payload,
        provider,
        admit,
        async |_, _, _| (),
    )
    .await?;
    Ok(response)
}