use crate::model::artifacts::ArtifactChecksumRecord;
use crate::model::execution_workflow::*;
use crate::model::operation_plan::{
OperationPlanRecord, OperationPlanRequest, PlannedOperationRecord, PlannedOperationRequest,
};
use crate::ops::persistence::{BackupLayoutGuard, ExecutionStageGuard, create_execution_workflow};
use std::path::PathBuf;
pub(crate) fn workflow() -> ExecutionWorkflowRecord {
ExecutionWorkflowRecord::new(
OperationPlanRecord::new(crate::tests::support::operation_plan::request()).unwrap(),
)
}
pub(crate) fn child_request(sequence: u64) -> OperationPlanRequest {
let stage = workflow().stage(sequence).unwrap().clone();
crate::tests::support::operation_plan::single_operation_request(
crate::tests::support::operation_plan::request(),
PlannedOperationRequest {
depends_on: vec![],
operation_sequence: 42,
target: stage.target().into(),
request: "78".repeat(32),
budget: stage.budget().clone(),
},
)
}
pub(crate) fn child(sequence: u64) -> OperationPlanRecord {
OperationPlanRecord::new(child_request(sequence)).unwrap()
}
pub(crate) fn prepare_single_operation(
prefix: &str,
target: &str,
request: &str,
) -> (
PathBuf,
BackupLayoutGuard,
ExecutionStageBindingRecord,
OperationPlanRecord,
) {
let workflow = workflow();
let mut child = child_request(0);
child.operations = vec![
PlannedOperationRecord::new(PlannedOperationRequest {
depends_on: vec![],
operation_sequence: 42,
target: target.into(),
request: request.into(),
budget: workflow.stage(0).unwrap().budget().clone(),
})
.unwrap(),
];
let plan = OperationPlanRecord::new(child).unwrap();
let prepared = prepare_stage(prefix, &workflow, 0, plan.clone());
(prepared.root, prepared.layout, prepared.binding, plan)
}
pub(crate) fn prepare_stage(
prefix: &str,
workflow: &ExecutionWorkflowRecord,
sequence: u64,
plan: OperationPlanRecord,
) -> PreparedStage {
let root = crate::tests::support::temp_dir(prefix);
let layout = BackupLayoutGuard::acquire(&root).unwrap();
create_execution_workflow(&layout, workflow).unwrap();
let binding = ExecutionStageBindingRecord::new(workflow, sequence, &plan, vec![]).unwrap();
drop(ExecutionStageGuard::prepare(&layout, binding.clone(), plan).unwrap());
PreparedStage {
root,
layout,
workflow: workflow.clone(),
binding,
sequence,
}
}
pub(crate) struct PreparedStage {
pub root: PathBuf,
pub layout: BackupLayoutGuard,
pub workflow: ExecutionWorkflowRecord,
pub binding: ExecutionStageBindingRecord,
sequence: u64,
}
impl PreparedStage {
pub fn open(&self) -> ExecutionStageGuard<'_> {
ExecutionStageGuard::open(
&self.layout,
self.binding.workflow(),
self.sequence,
&self.binding.digest(),
)
.unwrap()
}
}
pub(crate) fn assert_pending(stage: &ExecutionStageGuard<'_>, sequence: u64) {
let progress = crate::ops::persistence::read_execution_progress(
stage.layout().unwrap(),
&stage.plan().digest(),
)
.unwrap();
assert_eq!(progress.attempts.mutations_used, 1);
assert_eq!(progress.applied_operations, 0);
let journal = crate::ops::persistence::AttemptJournalGuard::open(
stage.layout().unwrap(),
&stage.plan().attempt_authority(sequence).unwrap(),
)
.unwrap();
assert_eq!(journal.record().unwrap().view().pending_mutation, Some(1));
}
pub(crate) fn record_applied(stage: &ExecutionStageGuard<'_>, sequence: u64, evidence: &str) {
use crate::model::attempt_journal::{MutationOutcomeRecord, MutationReceiptRequest};
use crate::ops::persistence::AttemptJournalGuard;
let mut journal = AttemptJournalGuard::open(
stage.layout().unwrap(),
&stage.plan().attempt_authority(sequence).unwrap(),
)
.unwrap();
let attempt = journal
.reserve_planned_mutation(&stage.plan().digest())
.unwrap();
journal
.record_mutation(MutationReceiptRequest {
attempt,
request: ArtifactChecksumRecord::from_hash(
stage.plan().operation(sequence).unwrap().request(),
)
.unwrap(),
outcome: MutationOutcomeRecord::Applied,
evidence: ArtifactChecksumRecord::from_hash(evidence).unwrap(),
})
.unwrap();
}
pub(crate) fn cancel_pending(
stage: &ExecutionStageGuard<'_>,
sequence: u64,
future: impl std::future::Future,
) -> Vec<u8> {
let mut future = Box::pin(future);
crate::tests::support::ready::assert_pending(future.as_mut());
assert!(
crate::ops::persistence::AttemptJournalGuard::open(
stage.layout().unwrap(),
&stage.plan().attempt_authority(sequence).unwrap()
)
.is_err()
);
let original = std::fs::read(
stage
.layout()
.unwrap()
.root()
.join(format!("attempt-{sequence}.json")),
)
.unwrap();
drop(future);
assert_pending(stage, sequence);
original
}