use super::*;
use crate::model::attempt_journal::AttemptBudgetRecord;
use crate::model::ic_snapshot_upload::IcSnapshotUploadAttemptError;
use crate::model::operation_plan::{OperationPlanRecord, PlanBudgetRecord, PlannedOperationRecord};
use crate::policy::{
ResponseIdentityMismatch, ic_snapshot_upload::IcSnapshotUploadAssociationError,
};
use crate::ports::IcProviderError;
use crate::tests::support::operation_plan::{declaration, operation_request, set_dependencies};
use crate::tests::support::provider::Provider;
use crate::tests::support::{
ready::ready, response_fault::ContextFault, retained_reply::write_synced,
};
use crate::workflow::ic_snapshot_upload::IcSnapshotUploadReplyError;
use crate::{
model::{
artifacts::ArtifactChecksumRecord,
execution_workflow::{ExecutionStageBindingRecord, ExecutionWorkflowRecord},
ic_snapshot_upload::IcSnapshotUploadReply,
},
ops::persistence::{
AttemptJournalGuard, BackupLayoutGuard, create_execution_workflow, read_execution_progress,
},
tests::support::{
ic_snapshot_upload::{DESTINATION_ID, unhex, upload_plan},
temp_dir,
},
};
use ic_platform::management_canister_types::SnapshotDataKind;
use std::{convert::Infallible, fs, io};
fn with_metadata(check: impl FnOnce(&IcSnapshotUploadRequest<'_>)) {
crate::tests::support::ic_snapshot_upload::with_metadata_upload(
&ArtifactChecksumRecord::from_bytes(b"retained source tree"),
check,
);
}
fn provider() -> Provider {
Provider::new(Ok(unhex("4449444c026c01b6b897890f016d7b0100031500ff")))
}
fn with_allocation_stage(
check: impl FnOnce(
&IcSnapshotUploadRequest<'_>,
&crate::tests::support::execution_workflow::PreparedStage,
ExecutionStageGuard<'_>,
Provider,
),
) {
with_metadata(|payload| {
let plan = upload_plan(payload);
let prepared = crate::tests::support::execution_workflow::prepare_stage(
"ic-backup-allocation-stage",
&ExecutionWorkflowRecord::new(plan.clone()),
7,
plan,
);
check(payload, &prepared, prepared.open(), provider());
});
}
fn admit(
_: &IcSnapshotUploadAttempt<'_, '_>,
) -> impl std::future::Future<Output = Result<(), Infallible>> {
std::future::ready(Ok(()))
}
fn receipt(
request: &IcSnapshotUploadAttempt<'_, '_>,
acknowledgement: &IcMutationAcknowledgement,
) -> MutationReceiptRequest {
crate::tests::support::update::applied_receipt(
request.mutation_attempt(),
&request.payload().binding_digest(),
&acknowledgement.input().evidence,
)
}
fn pending(stage: &ExecutionStageGuard<'_>) {
crate::tests::support::execution_workflow::assert_pending(stage, 7);
assert!(
stage
.reconcile_checkpoint(ArtifactChecksumRecord::from_bytes(b"not qualified"))
.is_err()
);
}
#[test]
fn qualified_retained_reply_checkpoints_exact_allocation_without_reissue_on_reopen() {
with_allocation_stage(|payload, prepared, stage, mut provider| {
let (acknowledgement, learned) = ready(allocate_snapshot(
&stage,
7,
payload,
&mut provider,
admit,
async |request, acknowledgement| {
assert!(
AttemptJournalGuard::open(stage.layout().unwrap(), request.authority())
.is_err()
);
let _file = write_synced(
&stage
.layout()
.unwrap()
.root()
.join("retained-allocation.candid"),
&acknowledgement.input().reply,
)
.unwrap();
Ok::<_, Infallible>(receipt(request, acknowledgement))
},
))
.unwrap();
let predecessor = stage.reconcile_checkpoint(learned).unwrap();
assert_eq!(
predecessor.learned_evidence(),
&IcSnapshotUploadReply::decode(payload, &acknowledgement.input().reply)
.unwrap()
.digest()
);
assert_eq!(provider.calls, 1);
drop(stage);
let (reopened, view) = ExecutionStageGuard::resume(
&prepared.layout,
prepared.binding.workflow(),
7,
&prepared.binding.digest(),
)
.unwrap();
assert_eq!(view.applied_operations, 1);
let settlement = crate::ops::persistence::read_execution_settlement(
reopened.layout().unwrap(),
&reopened.plan().digest(),
predecessor.settlement(),
)
.unwrap();
assert_eq!(&settlement.digest(), predecessor.settlement());
assert!(
ready(allocate_snapshot(
&reopened,
7,
payload,
&mut provider,
admit,
async |_, _| -> Result<_, Infallible> { panic!("Applied never reissues") }
))
.is_err()
);
assert_eq!(provider.calls, 1);
});
}
#[test]
fn data_sequence_and_multi_operation_stages_refuse_before_spending_or_provider() {
with_metadata(|payload| {
let data = IcSnapshotUploadRequest::data(
payload,
DESTINATION_ID,
SnapshotDataKind::WasmModule { offset: 0, size: 3 },
&[7, 8, 9],
)
.unwrap();
let plan = upload_plan(payload);
let prepared = crate::tests::support::execution_workflow::prepare_stage(
"ic-backup-allocation-stage",
&ExecutionWorkflowRecord::new(plan.clone()),
7,
plan,
);
let stage = prepared.open();
let mut provider = provider();
for (sequence, request) in [(6, payload), (7, &data)] {
assert!(matches!(
ready(allocate_snapshot(
&stage,
sequence,
request,
&mut provider,
admit,
async |_, _| -> Result<_, Infallible> { panic!("not allocation") }
)),
Err(IcSnapshotAllocationExecutionError::OriginalMismatch)
));
}
assert_eq!(
read_execution_progress(stage.layout().unwrap(), &stage.plan().digest())
.unwrap()
.attempts
.mutations_used,
0
);
let root = temp_dir("ic-backup-multi-allocation");
let multi_layout = BackupLayoutGuard::acquire(&root).unwrap();
let mut value = declaration(&upload_plan(payload));
let mut operation = operation_request(&value.operations[0]);
operation.operation_sequence = 8;
value
.operations
.push(PlannedOperationRecord::new(operation).unwrap());
set_dependencies(&mut value, [(7, vec![]), (8, vec![7])]);
value.budget = PlanBudgetRecord::new(4, 2).unwrap();
let plan = OperationPlanRecord::new(value).unwrap();
let mut allocation = declaration(&upload_plan(payload));
let mut stage = operation_request(&allocation.operations[0]);
stage.budget = AttemptBudgetRecord::new(4, 2).unwrap();
allocation.operations[0] = PlannedOperationRecord::new(stage).unwrap();
allocation.budget = PlanBudgetRecord::new(4, 2).unwrap();
let workflow = ExecutionWorkflowRecord::new(OperationPlanRecord::new(allocation).unwrap());
create_execution_workflow(&multi_layout, &workflow).unwrap();
let binding = ExecutionStageBindingRecord::new(&workflow, 7, &plan, vec![]).unwrap();
let multi = ExecutionStageGuard::prepare(&multi_layout, binding, plan).unwrap();
assert!(matches!(
ready(allocate_snapshot(
&multi,
7,
payload,
&mut provider,
admit,
async |_, _| -> Result<_, Infallible> { panic!("not singleton") }
)),
Err(IcSnapshotAllocationExecutionError::OriginalMismatch)
));
assert_eq!(provider.calls, 0);
});
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum AllocationFailure {
Provider(IcProviderError),
Admission,
TargetMismatch,
MalformedReply,
Unqualified,
}
impl AllocationFailure {
fn assert_error(
self,
error: IcSnapshotAllocationExecutionError<io::Error>,
reply: &Result<Vec<u8>, IcProviderError>,
) {
match (self, error) {
(
AllocationFailure::Unqualified,
IcSnapshotAllocationExecutionError::AfterReply {
source: IcSnapshotAllocationSettlementError::Qualification(_),
acknowledgement,
},
) => assert_eq!(
acknowledgement.input().reply,
reply.as_ref().unwrap().as_slice()
),
(
AllocationFailure::Provider(expected),
IcSnapshotAllocationExecutionError::Upload(
IcSnapshotUploadExecutionError::Provider(actual),
),
) => assert_eq!(actual, expected),
(
AllocationFailure::Admission,
IcSnapshotAllocationExecutionError::Upload(
IcSnapshotUploadExecutionError::Admission(_),
),
) => {}
(
failure,
IcSnapshotAllocationExecutionError::Upload(
IcSnapshotUploadExecutionError::AfterReply {
source: IcSnapshotUploadReplyError::Association(source),
acknowledgement,
},
),
) => {
assert!(matches!(
(failure, source),
(
AllocationFailure::MalformedReply,
IcSnapshotUploadAssociationError::Reply(_)
) | (
AllocationFailure::TargetMismatch,
IcSnapshotUploadAssociationError::Identity(
ResponseIdentityMismatch::Target
)
)
));
assert_eq!(
acknowledgement.input().reply,
reply.as_ref().unwrap().as_slice()
);
}
other => panic!("unexpected allocation refusal: {other:?}"),
}
}
}
#[test]
fn failed_allocation_admission_provider_or_reply_cannot_checkpoint_or_reissue() {
for failure in [
AllocationFailure::Provider(IcProviderError::Unavailable),
AllocationFailure::Provider(IcProviderError::Unsupported),
AllocationFailure::Provider(IcProviderError::Indeterminate),
AllocationFailure::Admission,
AllocationFailure::TargetMismatch,
AllocationFailure::MalformedReply,
AllocationFailure::Unqualified,
] {
with_allocation_stage(|payload, prepared, stage, mut provider| {
if let AllocationFailure::Provider(error) = failure {
provider.reply = Err(error);
}
if failure == AllocationFailure::TargetMismatch {
provider.target = Some("aaaaa-aa");
}
if failure == AllocationFailure::MalformedReply {
provider.reply = Ok(vec![]);
}
let result = ready(allocate_snapshot(
&stage,
7,
payload,
&mut provider,
async |_| {
if failure == AllocationFailure::Admission {
return Err(io::Error::other("fresh allocation admission refused"));
}
Ok::<_, io::Error>(())
},
async |_, _| {
assert_eq!(
failure,
AllocationFailure::Unqualified,
"unadmitted replies never reach qualification"
);
Err(io::Error::other("no authentic durable originals"))
},
));
failure.assert_error(result.unwrap_err(), &provider.reply);
pending(&stage);
drop(stage);
let (reopened, view) = ExecutionStageGuard::resume(
&prepared.layout,
prepared.binding.workflow(),
7,
&prepared.binding.digest(),
)
.unwrap();
assert_eq!(view.attempts.mutations_used, 1);
assert!(
ready(allocate_snapshot(
&reopened,
7,
payload,
&mut provider,
admit,
async |_, _| -> Result<_, Infallible> { panic!("pending never reissues") }
))
.is_err()
);
assert_eq!(
provider.calls,
usize::from(failure != AllocationFailure::Admission)
);
});
}
}
#[test]
fn stage_drift_during_admission_reply_or_qualification_keeps_original_pending() {
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum Boundary {
Admission,
Reply,
Qualification,
}
for boundary in [
Boundary::Admission,
Boundary::Reply,
Boundary::Qualification,
] {
with_allocation_stage(|payload, _, stage, mut provider| {
let path = stage.layout().unwrap().root().join("stage-binding.json");
let raw = provider.reply.clone().unwrap();
if boundary == Boundary::Reply {
provider.change_stage = Some(path.clone());
}
let error = ready(allocate_snapshot(
&stage,
7,
payload,
&mut provider,
async |_| {
if boundary == Boundary::Admission {
fs::write(&path, b"{}").unwrap();
}
Ok::<_, Infallible>(())
},
async |request, acknowledgement| {
assert_eq!(
boundary,
Boundary::Qualification,
"unadmitted reply cannot qualify"
);
fs::write(&path, b"{}").unwrap();
Ok::<_, Infallible>(receipt(request, acknowledgement))
},
))
.unwrap_err();
match (boundary, error) {
(
Boundary::Admission,
IcSnapshotAllocationExecutionError::Upload(
IcSnapshotUploadExecutionError::Reservation(
crate::workflow::ExecutionReservationError::Stage(_),
),
),
) => {}
(
Boundary::Reply,
IcSnapshotAllocationExecutionError::Upload(
IcSnapshotUploadExecutionError::AfterReply {
source: IcSnapshotUploadReplyError::Stage(_),
acknowledgement,
},
),
)
| (
Boundary::Qualification,
IcSnapshotAllocationExecutionError::AfterReply {
source: IcSnapshotAllocationSettlementError::Stage(_),
acknowledgement,
},
) => assert_eq!(acknowledgement.input().reply, raw),
other => panic!("unexpected stage drift refusal: {other:?}"),
}
let journal: crate::model::attempt_journal::AttemptJournalRecord =
serde_json::from_slice(
&fs::read(path.parent().unwrap().join("attempt-7.json")).unwrap(),
)
.unwrap();
assert_eq!(journal.view().pending_mutation, Some(1));
assert!(!journal.view().applied);
assert_eq!(provider.calls, usize::from(boundary != Boundary::Admission));
});
}
}
#[test]
fn occupied_checkpoint_preserves_applied_receipt_reply_and_original_evidence() {
with_allocation_stage(|payload, prepared, stage, mut provider| {
let path = stage
.layout()
.unwrap()
.root()
.join("execution-settlement.json");
let result = ready(allocate_snapshot(
&stage,
7,
payload,
&mut provider,
admit,
async |request, acknowledgement| {
fs::write(&path, b"occupied original evidence").unwrap();
Ok::<_, Infallible>(receipt(request, acknowledgement))
},
));
let (acknowledgement, learned) = result.unwrap();
assert!(stage.reconcile_checkpoint(learned).is_err());
assert_eq!(
acknowledgement.input().reply,
provider.reply.clone().unwrap()
);
assert_eq!(fs::read(&path).unwrap(), b"occupied original evidence");
drop(stage);
let (reopened, view) = ExecutionStageGuard::resume(
&prepared.layout,
prepared.binding.workflow(),
7,
&prepared.binding.digest(),
)
.unwrap();
assert_eq!(view.applied_operations, 1);
assert!(
ready(allocate_snapshot(
&reopened,
7,
payload,
&mut provider,
admit,
async |_, _| -> Result<_, Infallible> { panic!("Applied never reissues") }
))
.is_err()
);
assert_eq!(provider.calls, 1);
assert_eq!(fs::read(&path).unwrap(), b"occupied original evidence");
});
}
#[test]
fn changed_payload_source_network_or_release_reject_before_spending() {
#[derive(Clone, Copy)]
enum PlanFault {
Request,
Context(ContextFault),
}
with_metadata(|payload| {
for fault in [
PlanFault::Request,
PlanFault::Context(ContextFault::Network),
PlanFault::Context(ContextFault::Release),
] {
let mut plan = declaration(&upload_plan(payload));
match fault {
PlanFault::Request => {
let mut operation = operation_request(&plan.operations[0]);
operation.request = "00".repeat(32);
plan.operations[0] = PlannedOperationRecord::new(operation).unwrap();
}
PlanFault::Context(context) => context.apply(&mut plan.context, &"00".repeat(32)),
}
let plan = OperationPlanRecord::new(plan).unwrap();
let prepared = crate::tests::support::execution_workflow::prepare_stage(
"ic-backup-allocation-source-refusal",
&ExecutionWorkflowRecord::new(plan.clone()),
7,
plan,
);
let stage = prepared.open();
let mut provider = provider();
let error = ready(allocate_snapshot(
&stage,
7,
payload,
&mut provider,
async |_| -> Result<(), Infallible> { panic!("wrong originals") },
async |_, _| -> Result<_, Infallible> { panic!("wrong originals cannot qualify") },
))
.unwrap_err();
assert!(matches!(
(fault, error),
(
PlanFault::Request,
IcSnapshotAllocationExecutionError::OriginalMismatch
) | (
PlanFault::Context(_),
IcSnapshotAllocationExecutionError::Upload(
IcSnapshotUploadExecutionError::Request(
IcSnapshotUploadAttemptError::SourceContextMismatch
)
)
)
));
assert_eq!(
read_execution_progress(stage.layout().unwrap(), &stage.plan().digest())
.unwrap()
.attempts
.mutations_used,
0
);
assert_eq!(provider.calls, 0);
}
});
}