scientific_workflow/study/
plan.rs

1//! Deterministic, read-only study 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, StudyError, TaskMode};
11
12const STUDY_PLAN_FORMAT: &str = "scientific-workflow.study-plan.v1";
13
14/// Complete serializable phase/task graph registered with one study.
15#[derive(Clone, Debug, PartialEq, Serialize)]
16pub struct StudyPlan {
17    format: &'static str,
18    phases: Vec<StudyPlanPhase>,
19}
20
21#[derive(Clone, Debug, PartialEq, Serialize)]
22struct StudyPlanPhase {
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<StudyPlanTask>,
38}
39
40#[derive(Clone, Debug, PartialEq, Serialize)]
41struct StudyPlanTask {
42    id: String,
43    category: String,
44    label: String,
45    registration_order: usize,
46    mode: &'static str,
47    status: &'static str,
48    #[serde(skip_serializing_if = "Option::is_none")]
49    delay_rank: Option<usize>,
50    #[serde(skip_serializing_if = "Option::is_none")]
51    release_offset_ns: Option<u128>,
52    metadata: Value,
53}
54
55impl StudyPlan {
56    pub(crate) fn from_phases(phases: &[Phase]) -> Self {
57        Self {
58            format: STUDY_PLAN_FORMAT,
59            phases: phases
60                .iter()
61                .enumerate()
62                .map(|(registration_order, phase)| {
63                    let mut executable_rank = 0_usize;
64                    let tasks = phase
65                        .tasks()
66                        .iter()
67                        .enumerate()
68                        .map(|(task_order, task)| {
69                            let delay_rank = (!task.is_completed()).then(|| {
70                                let rank = executable_rank;
71                                executable_rank += 1;
72                                rank
73                            });
74                            let release_offset_ns = phase.delay_per_task().and_then(|delay| {
75                                delay_rank.map(|rank| delay.as_nanos().saturating_mul(rank as u128))
76                            });
77                            StudyPlanTask {
78                                id: task.id().to_string(),
79                                category: task.category_name().to_owned(),
80                                label: task.label().to_owned(),
81                                registration_order: task_order,
82                                mode: match task.mode() {
83                                    TaskMode::Progress => "progress",
84                                    TaskMode::OneShot => "one-shot",
85                                },
86                                status: if task.is_completed() {
87                                    "completed"
88                                } else {
89                                    "pending"
90                                },
91                                delay_rank,
92                                release_offset_ns,
93                                metadata: Value::Object(
94                                    task.metadata_iter()
95                                        .map(|(key, value)| (key.to_owned(), value.clone()))
96                                        .collect(),
97                                ),
98                            }
99                        })
100                        .collect();
101                    StudyPlanPhase {
102                        id: phase.id().get(),
103                        label: phase.label().to_owned(),
104                        registration_order,
105                        dependencies: phase
106                            .dependencies()
107                            .iter()
108                            .map(|dependency| dependency.get())
109                            .collect(),
110                        max_active_tasks: phase.max_active_tasks(),
111                        prepared_task_queue_capacity: phase.prepared_task_queue_capacity(),
112                        delay_per_task_ns: phase.delay_per_task().map(|value| value.as_nanos()),
113                        task_timeout_ns: phase.task_timeout().map(|value| value.as_nanos()),
114                        deadline_after_ns: phase.deadline_after().map(|value| value.as_nanos()),
115                        failure_policy: phase.failure_policy().as_str(),
116                        requires_confirmation: phase.requires_confirmation(),
117                        tasks,
118                    }
119                })
120                .collect(),
121        }
122    }
123
124    /// Serializes the plan as deterministic pretty JSON with a final newline.
125    pub fn to_pretty_json(&self) -> Result<Vec<u8>, StudyError> {
126        let mut bytes = serde_json::to_vec_pretty(self)
127            .map_err(|source| StudyError::SerializeStudyPlan { source })?;
128        bytes.push(b'\n');
129        Ok(bytes)
130    }
131
132    /// Writes the plan without overwriting different existing content.
133    pub fn write_json(&self, path: impl AsRef<Path>) -> Result<(), StudyError> {
134        let path = path.as_ref();
135        let bytes = self.to_pretty_json()?;
136        match OpenOptions::new().write(true).create_new(true).open(path) {
137            Ok(mut file) => file
138                .write_all(&bytes)
139                .and_then(|()| file.sync_all())
140                .map_err(|source| StudyError::WriteStudyPlan {
141                    path: path.to_path_buf(),
142                    source,
143                }),
144            Err(source) if source.kind() == std::io::ErrorKind::AlreadyExists => {
145                let existing = fs::read(path).map_err(|source| StudyError::WriteStudyPlan {
146                    path: path.to_path_buf(),
147                    source,
148                })?;
149                if existing == bytes {
150                    Ok(())
151                } else {
152                    Err(StudyError::StudyPlanConflict {
153                        path: path.to_path_buf(),
154                    })
155                }
156            }
157            Err(source) => Err(StudyError::WriteStudyPlan {
158                path: path.to_path_buf(),
159                source,
160            }),
161        }
162    }
163}
scientific_workflow/study/plan.rs

scientific_workflow/study/
plan.rs
