use super::*;
use crate::model::attempt_journal::AttemptBudgetRecord;
use crate::model::operation_plan::{OperationPlanRecord, PlanBudgetRecord, PlannedOperationRecord};
use crate::policy::ResponseIdentityMismatch;
use crate::policy::ic_snapshot_transfer_read::IcSnapshotTransferReadAssociationError;
use crate::ports::IcProviderError;
use crate::tests::support::encode_fixture;
use crate::tests::support::operation_plan::{declaration, operation_request};
use crate::tests::support::provider::Provider;
use crate::tests::support::{execution_workflow::cancel_pending, retained_reply::write_synced};
use crate::workflow::ic_snapshot_transfer_read::IcSnapshotTransferReadReplyError;
use crate::tests::support::ic_snapshot_download::{source, source_plan, values, workflow};
use crate::tests::support::ready::ready;
use crate::{
model::{
execution_workflow::{ExecutionStageBindingRecord, ExecutionWorkflowRecord},
ic_snapshot_download::IcSnapshotDownloadPlan,
ic_snapshot_metadata::IcSnapshotMetadataReply,
},
ops::persistence::{BackupLayoutGuard, create_execution_workflow, read_execution_progress},
tests::support::temp_dir,
};
use std::{convert::Infallible, fs, io};
fn provider() -> Provider {
Provider::new(Ok(encode_fixture(values()).unwrap()))
}
struct MetadataFixture<'a> {
prepared: &'a crate::tests::support::execution_workflow::PreparedStage,
payload: &'a IcSnapshotMetadataRequest,
}
fn with_metadata_stage(check: impl FnOnce(MetadataFixture<'_>, ExecutionStageGuard<'_>, Provider)) {
let workflow = workflow(12);
let payload = source();
crate::tests::support::execution_workflow::with_stage_provider(
"ic-backup-metadata-stage",
&workflow,
0,
source_plan(&payload),
provider(),
|prepared, stage, provider| {
check(
MetadataFixture {
prepared,
payload: &payload,
},
stage,
provider,
);
},
);
}
fn admit(
_: &IcSnapshotTransferReadRequest<'_, '_>,
) -> impl std::future::Future<Output = Result<(), Infallible>> {
std::future::ready(Ok(()))
}
fn receipt(
request: &IcSnapshotTransferReadRequest<'_, '_>,
response: &IcSnapshotTransferReadResponse,
) -> MutationReceiptRequest {
crate::tests::support::update::applied_receipt(
request.mutation_attempt(),
&request.payload().digest(),
&response.input().evidence,
)
}
#[test]
fn cancelled_qualification_retains_reply_and_original_pending_spending() {
with_metadata_stage(|fixture, stage, mut provider| {
let retained = fixture.prepared.root.join("retained-metadata.candid");
let future = read_snapshot_metadata(
&stage,
42,
fixture.payload,
&mut provider,
admit,
async |_, response| {
let _file = write_synced(&retained, &response.input().reply).unwrap();
std::future::pending::<Result<MutationReceiptRequest, Infallible>>().await
},
);
let path = stage.layout().unwrap().root().join("attempt-42.json");
let before = cancel_pending(&stage, 42, future);
assert_eq!(provider.calls, 1);
assert_eq!(
fs::read(&retained).unwrap(),
encode_fixture(values()).unwrap()
);
assert!(
ready(read_snapshot_metadata(
&stage,
42,
fixture.payload,
&mut provider,
admit,
async |_, _| -> Result<_, Infallible> { panic!("no requalification") },
))
.is_err()
);
assert_eq!(provider.calls, 1);
assert_eq!(fs::read(path).unwrap(), before);
});
}
#[test]
fn explicit_receipt_checkpoints_exact_metadata_and_admits_original_data_stage() {
with_metadata_stage(|fixture, stage, mut provider| {
let (response, learned) = ready(read_snapshot_metadata(
&stage,
42,
fixture.payload,
&mut provider,
admit,
async |request, response| {
assert!(
AttemptJournalGuard::open(stage.layout().unwrap(), request.authority())
.is_err()
);
let _file = write_synced(
&fixture.prepared.root.join("retained-metadata.candid"),
&response.input().reply,
)
.unwrap();
Ok::<_, Infallible>(receipt(request, response))
},
))
.unwrap();
let predecessor = stage.reconcile_checkpoint(learned).unwrap();
assert_eq!(provider.calls, 1);
let metadata =
IcSnapshotMetadataReply::decode(fixture.payload, &response.input().reply).unwrap();
assert_eq!(predecessor.learned_evidence(), &metadata.digest());
let download =
IcSnapshotDownloadPlan::new(&fixture.prepared.workflow, 7, &metadata, 32).unwrap();
let bound = download
.bind(&fixture.prepared.binding, stage.plan(), vec![predecessor])
.unwrap();
drop(stage);
let data_stage = ExecutionStageGuard::prepare(
&fixture.prepared.layout,
bound,
download.plan().unwrap().clone(),
)
.unwrap();
assert_eq!(
read_execution_progress(data_stage.layout().unwrap(), &data_stage.plan().digest())
.unwrap()
.attempts
.mutations_used,
0
);
drop(data_stage);
let (reopened, progress) = ExecutionStageGuard::resume(
&fixture.prepared.layout,
&fixture.prepared.workflow.digest(),
0,
&fixture.prepared.binding.digest(),
)
.unwrap();
assert_eq!(progress.applied_operations, 1);
assert!(
ready(read_snapshot_metadata(
&reopened,
42,
fixture.payload,
&mut provider,
admit,
async |_, _| -> Result<_, Infallible> { panic!("no reissue") }
))
.is_err()
);
assert_eq!(provider.calls, 1);
});
}
#[test]
fn lost_and_malformed_metadata_stop_pending_without_checkpoint_or_data_stage() {
for raw in [
Err(IcProviderError::Unavailable),
Err(IcProviderError::Unsupported),
Err(IcProviderError::Indeterminate),
Ok(vec![]),
] {
with_metadata_stage(|fixture, stage, mut provider| {
let failure = raw.as_ref().err().copied();
provider.reply = raw;
let error = ready(read_snapshot_metadata(
&stage,
42,
fixture.payload,
&mut provider,
admit,
async |_, _| -> Result<_, Infallible> { panic!("unadmitted reply") },
))
.unwrap_err();
assert_eq!(
error.response().map(|reply| reply.input().reply.as_slice()),
provider.reply.as_ref().ok().map(Vec::as_slice)
);
match (failure, error) {
(None, IcSnapshotMetadataExecutionError::Read(
crate::workflow::ic_snapshot_transfer_read::IcSnapshotTransferReadExecutionError::AfterReply {
source: crate::workflow::ic_snapshot_transfer_read::IcSnapshotTransferReadReplyError::Association(_),
response,
},
)) => assert_eq!(response.input().reply, Vec::<u8>::new()),
(Some(expected), IcSnapshotMetadataExecutionError::Read(
crate::workflow::ic_snapshot_transfer_read::IcSnapshotTransferReadExecutionError::Provider(actual),
)) => assert_eq!(actual, expected),
other => panic!("expected pending read refusal: {other:?}"),
}
let references = stage.layout().unwrap().restore_references().unwrap();
let originals = crate::tests::support::retained_reply::OriginalFiles::capture([
stage.layout().unwrap().root().join("attempt-42.json"),
stage.layout().unwrap().root().join("operation-plan.json"),
]);
assert!(
ready(read_snapshot_metadata(
&stage,
42,
fixture.payload,
&mut provider,
admit,
async |_, _| -> Result<_, Infallible> {
panic!("pending reply never qualifies")
},
))
.is_err()
);
originals.assert_unchanged();
assert_eq!(
stage.layout().unwrap().restore_references().unwrap(),
references
);
crate::tests::support::execution_workflow::assert_pending(&stage, 42);
assert_eq!(provider.calls, 1);
assert!(
!fixture
.prepared
.layout
.root()
.join("execution-stage-7")
.exists()
);
});
}
}
#[test]
fn qualification_rejection_retains_returned_bytes_and_pending_original() {
with_metadata_stage(|fixture, stage, mut provider| {
let expected = provider.reply.clone().unwrap();
let error = ready(read_snapshot_metadata(
&stage,
42,
fixture.payload,
&mut provider,
async |_| Ok::<_, io::Error>(()),
async |_, _| Err(io::Error::other("no durable authenticated originals")),
))
.unwrap_err();
assert!(matches!(
error,
IcSnapshotMetadataExecutionError::AfterReply {
source: IcSnapshotMetadataSettlementError::Qualification(_),
..
}
));
assert_eq!(error.response().unwrap().input().reply, expected);
crate::tests::support::execution_workflow::assert_pending(&stage, 42);
});
}
#[test]
fn changed_stage_during_qualification_retains_reply_and_pending_spending() {
with_metadata_stage(|fixture, stage, mut provider| {
let path = stage.layout().unwrap().root().join("stage-binding.json");
let result = ready(read_snapshot_metadata(
&stage,
42,
fixture.payload,
&mut provider,
admit,
async |request, response| {
fs::write(&path, b"{}").unwrap();
Ok::<_, Infallible>(receipt(request, response))
},
));
assert!(matches!(
result,
Err(IcSnapshotMetadataExecutionError::AfterReply {
source: IcSnapshotMetadataSettlementError::Stage(_),
..
})
));
let journal: crate::model::attempt_journal::AttemptJournalRecord = serde_json::from_slice(
&fs::read(path.parent().unwrap().join("attempt-42.json")).unwrap(),
)
.unwrap();
assert_eq!(journal.view().pending_mutation, Some(1));
});
}
#[test]
fn wrong_payload_or_sequence_refuse_before_effects() {
with_metadata_stage(|fixture, stage, mut provider| {
let originals = crate::tests::support::retained_reply::OriginalFiles::capture([
stage.layout().unwrap().root().join("attempt-42.json"),
stage.layout().unwrap().root().join("operation-plan.json"),
]);
let wrong = IcSnapshotMetadataRequest::new(fixture.payload.target(), &[1]).unwrap();
for (sequence, request) in [(41, fixture.payload), (42, &wrong)] {
assert!(matches!(
ready(read_snapshot_metadata(
&stage,
sequence,
request,
&mut provider,
admit,
async |_, _| -> Result<_, Infallible> { panic!("no qualification") }
)),
Err(IcSnapshotMetadataExecutionError::OriginalMismatch)
));
}
originals.assert_unchanged();
assert_eq!(provider.calls, 0);
});
}
#[test]
fn fresh_admission_rejection_keeps_original_pending() {
with_metadata_stage(|fixture, stage, mut provider| {
let result = ready(read_snapshot_metadata(
&stage,
42,
fixture.payload,
&mut provider,
async |_| Err(io::Error::other("not freshly admitted")),
async |_, _| -> Result<_, io::Error> { panic!("unadmitted read cannot qualify") },
));
assert!(matches!(
result,
Err(IcSnapshotMetadataExecutionError::Read(
IcSnapshotTransferReadExecutionError::Admission(_)
))
));
assert_eq!(provider.calls, 0);
crate::tests::support::execution_workflow::assert_pending(&stage, 42);
});
}
#[test]
fn dispatch_boundary_drift_retains_exact_reply_and_pending_original() {
#[derive(Clone, Copy, Debug)]
enum Drift {
Admission,
Reply,
Target,
Layout,
}
for drift in [Drift::Admission, Drift::Reply, Drift::Target, Drift::Layout] {
with_metadata_stage(|fixture, stage, mut provider| {
let root = stage.layout().unwrap().root().to_path_buf();
let path = root.join("stage-binding.json");
let original = fs::read(&path).unwrap();
let raw = provider.reply.clone().unwrap();
match drift {
Drift::Admission => {}
Drift::Reply => provider.change_stage = Some(path.clone()),
Drift::Target => provider.target = Some("aaaaa-aa"),
Drift::Layout => provider.replace_layout = Some(root.clone()),
}
let error = ready(read_snapshot_metadata(
&stage,
42,
fixture.payload,
&mut provider,
async |_| {
if matches!(drift, Drift::Admission) {
fs::write(&path, b"{}").unwrap();
}
Ok::<_, Infallible>(())
},
async |_, _| -> Result<_, Infallible> { panic!("drift cannot qualify") },
))
.unwrap_err();
match (drift, error) {
(
Drift::Admission,
IcSnapshotMetadataExecutionError::Read(
IcSnapshotTransferReadExecutionError::Reservation(
crate::workflow::ExecutionReservationError::Stage(_),
),
),
) => {}
(
drift,
IcSnapshotMetadataExecutionError::Read(
IcSnapshotTransferReadExecutionError::AfterReply { source, response },
),
) => {
assert!(matches!(
(drift, source),
(Drift::Reply, IcSnapshotTransferReadReplyError::Stage(_))
| (Drift::Layout, IcSnapshotTransferReadReplyError::Journal(_))
| (
Drift::Target,
IcSnapshotTransferReadReplyError::Association(
IcSnapshotTransferReadAssociationError::Identity(
ResponseIdentityMismatch::Target
)
)
)
));
assert_eq!(response.input().reply, raw);
}
other => panic!("unexpected drift rejection: {other:?}"),
}
let retained = if matches!(drift, Drift::Layout) {
root.with_extension("retained-original")
} else {
fs::write(path, original).unwrap(); 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,
usize::from(!matches!(drift, Drift::Admission))
);
});
}
}
#[test]
fn multi_operation_stage_rejects_before_any_original_spending() {
let mut allocation = declaration(workflow(12).allocation());
let mut stage = operation_request(&allocation.operations[0]);
stage.budget = AttemptBudgetRecord::new(2, stage.budget.observations()).unwrap();
allocation.operations[0] = PlannedOperationRecord::new(stage).unwrap();
let workflow = ExecutionWorkflowRecord::new(OperationPlanRecord::new(allocation).unwrap());
let payload = source();
let mut child = declaration(&source_plan(&payload));
let mut second = operation_request(&child.operations[0]);
second.operation_sequence = 43;
child
.operations
.push(PlannedOperationRecord::new(second).unwrap());
child.budget = PlanBudgetRecord::new(2, child.budget.observations()).unwrap();
let plan = OperationPlanRecord::new(child).unwrap();
let binding = ExecutionStageBindingRecord::new(&workflow, 0, &plan, vec![]).unwrap();
let root = temp_dir("ic-backup-singleton-metadata");
let layout = BackupLayoutGuard::acquire(&root).unwrap();
create_execution_workflow(&layout, &workflow).unwrap();
let stage = ExecutionStageGuard::prepare(&layout, binding, plan).unwrap();
let mut provider = provider();
assert!(matches!(
ready(read_snapshot_metadata(
&stage,
42,
&payload,
&mut provider,
admit,
async |_, _| -> Result<_, Infallible> { panic!("not a metadata stage") }
)),
Err(IcSnapshotMetadataExecutionError::OriginalMismatch)
));
assert_eq!(provider.calls, 0);
assert_eq!(
read_execution_progress(stage.layout().unwrap(), &stage.plan().digest())
.unwrap()
.attempts
.mutations_used,
0
);
}