use super::*;
use crate::tests::support::{
execution_workflow::{cancel_pending, child, prepare_stage, workflow},
ready::ready,
retained_reply::OriginalFiles,
};
use std::{cell::Cell, fs};
#[derive(Debug, thiserror::Error)]
enum Failure {
#[error(transparent)]
Reservation(#[from] ExecutionReservationError),
#[error("fixture stopped at its selected boundary")]
Injected,
}
#[test]
fn cancelled_or_failed_callbacks_preserve_originals_across_resume_and_deny_reentry() {
for cancel in [true, false] {
for during_admission in [true, false] {
let prepared = prepare_stage("ic-backup-reserved-step", &workflow(), 0, child(0));
let stage = prepared.open();
let authority = stage.plan().attempt_authority(42).unwrap();
let layout = stage.layout().unwrap();
layout
.retain_restore(
&prepared.root.join("unfinished.json"),
stage.plan().digest().hash(),
)
.unwrap();
let original = OriginalFiles::capture([
layout.root().join("operation-plan.json"),
layout.root().join("restore-references.json"),
]);
let calls = Cell::new(0);
let future = execute_reserved(
&stage,
&authority,
|_| Ok::<_, Failure>(()),
|journal| Ok(journal.view().pending_mutation.unwrap()),
async |attempt| {
assert_eq!(*attempt, 1);
assert!(AttemptJournalGuard::open(layout, &authority).is_err());
if during_admission {
if cancel {
std::future::pending::<()>().await;
}
return Err(Failure::Injected);
}
Ok(())
},
async |attempt, journal| {
calls.set(calls.get() + 1);
assert_eq!(journal.view().pending_mutation, Some(*attempt));
assert!(AttemptJournalGuard::open(layout, &authority).is_err());
if cancel {
std::future::pending::<()>().await;
}
Err::<(), _>(Failure::Injected)
},
async |_, _, ()| panic!("failed or cancelled submission cannot complete"),
);
if cancel {
cancel_pending(&stage, 42, future);
} else {
assert!(matches!(ready(future), Err(Failure::Injected)));
}
let journal_path = layout.root().join("attempt-42.json");
let bytes = fs::read(&journal_path).unwrap();
drop(stage);
let (stage, progress) = ExecutionStageGuard::resume(
&prepared.layout,
prepared.binding.workflow(),
0,
&prepared.binding.digest(),
)
.unwrap();
assert_eq!(progress.attempts.mutations_used, 1);
assert_eq!(progress.applied_operations, 0);
let journal = AttemptJournalGuard::open(stage.layout().unwrap(), &authority).unwrap();
assert_eq!(journal.record().unwrap().view().pending_mutation, Some(1));
assert_eq!(journal.record().unwrap().view().mutations_remaining, 0);
drop(journal);
assert!(matches!(
ready(execute_reserved::<(), (), (), Failure>(
&stage,
&authority,
|_| Ok(()),
|_| panic!("pending original cannot bind again"),
async |()| panic!("pending original cannot enter admission"),
async |(), _| panic!("pending original cannot submit again"),
async |_, (), ()| panic!("pending original cannot complete"),
)),
Err(Failure::Reservation(ExecutionReservationError::Progress(_)))
));
assert_eq!(calls.get(), usize::from(!during_admission));
assert_eq!(fs::read(journal_path).unwrap(), bytes);
original.assert_unchanged();
}
}
}
#[test]
fn held_missing_and_applied_journals_refuse_before_any_callback() {
use crate::model::artifacts::ArtifactChecksumRecord;
use crate::model::attempt_journal::{MutationOutcomeRecord, MutationReceiptRequest};
let prepared = prepare_stage("ic-backup-reservation-refusal", &workflow(), 0, child(0));
let stage = prepared.open();
let authority = stage.plan().attempt_authority(42).unwrap();
let layout = stage.layout().unwrap();
let refuse = || {
assert!(
ready(execute_reserved::<(), (), (), Failure>(
&stage,
&authority,
|_| Ok(()),
|_| panic!("unavailable or spent journal cannot bind"),
async |()| panic!("unavailable or spent journal cannot admit"),
async |(), _| panic!("unavailable or spent journal cannot submit"),
async |_, (), ()| panic!("unavailable or spent journal cannot complete"),
))
.is_err()
);
};
let mut journal = AttemptJournalGuard::open(layout, &authority).unwrap();
let path = journal.path();
let original = fs::read(&path).unwrap();
refuse();
assert_eq!(fs::read(&path).unwrap(), original);
journal
.reserve_planned_mutation(&stage.plan().digest())
.unwrap();
journal
.record_mutation(MutationReceiptRequest {
attempt: 1,
request: ArtifactChecksumRecord::from_hash(authority.binding().request()).unwrap(),
outcome: MutationOutcomeRecord::Applied,
evidence: ArtifactChecksumRecord::from_bytes(b"qualified original receipt"),
})
.unwrap();
let applied = fs::read(&path).unwrap();
drop(journal);
refuse();
assert_eq!(fs::read(&path).unwrap(), applied);
let retained = path.with_extension("retained-original");
fs::rename(&path, &retained).unwrap();
refuse();
assert!(!path.exists());
assert_eq!(fs::read(retained).unwrap(), applied);
}