scientific_workflow/runtime/
execution_plan.rs

1//! Deterministic, read-only workflow plan export.
2
3use std::fs::{self, OpenOptions};
4use std::io::Write;
5use std::path::Path;
6
7use serde::Serialize;
8use serde_json::Value;
9
10use super::{Phase, RuntimeError, TaskDisplayKind};
11
12const EXECUTION_PLAN_FORMAT: &str = "scientific-workflow.execution-plan.v1";
13
14/// Complete serializable phase/task graph registered with one runtime.
15#[derive(Clone, Debug, PartialEq, Serialize)]
16pub struct ExecutionPlan {
17    format: &'static str,
18    phases: Vec<ExecutionPlanPhase>,
19}
20
21#[derive(Clone, Debug, PartialEq, Serialize)]
22struct ExecutionPlanPhase {
23    id: u64,
24    label: String,
25    registration_order: usize,
26    dependencies: Vec<u64>,
27    max_active_tasks: usize,
28    prepared_task_queue_capacity: usize,
29    #[serde(skip_serializing_if = "Option::is_none")]
30    delay_per_task_ns: Option<u128>,
31    #[serde(skip_serializing_if = "Option::is_none")]
32    task_timeout_ns: Option<u128>,
33    #[serde(skip_serializing_if = "Option::is_none")]
34    deadline_after_ns: Option<u128>,
35    failure_policy: &'static str,
36    requires_confirmation: bool,
37    tasks: Vec<ExecutionPlanTask>,
38}
39
40#[derive(Clone, Debug, PartialEq, Serialize)]
41struct ExecutionPlanTask {
42    id: String,
43    kind: String,
44    label: String,
45    registration_order: usize,
46    configuration_ordinal: u64,
47    display_kind: &'static str,
48    status: &'static str,
49    #[serde(skip_serializing_if = "Option::is_none")]
50    delay_rank: Option<usize>,
51    #[serde(skip_serializing_if = "Option::is_none")]
52    release_offset_ns: Option<u128>,
53    #[serde(skip_serializing_if = "Option::is_none")]
54    display_parameters: Option<Vec<String>>,
55    configuration: Value,
56}
57
58impl ExecutionPlan {
59    pub(crate) fn from_phases(phases: &[Phase]) -> Self {
60        Self {
61            format: EXECUTION_PLAN_FORMAT,
62            phases: phases
63                .iter()
64                .enumerate()
65                .map(|(registration_order, phase)| {
66                    let mut executable_rank = 0_usize;
67                    let tasks = phase
68                        .tasks()
69                        .iter()
70                        .enumerate()
71                        .map(|(task_order, task)| {
72                            let delay_rank = (!task.is_reused()).then(|| {
73                                let rank = executable_rank;
74                                executable_rank += 1;
75                                rank
76                            });
77                            let release_offset_ns = phase.delay_per_task().and_then(|delay| {
78                                delay_rank.map(|rank| delay.as_nanos().saturating_mul(rank as u128))
79                            });
80                            ExecutionPlanTask {
81                                id: task.id().to_string(),
82                                kind: task.kind().to_owned(),
83                                label: task.label().to_owned(),
84                                registration_order: task_order,
85                                configuration_ordinal: task.configuration_ordinal(),
86                                display_kind: match task.display_kind() {
87                                    TaskDisplayKind::Progress => "progress",
88                                    TaskDisplayKind::Activity => "activity",
89                                },
90                                status: if task.is_reused() {
91                                    "reused"
92                                } else {
93                                    "pending"
94                                },
95                                delay_rank,
96                                release_offset_ns,
97                                display_parameters: task
98                                    .display_keys()
99                                    .map(|keys| keys.iter().map(|key| key.to_string()).collect()),
100                                configuration: task.configuration().resolved_json(),
101                            }
102                        })
103                        .collect();
104                    ExecutionPlanPhase {
105                        id: phase.id().get(),
106                        label: phase.label().to_owned(),
107                        registration_order,
108                        dependencies: phase
109                            .dependencies()
110                            .iter()
111                            .map(|dependency| dependency.get())
112                            .collect(),
113                        max_active_tasks: phase.max_active_tasks(),
114                        prepared_task_queue_capacity: phase.prepared_task_queue_capacity(),
115                        delay_per_task_ns: phase.delay_per_task().map(|value| value.as_nanos()),
116                        task_timeout_ns: phase.task_timeout().map(|value| value.as_nanos()),
117                        deadline_after_ns: phase.deadline_after().map(|value| value.as_nanos()),
118                        failure_policy: phase.failure_policy().as_str(),
119                        requires_confirmation: phase.requires_confirmation(),
120                        tasks,
121                    }
122                })
123                .collect(),
124        }
125    }
126
127    /// Serializes the plan as deterministic pretty JSON with a final newline.
128    pub fn to_pretty_json(&self) -> Result<Vec<u8>, RuntimeError> {
129        let mut bytes = serde_json::to_vec_pretty(self)
130            .map_err(|source| RuntimeError::SerializeExecutionPlan { source })?;
131        bytes.push(b'\n');
132        Ok(bytes)
133    }
134
135    /// Writes the plan without overwriting different existing content.
136    pub fn write_json(&self, path: impl AsRef<Path>) -> Result<(), RuntimeError> {
137        let path = path.as_ref();
138        let bytes = self.to_pretty_json()?;
139        match OpenOptions::new().write(true).create_new(true).open(path) {
140            Ok(mut file) => file
141                .write_all(&bytes)
142                .and_then(|()| file.sync_all())
143                .map_err(|source| RuntimeError::WriteExecutionPlan {
144                    path: path.to_path_buf(),
145                    source,
146                }),
147            Err(source) if source.kind() == std::io::ErrorKind::AlreadyExists => {
148                let existing =
149                    fs::read(path).map_err(|source| RuntimeError::WriteExecutionPlan {
150                        path: path.to_path_buf(),
151                        source,
152                    })?;
153                if existing == bytes {
154                    Ok(())
155                } else {
156                    Err(RuntimeError::ExecutionPlanConflict {
157                        path: path.to_path_buf(),
158                    })
159                }
160            }
161            Err(source) => Err(RuntimeError::WriteExecutionPlan {
162                path: path.to_path_buf(),
163                source,
164            }),
165        }
166    }
167}
scientific_workflow/runtime/execution_plan.rs

scientific_workflow/runtime/
execution_plan.rs
