use super::*;
use crate::tests::support::{
execution_workflow::{cancel_pending, prepare_single_operation},
ready::ready,
};
#[derive(Debug, thiserror::Error)]
#[error("selected boundary failed")]
struct Failure;
struct Provider {
calls: usize,
checks: u8,
refuse_check: Option<u8>,
fault: u8,
outcome: ObservationOutcomeRecord,
checkpoint: Option<std::path::PathBuf>,
}
impl ApplicationIngressProvider for Provider {
type Error = Failure;
type Reply = Vec<u8>;
fn check(&mut self, _: &ApplicationRequest<'_>) -> Result<(), Failure> {
self.checks += 1;
if self.refuse_check == Some(self.checks) {
return Err(Failure);
}
Ok(())
}
async fn admit(&mut self, _: &ApplicationRequest<'_>) -> Result<(), Failure> {
if self.fault == 1 {
std::future::pending::<()>().await;
}
Ok(())
}
async fn submit(
&mut self,
_: &ApplicationRequest<'_>,
_: &AttemptJournalRecord,
) -> Result<Vec<u8>, Failure> {
self.calls += 1;
if self.fault == 2 {
return Err(Failure);
}
if self.fault == 5 {
std::future::pending::<()>().await;
}
Ok(b"retained exact reply".to_vec())
}
async fn qualify(
&mut self,
_: &ApplicationRequest<'_>,
_: &Vec<u8>,
) -> Result<ApplicationQualification, Failure> {
if self.fault == 3 {
return Err(Failure);
}
if self.fault == 4 {
std::future::pending::<()>().await;
}
if let Some(path) = &self.checkpoint {
std::fs::write(path, b"occupied checkpoint").unwrap();
}
Ok(ApplicationQualification {
outcome: self.outcome,
evidence: ArtifactChecksumRecord::from_bytes(b"fixture attribution only"),
learned: ArtifactChecksumRecord::from_bytes(b"fixture learned input"),
})
}
}
fn provider(fault: u8) -> Provider {
Provider {
calls: 0,
checks: 0,
refuse_check: None,
fault,
outcome: ObservationOutcomeRecord::Applied,
checkpoint: None,
}
}
fn payload(method: &str) -> ApplicationIngressPayload {
ApplicationIngressPayload::new(
"renrk-eyaaa-aaaaa-aaada-cai",
"renrk-eyaaa-aaaaa-aaada-cai",
method,
b"opaque",
128,
)
.unwrap()
}
#[test]
fn cancellation_and_lost_replies_preserve_spending_and_allow_only_original_observation() {
for fault in 1..=5 {
let payload = payload("prepare");
let (root, layout, binding, _) =
prepare_single_operation("opaque-recovery", payload.target(), payload.digest().hash());
let stage =
ExecutionStageGuard::open(&layout, binding.workflow(), 0, &binding.digest()).unwrap();
let mut transport = provider(fault);
if [1, 4, 5].contains(&fault) {
cancel_pending(&stage, 42, execute(&stage, 42, &payload, &mut transport));
} else {
let error = ready(execute(&stage, 42, &payload, &mut transport)).unwrap_err();
if fault == 3 {
assert!(
matches!(error, ApplicationExecutionError::AfterReply { reply, .. } if *reply == b"retained exact reply")
);
} else {
assert!(matches!(error, ApplicationExecutionError::BeforeReply(_)));
}
}
let previous = transport.calls;
assert!(ready(execute(&stage, 42, &payload, &mut transport)).is_err());
assert_eq!(transport.calls, previous);
drop(stage);
let (stage, progress) =
ExecutionStageGuard::resume(&layout, binding.workflow(), 0, &binding.digest()).unwrap();
assert_eq!(progress.attempts.mutations_used, 1);
let mut recovery = provider(0);
let result = ready(reconcile(
&stage,
42,
&self::payload("status"),
&mut recovery,
))
.unwrap();
assert_eq!(result.outcome, ObservationOutcomeRecord::Applied);
assert!(result.retained.is_some());
let calls = recovery.calls;
stage
.replay_checkpoint(ArtifactChecksumRecord::from_bytes(b"fixture learned input"))
.unwrap();
assert!(
ready(reconcile(
&stage,
42,
&self::payload("status"),
&mut recovery
))
.is_err()
);
assert_eq!(calls, recovery.calls);
drop(stage);
drop(layout);
std::fs::remove_dir_all(root).unwrap();
}
}
#[test]
fn observation_uncertainty_and_occupied_observations_never_create_mutation_authority() {
for outcome in [
ObservationOutcomeRecord::NotApplied,
ObservationOutcomeRecord::Uncertain,
] {
for fault in [0, 2, 3, 4] {
let payload = payload("prepare");
let (root, layout, binding, _) = prepare_single_operation(
"opaque-uncertain",
payload.target(),
payload.digest().hash(),
);
let stage =
ExecutionStageGuard::open(&layout, binding.workflow(), 0, &binding.digest())
.unwrap();
assert!(ready(execute(&stage, 42, &payload, &mut provider(2))).is_err());
let mut recovery = provider(fault);
recovery.outcome = outcome;
let observation = self::payload("status");
if fault == 4 {
cancel_pending(
&stage,
42,
reconcile(&stage, 42, &observation, &mut recovery),
);
} else {
let result = ready(reconcile(&stage, 42, &observation, &mut recovery));
if fault == 0 {
let result = result.unwrap();
assert_eq!(result.outcome, outcome);
assert!(result.retained.is_none());
} else {
assert!(result.is_err());
}
}
let calls = recovery.calls;
assert!(ready(execute(&stage, 42, &payload, &mut recovery)).is_err());
assert!(ready(reconcile(&stage, 42, &observation, &mut recovery)).is_err());
assert_eq!(recovery.calls, calls);
drop(stage);
drop(layout);
std::fs::remove_dir_all(root).unwrap();
}
}
}
#[test]
fn checkpoint_failure_retains_exact_reply_and_applied_journal_without_redispatch() {
let payload = payload("prepare");
let (root, layout, binding, _) = prepare_single_operation(
"opaque-checkpoint",
payload.target(),
payload.digest().hash(),
);
let stage =
ExecutionStageGuard::open(&layout, binding.workflow(), 0, &binding.digest()).unwrap();
let path = stage
.layout()
.unwrap()
.root()
.join("execution-settlement.json");
let mut provider = provider(0);
provider.checkpoint = Some(path.clone());
let error = ready(execute(&stage, 42, &payload, &mut provider)).unwrap_err();
assert!(matches!(error, ApplicationExecutionError::AfterReply {
source: ApplicationExecutionCause::Checkpoint(_), reply,
} if *reply == b"retained exact reply"));
let journal = crate::ops::persistence::AttemptJournalGuard::open(
stage.layout().unwrap(),
&stage.plan().attempt_authority(42).unwrap(),
)
.unwrap();
assert!(journal.record().unwrap().view().applied);
drop(journal);
assert!(ready(execute(&stage, 42, &payload, &mut provider)).is_err());
assert_eq!(provider.calls, 1);
assert_eq!(std::fs::read(path).unwrap(), b"occupied checkpoint");
drop(stage);
drop(layout);
std::fs::remove_dir_all(root).unwrap();
}
#[test]
fn successful_direct_reply_and_cold_checkpoint_replay_are_effect_free() {
let payload = payload("prepare");
let (root, layout, binding, _) =
prepare_single_operation("opaque-direct", payload.target(), payload.digest().hash());
let stage =
ExecutionStageGuard::open(&layout, binding.workflow(), 0, &binding.digest()).unwrap();
let mut provider = provider(0);
let result = ready(execute(&stage, 42, &payload, &mut provider)).unwrap();
assert_eq!(result.reply, b"retained exact reply");
assert_eq!(
result.retained.as_ref().unwrap().learned_evidence(),
&ArtifactChecksumRecord::from_bytes(b"fixture learned input")
);
assert!(result.retained.is_some());
drop(stage);
let (stage, progress) =
ExecutionStageGuard::resume(&layout, binding.workflow(), 0, &binding.digest()).unwrap();
assert_eq!(progress.applied_operations, 1);
stage
.replay_checkpoint(ArtifactChecksumRecord::from_bytes(b"fixture learned input"))
.unwrap();
assert!(ready(execute(&stage, 42, &payload, &mut provider)).is_err());
assert_eq!(provider.calls, 1);
drop(stage);
drop(layout);
std::fs::remove_dir_all(root).unwrap();
}
#[test]
fn changed_payload_and_unqualified_mutation_outcomes_cannot_publish_receipts() {
for outcome in [
ObservationOutcomeRecord::NotApplied,
ObservationOutcomeRecord::Uncertain,
] {
let payload = payload("prepare");
let (root, layout, binding, _) =
prepare_single_operation("opaque-claims", payload.target(), payload.digest().hash());
let stage =
ExecutionStageGuard::open(&layout, binding.workflow(), 0, &binding.digest()).unwrap();
let mut provider = provider(0);
provider.outcome = outcome;
assert!(matches!(
ready(execute(
&stage,
42,
&self::payload("changed"),
&mut provider
)),
Err(ApplicationExecutionError::BeforeReply(
ApplicationExecutionCause::Request(ApplicationIngressError::PayloadMismatch)
))
));
assert_eq!(provider.calls, 0);
assert!(
matches!(ready(execute(&stage, 42, &payload, &mut provider)),
Err(ApplicationExecutionError::AfterReply { source: ApplicationExecutionCause::MutationOutcome, reply }) if *reply == b"retained exact reply")
);
crate::tests::support::execution_workflow::assert_pending(&stage, 42);
drop(stage);
drop(layout);
std::fs::remove_dir_all(root).unwrap();
}
}
#[test]
fn authority_is_readmitted_after_each_await_before_dispatch_or_receipt() {
for boundary in 1..=4 {
let payload = payload("prepare");
let (root, layout, binding, _) = prepare_single_operation(
"opaque-authority",
payload.target(),
payload.digest().hash(),
);
let stage =
ExecutionStageGuard::open(&layout, binding.workflow(), 0, &binding.digest()).unwrap();
let mut provider = provider(0);
provider.refuse_check = Some(boundary);
let error = ready(execute(&stage, 42, &payload, &mut provider)).unwrap_err();
assert_eq!(provider.checks, boundary);
assert_eq!(provider.calls, usize::from(boundary > 2));
match error {
ApplicationExecutionError::BeforeReply(ApplicationExecutionCause::Application(_)) => {
assert!(boundary <= 2);
}
ApplicationExecutionError::AfterReply {
source: ApplicationExecutionCause::Application(_),
reply,
} => {
assert!(boundary > 2);
assert_eq!(*reply, b"retained exact reply");
}
error => panic!("unexpected refusal: {error:?}"),
}
crate::tests::support::execution_workflow::assert_pending(&stage, 42);
drop(stage);
drop(layout);
std::fs::remove_dir_all(root).unwrap();
}
}