use super::*;
use crate::ops::persistence::AttemptJournalGuard;
use crate::tests::support::execution_workflow::assert_pending;
use crate::tests::support::execution_workflow::{prepare_single_operation, workflow};
use crate::tests::support::provider::Provider;
use crate::tests::support::ready::ready;
use crate::workflow::ic_snapshot_capture::IcSnapshotCaptureReplyError;
use crate::{
model::{
artifacts::ArtifactChecksumRecord,
attempt_journal::{AttemptJournalRecord, MutationOutcomeRecord, MutationReceiptRequest},
execution_workflow::ExecutionStageBindingRecord,
ic_request::IcManagementRequest,
operation_plan::OperationPlanRecord,
},
ops::persistence::BackupLayoutGuard,
tests::support::ic_mutation::reply,
};
use std::{convert::Infallible, fs, path::PathBuf};
fn payload(method: IcManagementMethodRecord) -> IcManagementRequestRecord {
IcManagementRequestRecord::new(IcManagementRequest {
method,
target: workflow().stage(0).unwrap().target().into(),
snapshot_id: (method == IcManagementMethodRecord::LoadCanisterSnapshot)
.then(|| vec![0, 255, 128]),
})
.unwrap()
}
fn prepare(
payload: &IcManagementRequestRecord,
) -> (
PathBuf,
BackupLayoutGuard,
ExecutionStageBindingRecord,
OperationPlanRecord,
) {
prepare_single_operation(
"ic-backup-capture-step",
payload.target(),
payload.digest().hash(),
)
}
fn admit(_: &IcMutationRequest<'_>) -> impl std::future::Future<Output = Result<(), Infallible>> {
std::future::ready(Ok(()))
}
fn provider(reply: Result<Vec<u8>, IcMutationProviderError>) -> Provider<IcMutationProviderError> {
let mut provider = Provider::new(reply);
provider.method = Some(IcManagementMethodRecord::TakeCanisterSnapshot);
provider.evidence = Some(ArtifactChecksumRecord::from_bytes(
b"passive fixture association",
));
provider
}
#[test]
fn one_durably_reserved_capture_holds_exclusion_and_leaves_receipt_qualification_explicit() {
let payload = payload(IcManagementMethodRecord::TakeCanisterSnapshot);
let raw = reply(payload.method());
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 acknowledgement = ready(capture_snapshot(
&stage,
42,
&payload,
&mut provider,
async |request| {
let retained: 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_eq!(request.payload().digest(), payload.digest());
assert!(
AttemptJournalGuard::open(stage.layout().unwrap(), request.authority()).is_err()
);
Ok::<(), Infallible>(())
},
))
.unwrap();
assert_eq!(acknowledgement.input().reply, raw);
assert_pending(&stage, 42);
assert!(
ready(capture_snapshot(
&stage,
42,
&payload,
&mut provider,
async |_| -> Result<(), Infallible> {
panic!("pending original cannot re-enter admission")
}
))
.is_err()
);
let mut journal = AttemptJournalGuard::open(
stage.layout().unwrap(),
&plan.attempt_authority(42).unwrap(),
)
.unwrap();
let request = IcMutationRequest::new(&plan, 42, journal.record().unwrap(), &payload).unwrap();
let view =
validate_acknowledgement(&request, journal.record().unwrap(), &acknowledgement).unwrap();
let crate::policy::ic_mutation::IcMutationReplyView::Capture(reply) = view.reply() else {
panic!("capture wire")
};
assert_eq!(reply.snapshots()[0].id(), &[0, 255, 128]);
journal
.record_mutation(MutationReceiptRequest {
attempt: 1,
request: payload.digest(),
outcome: MutationOutcomeRecord::Applied,
evidence: acknowledgement.input().evidence.clone(),
})
.unwrap();
drop(journal);
assert!(ready(capture_snapshot(&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 every_other_management_method_rejects_before_spending_callback_or_provider() {
for method in [
IcManagementMethodRecord::StopCanister,
IcManagementMethodRecord::StartCanister,
IcManagementMethodRecord::LoadCanisterSnapshot,
IcManagementMethodRecord::CanisterStatus,
IcManagementMethodRecord::ListCanisterSnapshots,
] {
let payload = payload(method);
let (root, layout, binding, _) = 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(reply(method)));
assert!(matches!(
ready(capture_snapshot(
&stage,
42,
&payload,
&mut provider,
async |_| -> Result<(), Infallible> {
panic!("wrong method never reaches admission")
}
)),
Err(IcSnapshotCaptureExecutionError::CaptureRequired)
));
assert_eq!(fs::read(&path).unwrap(), original);
assert_eq!(provider.calls, 0);
drop(stage);
drop(layout);
fs::remove_dir_all(root).unwrap();
}
}
#[test]
fn changed_payload_and_unknown_operation_refuse_without_consumption() {
let payload = payload(IcManagementMethodRecord::TakeCanisterSnapshot);
let (root, layout, binding, _) = 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(reply(payload.method())));
let other = IcManagementRequestRecord::new(IcManagementRequest {
method: payload.method(),
target: "aaaaa-aa".into(),
snapshot_id: None,
})
.unwrap();
assert!(matches!(
ready(capture_snapshot(&stage, 42, &other, &mut provider, admit)),
Err(IcSnapshotCaptureExecutionError::Request(_))
));
assert!(matches!(
ready(capture_snapshot(&stage, 99, &payload, &mut provider, admit)),
Err(IcSnapshotCaptureExecutionError::Plan(_))
));
assert_eq!(fs::read(&path).unwrap(), original);
assert_eq!(provider.calls, 0);
drop(stage);
drop(layout);
fs::remove_dir_all(root).unwrap();
}
#[test]
fn fresh_consistency_rejection_consumes_original_reservation_without_dispatch_or_recapture() {
let payload = payload(IcManagementMethodRecord::TakeCanisterSnapshot);
let (root, layout, binding, _) = prepare(&payload);
let stage =
ExecutionStageGuard::open(&layout, binding.workflow(), 0, &binding.digest()).unwrap();
let mut provider = provider(Ok(reply(payload.method())));
assert!(matches!(
ready(capture_snapshot(
&stage,
42,
&payload,
&mut provider,
async |_| Err(std::io::Error::other("fixture drain refused"))
)),
Err(IcSnapshotCaptureExecutionError::Admission(_))
));
assert_pending(&stage, 42);
assert!(ready(capture_snapshot(&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 provider_errors_keep_their_exact_capture_projection_and_pending_attempt() {
for failure in [
IcMutationProviderError::Unavailable,
IcMutationProviderError::Unsupported,
IcMutationProviderError::Indeterminate,
] {
let payload = payload(IcManagementMethodRecord::TakeCanisterSnapshot);
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(capture_snapshot(&stage, 42, &payload, &mut provider, admit)),
Err(IcSnapshotCaptureExecutionError::Provider(actual)) if actual == failure
));
assert_pending(&stage, 42);
assert_eq!(provider.calls, 1);
}
}
#[test]
fn malformed_and_wrong_target_acknowledgements_retain_original_bytes_without_receipts() {
for malformed in [true, false] {
let payload = payload(IcManagementMethodRecord::TakeCanisterSnapshot);
let (root, layout, binding, _) = prepare(&payload);
let stage =
ExecutionStageGuard::open(&layout, binding.workflow(), 0, &binding.digest()).unwrap();
let raw = if malformed {
b"bounded malformed reply".to_vec()
} else {
reply(payload.method())
};
let mut provider = provider(Ok(raw.clone()));
if !malformed {
provider.target = Some("aaaaa-aa");
}
let error =
ready(capture_snapshot(&stage, 42, &payload, &mut provider, admit)).unwrap_err();
let IcSnapshotCaptureExecutionError::AfterReply {
source: IcSnapshotCaptureReplyError::Association(_),
acknowledgement,
} = error
else {
panic!("passive rejection retains reply")
};
assert_eq!(acknowledgement.input().reply, raw);
assert_eq!(provider.calls, 1);
assert_pending(&stage, 42);
drop(stage);
drop(layout);
fs::remove_dir_all(root).unwrap();
}
}
#[derive(Debug, Eq, PartialEq)]
enum Drift {
Admission,
StageAfterReply,
LayoutAfterReply,
}
#[test]
fn drift_at_admission_stops_dispatch_and_post_reply_stage_or_layout_drift_retains_acknowledgement()
{
for boundary in [
Drift::Admission,
Drift::StageAfterReply,
Drift::LayoutAfterReply,
] {
let payload = payload(IcManagementMethodRecord::TakeCanisterSnapshot);
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 path = stage_root.join("stage-binding.json");
let raw = reply(payload.method());
let mut provider = provider(Ok(raw.clone()));
match boundary {
Drift::StageAfterReply => provider.change_stage = Some(path.clone()),
Drift::LayoutAfterReply => provider.replace_layout = Some(stage_root.clone()),
Drift::Admission => {}
}
let error = ready(capture_snapshot(
&stage,
42,
&payload,
&mut provider,
async |_| {
if boundary == Drift::Admission {
fs::write(&path, b"{}").unwrap();
}
Ok::<(), Infallible>(())
},
))
.unwrap_err();
match error {
IcSnapshotCaptureExecutionError::Stage(_) if boundary == Drift::Admission => {}
IcSnapshotCaptureExecutionError::AfterReply {
source: IcSnapshotCaptureReplyError::Stage(_),
acknowledgement,
} if boundary == Drift::StageAfterReply => {
assert_eq!(acknowledgement.input().reply, raw);
}
IcSnapshotCaptureExecutionError::AfterReply {
source: IcSnapshotCaptureReplyError::Journal(_),
acknowledgement,
} if boundary == Drift::LayoutAfterReply => {
assert_eq!(acknowledgement.input().reply, raw);
}
other => panic!("unexpected boundary failure: {other:?}"),
}
assert_eq!(provider.calls, usize::from(boundary != Drift::Admission));
let retained = if boundary == Drift::LayoutAfterReply {
stage_root.with_extension("retained-original")
} else {
stage_root
};
let journal: AttemptJournalRecord =
serde_json::from_slice(&fs::read(retained.join("attempt-42.json")).unwrap()).unwrap();
assert_eq!(journal.view().pending_mutation, Some(1));
assert_eq!(journal.view().mutations_used, 1);
assert!(!journal.view().applied);
drop(stage);
drop(layout);
fs::remove_dir_all(root).unwrap();
}
}