use super::*;
use crate::ports::IcProviderError;
use crate::tests::support::encode_fixture;
mod fixture;
use crate::tests::support::ic_snapshot_download::{source, source_plan, values, workflow};
use crate::tests::support::ready::ready;
use crate::tests::support::retained_reply::write_synced;
use crate::workflow::ic_snapshot_transfer_read::IcSnapshotTransferReadReplyError;
use crate::workflow::ic_snapshot_transfer_read::read_snapshot_with;
use crate::{
model::{
attempt_journal::MutationOutcomeRecord,
download_journal::{ArtifactStateRecord, DownloadArtifactRequest},
execution_workflow::{ExecutionStageBindingRecord, ExecutionWorkflowRecord},
ic_snapshot_metadata::IcSnapshotMetadataReply,
},
ops::persistence::{BackupLayoutGuard, DownloadJournalGuard, create_execution_workflow},
tests::support::temp_dir,
};
use fixture::{DownloadFixture, assert_progress, with_download, writer};
use ic_platform::management_canister_types::{ReadCanisterSnapshotDataResult, SnapshotDataKind};
use std::{convert::Infallible, fs, io, path::PathBuf};
const TOKEN: &str = "qualified-original";
fn prepare(
workflow: &ExecutionWorkflowRecord,
metadata: &IcSnapshotMetadataReply<'_>,
download: &IcSnapshotDownloadPlan<'_, '_>,
) -> (PathBuf, BackupLayoutGuard, ExecutionStageBindingRecord) {
let root = temp_dir("ic-backup-complete-download");
let layout = BackupLayoutGuard::acquire(&root).unwrap();
create_execution_workflow(&layout, workflow).unwrap();
let original = source_plan(metadata.request());
let binding = ExecutionStageBindingRecord::new(workflow, 0, &original, vec![]).unwrap();
let metadata_stage =
ExecutionStageGuard::prepare(&layout, binding.clone(), original.clone()).unwrap();
crate::tests::support::execution_workflow::record_applied(
&metadata_stage,
42,
metadata.digest().hash(),
);
let predecessor = metadata_stage
.reconcile_checkpoint(metadata.digest())
.unwrap();
drop(metadata_stage);
let binding = download
.bind(&binding, &original, vec![predecessor])
.unwrap();
let stage =
ExecutionStageGuard::prepare(&layout, binding.clone(), download.plan().unwrap().clone())
.unwrap();
fs::create_dir(stage.layout().unwrap().root().join("artifacts")).unwrap();
stage
.layout()
.unwrap()
.retain_restore(
&root.join("unfinished-restore.json"),
stage.plan().digest().hash(),
)
.unwrap();
drop(stage);
(root, layout, binding)
}
fn artifacts<'a>(
stage: &'a ExecutionStageGuard<'_>,
metadata: &IcSnapshotMetadataReply<'_>,
intent: &str,
) -> DownloadJournalGuard<'a> {
DownloadJournalGuard::create(
stage.layout().unwrap(),
intent,
DownloadArtifactRequest {
canister_id: metadata.request().target().into(),
snapshot_id: TOKEN.into(),
snapshot_taken_at_timestamp: metadata.metadata().taken_at_timestamp,
snapshot_total_size_bytes: u64::MAX,
},
)
.unwrap()
}
#[derive(Clone, Copy)]
enum DownloadFault {
None,
Lost { call: usize },
Malformed { call: usize },
Qualification,
WrongReceipt,
}
struct Provider {
calls: usize,
failure: DownloadFault,
}
impl Provider {
fn new(failure: DownloadFault) -> Self {
Self { calls: 0, failure }
}
}
impl IcSnapshotTransferReadProvider for Provider {
fn read_snapshot(
&mut self,
request: &IcSnapshotTransferReadRequest<'_, '_>,
journal: &crate::model::attempt_journal::AttemptJournalRecord,
) -> impl std::future::Future<Output = Result<IcSnapshotTransferReadResponse, IcProviderError>>
{
std::future::ready((|| {
request.validate_journal(journal).unwrap();
let index = self.calls;
self.calls += 1;
let IcSnapshotTransferReadPayload::Data(payload) = request.payload() else {
panic!("data only")
};
let chunk = match payload.kind() {
SnapshotDataKind::WasmModule { size, .. }
| SnapshotDataKind::WasmMemory { size, .. }
| SnapshotDataKind::StableMemory { size, .. } => {
vec![7; usize::try_from(*size).unwrap()]
}
SnapshotDataKind::WasmChunk { hash }
if *hash == payload.metadata().metadata().wasm_chunk_store[0].hash =>
{
vec![0, 255, 17]
}
SnapshotDataKind::WasmChunk { .. } => vec![],
};
let raw = match self.failure {
DownloadFault::Lost { call } if call == index => {
return Err(IcProviderError::Indeterminate);
}
DownloadFault::Malformed { call } if call == index => vec![],
_ => encode_fixture(ReadCanisterSnapshotDataResult { chunk }).unwrap(),
};
Ok(
IcSnapshotTransferReadResponse::new(crate::tests::support::update::input(
request.authority(),
request.mutation_attempt(),
request.plan().context(),
raw,
))
.unwrap(),
)
})())
}
}
fn admit(
_: &IcSnapshotTransferReadRequest<'_, '_>,
) -> impl std::future::Future<Output = Result<(), Infallible>> {
std::future::ready(Ok(()))
}
#[expect(
clippy::unnecessary_wraps,
reason = "fixture qualification explicitly returns a receipt under the required fallible integration contract"
)]
fn qualify(
request: &IcSnapshotTransferReadRequest<'_, '_>,
response: &IcSnapshotTransferReadResponse,
) -> Result<MutationReceiptRequest, Infallible> {
Ok(crate::tests::support::update::applied_receipt(
request.mutation_attempt(),
&request.payload().digest(),
&response.input().evidence,
))
}
fn raw_metadata() -> Vec<u8> {
encode_fixture(values()).unwrap()
}
#[test]
fn cancelled_second_qualification_retains_partial_bytes_first_receipt_and_pending_reply() {
use std::cell::Cell;
with_download(
|DownloadFixture {
root,
stage,
metadata,
raw,
download,
..
}| {
let mut artifacts = artifacts(&stage, metadata, stage.plan().digest().hash());
let partial = stage
.layout()
.unwrap()
.root()
.join(artifacts.record().unwrap().artifacts()[0].staging_path())
.join("wasm-module.bin");
let writer = writer(&mut artifacts, metadata, raw);
let mut provider = Provider::new(DownloadFault::None);
let qualified = Cell::new(0);
let mut future = Box::pin(download_snapshot(
&stage,
download,
writer,
&mut provider,
admit,
async |request, response| {
let sequence = request.authority().binding().operation_sequence();
let _file = write_synced(
&root.join(format!("reply-{sequence}.candid")),
&response.input().reply,
)
.unwrap();
qualified.set(qualified.get() + 1);
if sequence == 1 {
std::future::pending::<()>().await;
}
qualify(request, response)
},
));
crate::tests::support::ready::assert_pending(future.as_mut());
let authority = stage.plan().attempt_authority(1).unwrap();
assert!(AttemptJournalGuard::open(stage.layout().unwrap(), &authority).is_err());
let path = stage.layout().unwrap().root().join("attempt-1.json");
let before = fs::read(&path).unwrap();
let reply = fs::read(root.join("reply-1.candid")).unwrap();
let original_partial = fs::read(&partial).unwrap();
assert_eq!(original_partial, [7; 32]);
drop(future);
assert_eq!(provider.calls, 2);
assert_eq!(qualified.get(), 2);
assert_progress(&stage, 1, 2);
let journal = AttemptJournalGuard::open(stage.layout().unwrap(), &authority).unwrap();
assert_eq!(journal.record().unwrap().view().pending_mutation, Some(1));
drop(journal);
assert_eq!(
artifacts.record().unwrap().artifacts()[0].state(),
ArtifactStateRecord::Created
);
assert!(
artifacts
.stage_ic_snapshot_artifact(TOKEN, metadata, raw)
.is_err()
);
assert!(
ready(read_snapshot_with(
&stage,
1,
IcSnapshotTransferReadPayload::Data(
&download.requests().nth(1).unwrap().unwrap()
),
&mut provider,
admit,
async |_, _, _| ()
))
.is_err()
);
assert_eq!(provider.calls, 2);
assert_eq!(fs::read(path).unwrap(), before);
assert_eq!(fs::read(root.join("reply-1.candid")).unwrap(), reply);
assert_eq!(fs::read(partial).unwrap(), original_partial);
},
);
}
#[test]
#[expect(
clippy::too_many_lines,
reason = "keep each typed rejection, exact prefix and cold original-journal reopen in one trace"
)]
fn failed_reads_and_qualification_preserve_prefix_and_originals_across_reopen() {
for (failure, completed) in [
(DownloadFault::Lost { call: 0 }, 0),
(DownloadFault::Lost { call: 1 }, 1),
(DownloadFault::Malformed { call: 1 }, 1),
(DownloadFault::Qualification, 1),
(DownloadFault::WrongReceipt, 1),
] {
let root = with_download(
|DownloadFixture {
stage,
metadata,
raw,
download,
layout,
workflow,
binding,
..
}| {
let references = stage.layout().unwrap().restore_references().unwrap();
let mut artifacts = artifacts(&stage, metadata, stage.plan().digest().hash());
let writer = writer(&mut artifacts, metadata, raw);
let path = writer.path().to_path_buf();
let mut provider = Provider::new(failure);
let error = ready(download_snapshot(
&stage,
download,
writer,
&mut provider,
async |_| Ok::<_, io::Error>(()),
async |request, response| {
let mut receipt = qualify(request, response).unwrap();
if request.authority().binding().operation_sequence() == 1 {
match failure {
DownloadFault::Qualification => {
return Err(io::Error::other("authentication rejected"));
}
DownloadFault::WrongReceipt => {
receipt.outcome = MutationOutcomeRecord::NotApplied;
}
_ => {}
}
}
Ok::<_, io::Error>(receipt)
},
))
.unwrap_err();
match (failure, error) {
(
DownloadFault::Lost { .. },
IcSnapshotDownloadExecutionError::Read(
IcSnapshotTransferReadExecutionError::Provider(
IcProviderError::Indeterminate,
),
),
) => {}
(
DownloadFault::Malformed { .. },
IcSnapshotDownloadExecutionError::Read(
IcSnapshotTransferReadExecutionError::AfterReply {
source: IcSnapshotTransferReadReplyError::Association(_),
response,
},
),
) => assert_eq!(response.input().reply, [] as [u8; 0]),
(
DownloadFault::Qualification,
IcSnapshotDownloadExecutionError::AfterReply {
operation_sequence: 1,
source: IcSnapshotDownloadReplyError::Qualification(_),
response,
},
)
| (
DownloadFault::WrongReceipt,
IcSnapshotDownloadExecutionError::AfterReply {
operation_sequence: 1,
source: IcSnapshotDownloadReplyError::ReceiptRequired,
response,
},
) => assert_eq!(
response.input().reply,
encode_fixture(ReadCanisterSnapshotDataResult { chunk: vec![7; 32] })
.unwrap()
),
(_, other) => panic!("unexpected download rejection: {other:?}"),
}
assert_eq!(provider.calls, completed + 1);
assert_eq!(
fs::read(path.join("wasm-module.bin")).unwrap(),
vec![7; 32 * completed]
);
assert_eq!(
artifacts.record().unwrap().artifacts()[0].state(),
ArtifactStateRecord::Created
);
assert!(
artifacts
.stage_ic_snapshot_artifact(TOKEN, metadata, raw)
.is_err()
);
let progress =
assert_progress(&stage, completed, u32::try_from(completed + 1).unwrap());
let pending_path = stage
.layout()
.unwrap()
.root()
.join(format!("attempt-{completed}.json"));
let original = fs::read(&pending_path).unwrap();
assert!(
stage
.reconcile_checkpoint(ArtifactChecksumRecord::from_bytes(b"not complete"))
.is_err()
);
drop(artifacts);
drop(stage);
let (stage, current) =
ExecutionStageGuard::resume(layout, &workflow.digest(), 7, &binding.digest())
.unwrap();
assert_eq!(current, progress);
assert_eq!(fs::read(&pending_path).unwrap(), original);
assert_eq!(
stage.layout().unwrap().restore_references().unwrap(),
references
);
assert_eq!(provider.calls, completed + 1);
drop(stage);
},
);
fs::remove_dir_all(root).unwrap();
}
}
#[test]
fn consumed_missing_or_held_originals_never_enter_fresh_admission_or_provider() {
#[derive(Clone, Copy)]
enum OriginalFault {
Consumed,
Missing,
Held,
}
for fault in [
OriginalFault::Consumed,
OriginalFault::Missing,
OriginalFault::Held,
] {
let root = with_download(
|DownloadFixture {
stage,
metadata,
raw,
download,
..
}| {
let authority = stage.plan().attempt_authority(0).unwrap();
let mut held =
AttemptJournalGuard::open(stage.layout().unwrap(), &authority).unwrap();
if matches!(fault, OriginalFault::Consumed) {
held.reserve_planned_mutation(&stage.plan().digest())
.unwrap();
}
let held = if matches!(fault, OriginalFault::Held) {
Some(held)
} else {
drop(held);
None
};
if matches!(fault, OriginalFault::Missing) {
let path = stage.layout().unwrap().root().join("attempt-0.json");
fs::rename(&path, path.with_extension("retained-original")).unwrap();
}
let mut artifacts = artifacts(&stage, metadata, stage.plan().digest().hash());
let writer = writer(&mut artifacts, metadata, raw);
let mut provider = Provider::new(DownloadFault::None);
let error = ready(download_snapshot(
&stage,
download,
writer,
&mut provider,
async |_| -> Result<(), Infallible> { panic!("original rejection") },
async |request, response| qualify(request, response),
))
.unwrap_err();
assert!(matches!(
error,
IcSnapshotDownloadExecutionError::AlreadyAttempted
| IcSnapshotDownloadExecutionError::Progress(_)
));
assert_eq!(provider.calls, 0);
drop(held);
drop(artifacts);
drop(stage);
},
);
fs::remove_dir_all(root).unwrap();
}
}
#[test]
fn stage_or_local_byte_drift_during_qualification_retains_reply_and_pending_attempt() {
for stage_changed in [false, true] {
let root = with_download(
|DownloadFixture {
stage,
metadata,
raw,
download,
..
}| {
let stage_root = stage.layout().unwrap().root().to_path_buf();
let mut artifacts = artifacts(&stage, metadata, stage.plan().digest().hash());
let writer = writer(&mut artifacts, metadata, raw);
let path = writer.path().to_path_buf();
let mut provider = Provider::new(DownloadFault::None);
let error = ready(download_snapshot(
&stage,
download,
writer,
&mut provider,
admit,
async |request, response| {
if stage_changed {
fs::write(stage_root.join("stage-binding.json"), b"{}").unwrap();
} else {
fs::write(path.join("wasm-module.bin"), b"changed").unwrap();
}
qualify(request, response)
},
))
.unwrap_err();
let IcSnapshotDownloadExecutionError::AfterReply {
source, response, ..
} = error
else {
panic!("retained post-reply error")
};
assert!(matches!(
source,
IcSnapshotDownloadReplyError::Stage(_)
| IcSnapshotDownloadReplyError::Artifact(_)
));
assert_ne!(response.input().reply, [] as [u8; 0]);
assert_eq!(provider.calls, 1);
let journal: crate::model::attempt_journal::AttemptJournalRecord =
serde_json::from_slice(&fs::read(stage_root.join("attempt-0.json")).unwrap())
.unwrap();
assert_eq!(journal.view().pending_mutation, Some(1));
assert!(!journal.view().applied);
drop(artifacts);
drop(stage);
},
);
fs::remove_dir_all(root).unwrap();
}
}
#[derive(Clone, Copy)]
enum WriterFault {
Metadata,
Intent,
Coverage,
ReadFree,
}
#[test]
fn changed_writer_metadata_intent_prior_coverage_and_read_free_plan_reject_without_spending() {
for fault in [
WriterFault::Metadata,
WriterFault::Intent,
WriterFault::Coverage,
WriterFault::ReadFree,
] {
let root = with_download(
|DownloadFixture {
stage,
metadata,
raw,
download,
workflow,
request,
..
}| {
let mut other = values();
other.wasm_module_size += 1;
let other_raw = encode_fixture(other).unwrap();
let other_metadata = IcSnapshotMetadataReply::decode(request, &other_raw).unwrap();
let (selected_metadata, selected_raw) = if matches!(fault, WriterFault::Metadata) {
(&other_metadata, &other_raw)
} else {
(metadata, raw)
};
let intent = if matches!(fault, WriterFault::Intent) {
"44".repeat(32)
} else {
stage.plan().digest().hash().into()
};
let mut artifacts = artifacts(&stage, selected_metadata, &intent);
let mut writer = artifacts
.stage_ic_snapshot_artifact(TOKEN, selected_metadata, selected_raw)
.unwrap();
if matches!(fault, WriterFault::Coverage) {
let raw = encode_fixture(ReadCanisterSnapshotDataResult { chunk: vec![7; 32] })
.unwrap();
let request = download.requests().next().unwrap().unwrap();
let reply =
crate::model::ic_snapshot_data::IcSnapshotDataReply::decode(&request, &raw)
.unwrap();
writer = writer.append(&reply).unwrap();
}
let mut empty = values();
empty.wasm_module_size = 0;
empty.wasm_memory_size = 0;
empty.stable_memory_size = 0;
empty.wasm_chunk_store.clear();
let empty_raw = encode_fixture(empty).unwrap();
let empty_metadata = IcSnapshotMetadataReply::decode(request, &empty_raw).unwrap();
let empty_download =
IcSnapshotDownloadPlan::new(workflow, 7, &empty_metadata, 32).unwrap();
let selected = if matches!(fault, WriterFault::ReadFree) {
&empty_download
} else {
download
};
let mut provider = Provider::new(DownloadFault::None);
let error = ready(download_snapshot(
&stage,
selected,
writer,
&mut provider,
async |_| -> Result<(), Infallible> {
panic!("invalid original must reject before admission")
},
async |request, response| qualify(request, response),
))
.unwrap_err();
assert!(matches!(
error,
IcSnapshotDownloadExecutionError::OriginalMismatch
| IcSnapshotDownloadExecutionError::Artifact(_)
| IcSnapshotDownloadExecutionError::Planning(
SnapshotDataPlanningError::NoDataOperations
)
));
assert_eq!(provider.calls, 0);
assert_eq!(
read_execution_progress(stage.layout().unwrap(), &stage.plan().digest())
.unwrap()
.attempts
.mutations_used,
0
);
drop(artifacts);
drop(stage);
},
);
fs::remove_dir_all(root).unwrap();
}
}
#[test]
fn receipt_persistence_failure_retains_appended_bytes_reply_and_original_pending_reservation() {
let root = with_download(
|DownloadFixture {
stage,
metadata,
raw,
download,
..
}| {
let mut artifacts = artifacts(&stage, metadata, stage.plan().digest().hash());
let writer = writer(&mut artifacts, metadata, raw);
let path = writer.path().to_path_buf();
let journal_path = stage.layout().unwrap().root().join("attempt-0.json");
let retained = journal_path.with_extension("retained-original");
let mut provider = Provider::new(DownloadFault::None);
let error = ready(download_snapshot(
&stage,
download,
writer,
&mut provider,
admit,
async |request, response| {
fs::rename(&journal_path, &retained).unwrap();
fs::create_dir(&journal_path).unwrap();
qualify(request, response)
},
))
.unwrap_err();
let IcSnapshotDownloadExecutionError::AfterReply {
source: IcSnapshotDownloadReplyError::Journal(_),
response,
..
} = error
else {
panic!("retained receipt-write rejection")
};
assert_ne!(response.input().reply, [] as [u8; 0]);
assert_eq!(provider.calls, 1);
assert_eq!(fs::read(path.join("wasm-module.bin")).unwrap(), vec![7; 32]);
assert_eq!(
artifacts.record().unwrap().artifacts()[0].state(),
ArtifactStateRecord::Created
);
let original: crate::model::attempt_journal::AttemptJournalRecord =
serde_json::from_slice(&fs::read(retained).unwrap()).unwrap();
assert_eq!(original.view().pending_mutation, Some(1));
assert!(!original.view().applied);
drop(artifacts);
drop(stage);
},
);
fs::remove_dir_all(root).unwrap();
}