use crate::WorkGraphError;
use crate::generated::{
protocol_work_execution_cancellation_evidence_projection as cancellation_evidence_protocol,
protocol_work_execution_failure_evidence_projection as failure_evidence_protocol,
protocol_work_execution_flow_launch as flow_launch_protocol,
protocol_work_execution_flow_observation as flow_observation_protocol,
protocol_work_execution_launch_failure_evidence_projection as launch_failure_evidence_protocol,
protocol_work_execution_quarantined_launch_resolution as quarantined_launch_protocol,
protocol_work_execution_success_evidence_projection as success_evidence_protocol,
protocol_work_execution_uncertain_launch_resolution as uncertain_launch_protocol,
protocol_work_execution_work_closure as work_closure_protocol,
};
use crate::machines::work_execution_lifecycle as execution_dsl;
use crate::types::{WorkExecutionBinding, WorkExecutionBindingId, WorkExecutionMachineState};
pub use execution_dsl::{WorkExecutionLifecycleEffect, WorkExecutionLifecycleState};
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub enum WorkExecutionObservation {
FlowStarted,
FlowRunning,
FlowCompleted,
FlowFailed { detail: Option<String> },
FlowCanceled { detail: Option<String> },
FlowRunLost { detail: String },
LaunchUncertain { detail: String },
LaunchQuarantined { detail: String },
LaunchFailed { detail: String },
EvidenceProjected,
FlowFailureEvidenceProjected,
FlowCancellationEvidenceProjected,
LaunchFailureEvidenceProjected,
WorkClosed,
WorkClosureRefused { detail: String },
}
#[derive(Debug, Default, Clone, Copy)]
pub struct WorkExecutionMachine;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WorkExecutionTransition {
pub binding: WorkExecutionBinding,
pub effect: WorkExecutionLifecycleEffect,
}
pub struct WorkExecutionBindCommit {
binding: WorkExecutionBinding,
effect: WorkExecutionLifecycleEffect,
}
impl WorkExecutionBindCommit {
pub(crate) fn into_parts(self) -> (WorkExecutionBinding, WorkExecutionLifecycleEffect) {
(self.binding, self.effect)
}
pub(crate) fn binding(&self) -> &WorkExecutionBinding {
&self.binding
}
pub(crate) fn effect(&self) -> &WorkExecutionLifecycleEffect {
&self.effect
}
}
pub struct WorkExecutionObservationCommit {
previous: WorkExecutionBinding,
observation: WorkExecutionObservation,
binding: WorkExecutionBinding,
effect: WorkExecutionLifecycleEffect,
}
impl WorkExecutionObservationCommit {
pub(crate) fn into_parts(
self,
) -> (
WorkExecutionBinding,
WorkExecutionObservation,
WorkExecutionBinding,
WorkExecutionLifecycleEffect,
) {
(self.previous, self.observation, self.binding, self.effect)
}
pub(crate) fn binding(&self) -> &WorkExecutionBinding {
&self.binding
}
pub(crate) fn effect(&self) -> &WorkExecutionLifecycleEffect {
&self.effect
}
}
impl WorkExecutionMachine {
pub(crate) fn prepare_bind(
binding: WorkExecutionBinding,
) -> Result<WorkExecutionBindCommit, WorkGraphError> {
binding.validate()?;
let (expected, effect) = Self::bind(&binding.binding_id, binding.target.run_id())?;
if binding.machine_state != expected {
return Err(WorkGraphError::InvalidInput(format!(
"work execution binding {} was not initialized by WorkExecutionLifecycleMachine",
binding.binding_id
)));
}
Ok(WorkExecutionBindCommit { binding, effect })
}
pub(crate) fn prepare_observation(
previous: WorkExecutionBinding,
expected_revision: u64,
observation: WorkExecutionObservation,
) -> Result<WorkExecutionObservationCommit, WorkGraphError> {
let (binding, effect) =
Self::observe(previous.clone(), expected_revision, observation.clone())?;
Ok(WorkExecutionObservationCommit {
previous,
observation,
binding,
effect,
})
}
pub fn recover_effect(
binding: &WorkExecutionBinding,
) -> Result<WorkExecutionLifecycleEffect, WorkGraphError> {
validate_projection(binding)?;
let mut authority =
execution_dsl::WorkExecutionLifecycleMachineAuthority::recover_from_state(
binding.machine_state.clone(),
)
.map_err(|error| {
WorkGraphError::InvalidTransition(format!(
"work execution {} refused recovery: {error:?}",
binding.binding_id
))
})?;
let transition = execution_dsl::WorkExecutionLifecycleMachineMutator::apply(
&mut authority,
execution_dsl::WorkExecutionLifecycleInput::Recover {},
)
.map_err(|error| {
WorkGraphError::InvalidTransition(format!(
"work execution {} refused effect recovery: {error:?}",
binding.binding_id
))
})?;
validate_handoff_obligation(&transition)?;
exactly_one_effect(transition.effects())
}
pub fn bind(
binding_id: &WorkExecutionBindingId,
run_id: &str,
) -> Result<(WorkExecutionMachineState, WorkExecutionLifecycleEffect), WorkGraphError> {
let mut authority = execution_dsl::WorkExecutionLifecycleMachineAuthority::new();
let transition = execution_dsl::WorkExecutionLifecycleMachineMutator::apply(
&mut authority,
execution_dsl::WorkExecutionLifecycleInput::Bind {
binding_id: binding_id.as_str().to_string(),
run_id: run_id.to_string(),
},
)
.map_err(|error| {
WorkGraphError::InvalidTransition(format!(
"generated work execution binding transition refused: {error:?}"
))
})?;
validate_handoff_obligation(&transition)?;
let effect = exactly_one_effect(transition.effects())?;
Ok((authority.state().clone(), effect))
}
pub fn observe(
mut binding: WorkExecutionBinding,
expected_revision: u64,
observation: WorkExecutionObservation,
) -> Result<(WorkExecutionBinding, WorkExecutionLifecycleEffect), WorkGraphError> {
validate_observation_detail(&observation)?;
if binding.machine_state.revision != expected_revision {
return Err(WorkGraphError::Conflict(format!(
"stale work execution revision for {}: expected {}, actual {}",
binding.binding_id, expected_revision, binding.machine_state.revision
)));
}
validate_projection(&binding)?;
let mut authority =
execution_dsl::WorkExecutionLifecycleMachineAuthority::recover_from_state(
binding.machine_state.clone(),
)
.map_err(|error| {
WorkGraphError::InvalidTransition(format!(
"work execution {} refused recovery: {error:?}",
binding.binding_id
))
})?;
let mut recovery_authority =
execution_dsl::WorkExecutionLifecycleMachineAuthority::recover_from_state(
binding.machine_state.clone(),
)
.map_err(|error| {
WorkGraphError::InvalidTransition(format!(
"work execution {} refused handoff recovery: {error:?}",
binding.binding_id
))
})?;
let recovery_transition = execution_dsl::WorkExecutionLifecycleMachineMutator::apply(
&mut recovery_authority,
execution_dsl::WorkExecutionLifecycleInput::Recover {},
)
.map_err(|error| {
WorkGraphError::InvalidTransition(format!(
"work execution {} refused handoff effect recovery: {error:?}",
binding.binding_id
))
})?;
validate_handoff_obligation(&recovery_transition)?;
let recovered_effect = exactly_one_effect(recovery_transition.effects())?;
let observation_debug = format!("{observation:?}");
macro_rules! submit {
($protocol:ident, $function:ident $(, $argument:expr)*) => {{
let obligation = exactly_one_obligation(
$protocol::extract_obligations(&recovery_transition),
&binding.binding_id,
)?;
$protocol::$function(&mut authority, obligation $(, $argument)*)
}};
}
let transition = match (&recovered_effect, observation) {
(WorkExecutionLifecycleEffect::FlowLaunchRequested { .. }, WorkExecutionObservation::FlowStarted) =>
submit!(flow_launch_protocol, submit_confirm_flow_started),
(WorkExecutionLifecycleEffect::FlowLaunchRequested { .. }, WorkExecutionObservation::FlowRunning) =>
submit!(flow_launch_protocol, submit_observe_flow_running),
(WorkExecutionLifecycleEffect::FlowLaunchRequested { .. }, WorkExecutionObservation::FlowCompleted) =>
submit!(flow_launch_protocol, submit_observe_flow_completed),
(WorkExecutionLifecycleEffect::FlowLaunchRequested { .. }, WorkExecutionObservation::FlowFailed { detail }) =>
submit!(flow_launch_protocol, submit_observe_flow_failed, detail),
(WorkExecutionLifecycleEffect::FlowLaunchRequested { .. }, WorkExecutionObservation::FlowCanceled { detail }) =>
submit!(flow_launch_protocol, submit_observe_flow_canceled, detail),
(WorkExecutionLifecycleEffect::FlowLaunchRequested { .. }, WorkExecutionObservation::LaunchUncertain { detail }) =>
submit!(flow_launch_protocol, submit_mark_launch_uncertain, detail),
(WorkExecutionLifecycleEffect::FlowLaunchRequested { .. }, WorkExecutionObservation::LaunchQuarantined { detail }) =>
submit!(flow_launch_protocol, submit_quarantine_launch, detail),
(WorkExecutionLifecycleEffect::FlowLaunchRequested { .. }, WorkExecutionObservation::LaunchFailed { detail }) =>
submit!(flow_launch_protocol, submit_resolve_launch_failed, detail),
(WorkExecutionLifecycleEffect::FlowLaunchAccepted { .. }, WorkExecutionObservation::FlowRunning) =>
submit!(flow_observation_protocol, submit_observe_flow_running),
(WorkExecutionLifecycleEffect::FlowLaunchAccepted { .. }, WorkExecutionObservation::FlowCompleted) =>
submit!(flow_observation_protocol, submit_observe_flow_completed),
(WorkExecutionLifecycleEffect::FlowLaunchAccepted { .. }, WorkExecutionObservation::FlowFailed { detail }) =>
submit!(flow_observation_protocol, submit_observe_flow_failed, detail),
(WorkExecutionLifecycleEffect::FlowLaunchAccepted { .. }, WorkExecutionObservation::FlowCanceled { detail }) =>
submit!(flow_observation_protocol, submit_observe_flow_canceled, detail),
(WorkExecutionLifecycleEffect::FlowLaunchAccepted { .. }, WorkExecutionObservation::FlowRunLost { detail }) =>
submit!(flow_observation_protocol, submit_observe_run_lost, detail),
(WorkExecutionLifecycleEffect::FlowLaunchUncertain { .. }, WorkExecutionObservation::FlowStarted) =>
submit!(uncertain_launch_protocol, submit_confirm_flow_started),
(WorkExecutionLifecycleEffect::FlowLaunchUncertain { .. }, WorkExecutionObservation::FlowRunning) =>
submit!(uncertain_launch_protocol, submit_observe_flow_running),
(WorkExecutionLifecycleEffect::FlowLaunchUncertain { .. }, WorkExecutionObservation::FlowCompleted) =>
submit!(uncertain_launch_protocol, submit_observe_flow_completed),
(WorkExecutionLifecycleEffect::FlowLaunchUncertain { .. }, WorkExecutionObservation::FlowFailed { detail }) =>
submit!(uncertain_launch_protocol, submit_observe_flow_failed, detail),
(WorkExecutionLifecycleEffect::FlowLaunchUncertain { .. }, WorkExecutionObservation::FlowCanceled { detail }) =>
submit!(uncertain_launch_protocol, submit_observe_flow_canceled, detail),
(WorkExecutionLifecycleEffect::FlowLaunchUncertain { .. }, WorkExecutionObservation::LaunchFailed { detail }) =>
submit!(uncertain_launch_protocol, submit_resolve_launch_failed, detail),
(WorkExecutionLifecycleEffect::FlowLaunchUncertain { .. }, WorkExecutionObservation::LaunchQuarantined { detail }) =>
submit!(uncertain_launch_protocol, submit_quarantine_launch, detail),
(WorkExecutionLifecycleEffect::FlowLaunchQuarantined { .. }, WorkExecutionObservation::FlowStarted) =>
submit!(quarantined_launch_protocol, submit_confirm_flow_started),
(WorkExecutionLifecycleEffect::FlowLaunchQuarantined { .. }, WorkExecutionObservation::FlowRunning) =>
submit!(quarantined_launch_protocol, submit_observe_flow_running),
(WorkExecutionLifecycleEffect::FlowLaunchQuarantined { .. }, WorkExecutionObservation::FlowCompleted) =>
submit!(quarantined_launch_protocol, submit_observe_flow_completed),
(WorkExecutionLifecycleEffect::FlowLaunchQuarantined { .. }, WorkExecutionObservation::FlowFailed { detail }) =>
submit!(quarantined_launch_protocol, submit_observe_flow_failed, detail),
(WorkExecutionLifecycleEffect::FlowLaunchQuarantined { .. }, WorkExecutionObservation::FlowCanceled { detail }) =>
submit!(quarantined_launch_protocol, submit_observe_flow_canceled, detail),
(WorkExecutionLifecycleEffect::EvidenceProjectionRequested { .. }, WorkExecutionObservation::EvidenceProjected) =>
submit!(success_evidence_protocol, submit_confirm_evidence_projected),
(WorkExecutionLifecycleEffect::EvidenceProjectionRequested { .. }, WorkExecutionObservation::FlowRunLost { detail }) =>
submit!(success_evidence_protocol, submit_observe_run_lost, detail),
(WorkExecutionLifecycleEffect::FlowFailureEvidenceProjectionRequested { .. }, WorkExecutionObservation::FlowFailureEvidenceProjected) =>
submit!(failure_evidence_protocol, submit_confirm_flow_failure_evidence_projected),
(WorkExecutionLifecycleEffect::FlowCancellationEvidenceProjectionRequested { .. }, WorkExecutionObservation::FlowCancellationEvidenceProjected) =>
submit!(cancellation_evidence_protocol, submit_confirm_flow_cancellation_evidence_projected),
(WorkExecutionLifecycleEffect::LaunchFailureEvidenceProjectionRequested { .. }, WorkExecutionObservation::LaunchFailureEvidenceProjected) =>
submit!(launch_failure_evidence_protocol, submit_confirm_launch_failure_evidence_projected),
(WorkExecutionLifecycleEffect::WorkClosureRequested { .. }, WorkExecutionObservation::WorkClosed) =>
submit!(work_closure_protocol, submit_confirm_work_closed),
(WorkExecutionLifecycleEffect::WorkClosureRequested { .. }, WorkExecutionObservation::WorkClosureRefused { detail }) =>
submit!(work_closure_protocol, submit_refuse_work_closure, detail),
_ => {
return Err(WorkGraphError::InvalidTransition(format!(
"work execution {} observation {observation_debug} is not admitted by the generated owner-feedback protocol for {recovered_effect:?}",
binding.binding_id
)));
}
}
.map_err(|error| {
WorkGraphError::InvalidTransition(format!(
"work execution {} refused observation: {error:?}",
binding.binding_id
))
})?;
validate_handoff_obligation(&transition)?;
let effect = exactly_one_effect(transition.effects())?;
binding.machine_state = authority.state().clone();
validate_projection(&binding)?;
Ok((binding, effect))
}
pub fn validate_projection(binding: &WorkExecutionBinding) -> Result<(), WorkGraphError> {
validate_projection(binding)
}
pub fn retry_eligible(binding: &WorkExecutionBinding) -> Result<bool, WorkGraphError> {
validate_projection(binding)?;
let mut authority =
execution_dsl::WorkExecutionLifecycleMachineAuthority::recover_from_state(
binding.machine_state.clone(),
)
.map_err(|error| {
WorkGraphError::InvalidTransition(format!(
"work execution {} refused retry classification recovery: {error:?}",
binding.binding_id
))
})?;
let transition = execution_dsl::WorkExecutionLifecycleMachineMutator::apply(
&mut authority,
execution_dsl::WorkExecutionLifecycleInput::ClassifyRetryEligibility {},
)
.map_err(|error| {
WorkGraphError::InvalidTransition(format!(
"work execution {} refused retry classification: {error:?}",
binding.binding_id
))
})?;
match exactly_one_effect(transition.effects())? {
WorkExecutionLifecycleEffect::RetryEligibilityClassified { eligible } => Ok(eligible),
effect => Err(WorkGraphError::Store(format!(
"work execution {} retry classifier emitted unexpected effect {effect:?}",
binding.binding_id
))),
}
}
}
fn validate_observation_detail(
observation: &WorkExecutionObservation,
) -> Result<(), WorkGraphError> {
const MAX_DETAIL_BYTES: usize = 4096;
let detail = match observation {
WorkExecutionObservation::FlowFailed { detail }
| WorkExecutionObservation::FlowCanceled { detail } => detail.as_deref(),
WorkExecutionObservation::FlowRunLost { detail }
| WorkExecutionObservation::LaunchUncertain { detail }
| WorkExecutionObservation::LaunchQuarantined { detail }
| WorkExecutionObservation::LaunchFailed { detail }
| WorkExecutionObservation::WorkClosureRefused { detail } => Some(detail.as_str()),
_ => None,
};
if let Some(detail) = detail
&& (detail.len() > MAX_DETAIL_BYTES || detail.chars().any(char::is_control))
{
return Err(WorkGraphError::InvalidInput(format!(
"work execution observation detail must be single-line text no longer than {MAX_DETAIL_BYTES} bytes"
)));
}
Ok(())
}
fn validate_handoff_obligation(
transition: &execution_dsl::WorkExecutionLifecycleMachineTransition,
) -> Result<(), WorkGraphError> {
let obligation_count = match transition.effects() {
[WorkExecutionLifecycleEffect::FlowLaunchRequested { .. }] => {
flow_launch_protocol::extract_obligations(transition).len()
}
[WorkExecutionLifecycleEffect::FlowLaunchAccepted { .. }] => {
flow_observation_protocol::extract_obligations(transition).len()
}
[WorkExecutionLifecycleEffect::FlowLaunchUncertain { .. }] => {
uncertain_launch_protocol::extract_obligations(transition).len()
}
[WorkExecutionLifecycleEffect::FlowLaunchQuarantined { .. }] => {
quarantined_launch_protocol::extract_obligations(transition).len()
}
[WorkExecutionLifecycleEffect::EvidenceProjectionRequested { .. }] => {
success_evidence_protocol::extract_obligations(transition).len()
}
[WorkExecutionLifecycleEffect::FlowFailureEvidenceProjectionRequested { .. }] => {
failure_evidence_protocol::extract_obligations(transition).len()
}
[WorkExecutionLifecycleEffect::FlowCancellationEvidenceProjectionRequested { .. }] => {
cancellation_evidence_protocol::extract_obligations(transition).len()
}
[WorkExecutionLifecycleEffect::LaunchFailureEvidenceProjectionRequested { .. }] => {
launch_failure_evidence_protocol::extract_obligations(transition).len()
}
[WorkExecutionLifecycleEffect::WorkClosureRequested { .. }] => {
work_closure_protocol::extract_obligations(transition).len()
}
_ => return Ok(()),
};
if obligation_count != 1 {
return Err(WorkGraphError::Store(format!(
"generated work execution transition projected {obligation_count} handoff obligations, expected exactly one"
)));
}
Ok(())
}
fn validate_projection(binding: &WorkExecutionBinding) -> Result<(), WorkGraphError> {
if binding.machine_state.binding_id != binding.binding_id.as_str() {
return Err(WorkGraphError::Store(format!(
"work execution {} machine binding identity does not match its projection",
binding.binding_id
)));
}
if binding.machine_state.run_id != binding.target.run_id() {
return Err(WorkGraphError::Store(format!(
"work execution {} machine run identity does not match its target",
binding.binding_id
)));
}
Ok(())
}
fn exactly_one_effect(
effects: &[WorkExecutionLifecycleEffect],
) -> Result<WorkExecutionLifecycleEffect, WorkGraphError> {
match effects {
[effect] => Ok(effect.clone()),
_ => Err(WorkGraphError::Store(format!(
"generated work execution transition emitted {} effects, expected exactly one",
effects.len()
))),
}
}
fn exactly_one_obligation<T>(
obligations: Vec<T>,
binding_id: &WorkExecutionBindingId,
) -> Result<T, WorkGraphError> {
let count = obligations.len();
obligations.into_iter().next().filter(|_| count == 1).ok_or_else(|| {
WorkGraphError::Store(format!(
"generated work execution handoff for {binding_id} projected {count} obligations, expected exactly one"
))
})
}
#[cfg(test)]
#[allow(clippy::expect_used)]
mod tests {
use super::*;
use crate::{WorkExecutionTarget, WorkItemId, WorkItemRef, WorkNamespace};
use chrono::Utc;
use serde_json::json;
fn binding() -> WorkExecutionBinding {
let binding_id = WorkExecutionBindingId::new("execution-machine-test").expect("id");
let target = WorkExecutionTarget::mob_flow(
"mob",
"flow",
format!("sha256:{}", "c".repeat(64)),
"1ae92ab4-8afe-5ad2-b9c3-fccae4f569a5",
crate::WorkExecutionAuthority::TargetOwner,
json!({}),
)
.expect("target");
let (machine_state, _) =
WorkExecutionMachine::bind(&binding_id, target.run_id()).expect("bind");
WorkExecutionBinding {
binding_id,
work_ref: WorkItemRef {
realm_id: "realm".to_string(),
namespace: WorkNamespace::default(),
item_id: WorkItemId::new("item").expect("item"),
},
target,
idempotency_key: "key".to_string(),
correlation_id: "229650c5-9372-53e9-9c3a-831638a47c77".to_string(),
supersedes: None,
machine_state,
created_at: Utc::now(),
}
}
#[test]
fn failed_flow_requires_evidence_projection_before_terminal_attempt() {
let binding = binding();
let (running, _) =
WorkExecutionMachine::observe(binding, 1, WorkExecutionObservation::FlowRunning)
.expect("running");
let (projecting, effect) = WorkExecutionMachine::observe(
running,
2,
WorkExecutionObservation::FlowFailed {
detail: Some("step failed".to_string()),
},
)
.expect("failure observed");
assert!(matches!(
effect,
WorkExecutionLifecycleEffect::FlowFailureEvidenceProjectionRequested { .. }
));
let (terminal, effect) = WorkExecutionMachine::observe(
projecting,
3,
WorkExecutionObservation::FlowFailureEvidenceProjected,
)
.expect("failure evidence projected");
assert!(matches!(
effect,
WorkExecutionLifecycleEffect::FlowFailed { .. }
));
assert!(matches!(
WorkExecutionMachine::recover_effect(&terminal).expect("recover terminal"),
WorkExecutionLifecycleEffect::FlowFailed { .. }
));
}
#[test]
fn uncertain_launch_recovers_as_uncertain_without_redrive_request() {
let binding = binding();
let (uncertain, effect) = WorkExecutionMachine::observe(
binding,
1,
WorkExecutionObservation::LaunchUncertain {
detail: "intent exists without run".to_string(),
},
)
.expect("uncertain");
assert!(matches!(
effect,
WorkExecutionLifecycleEffect::FlowLaunchUncertain { .. }
));
assert!(matches!(
WorkExecutionMachine::recover_effect(&uncertain).expect("recover uncertain"),
WorkExecutionLifecycleEffect::FlowLaunchUncertain { .. }
));
}
#[test]
fn observations_are_admitted_only_by_the_current_generated_handoff() {
let binding = binding();
let (accepted, _) =
WorkExecutionMachine::observe(binding, 1, WorkExecutionObservation::FlowStarted)
.expect("launch accepted");
let error =
WorkExecutionMachine::observe(accepted, 2, WorkExecutionObservation::FlowStarted)
.expect_err("flow-observation protocol does not admit a second start");
assert!(matches!(error, WorkGraphError::InvalidTransition(_)));
}
#[test]
fn quarantined_launch_accepts_only_observed_exact_run_feedback() {
let binding = binding();
let (quarantined, effect) = WorkExecutionMachine::observe(
binding,
1,
WorkExecutionObservation::LaunchQuarantined {
detail: "realizing ledger has no exact run".to_string(),
},
)
.expect("quarantine");
assert!(matches!(
effect,
WorkExecutionLifecycleEffect::FlowLaunchQuarantined { .. }
));
assert!(
!WorkExecutionMachine::retry_eligible(&quarantined)
.expect("classify quarantined launch"),
"quarantine may still have a live realizer and cannot authorize supersession"
);
let (_, effect) =
WorkExecutionMachine::observe(quarantined, 2, WorkExecutionObservation::FlowStarted)
.expect("exact run observation feedback");
assert!(matches!(
effect,
WorkExecutionLifecycleEffect::FlowLaunchAccepted { .. }
));
}
}