scientific_workflow/study/
record.rs

1//! Always-on, compact study lifecycle recording.
2
3use std::collections::HashMap;
4use std::fs::{self, File, OpenOptions};
5use std::io::Write;
6use std::path::{Path, PathBuf};
7use std::sync::{Arc, Mutex, MutexGuard};
8use std::time::Instant;
9
10use serde::Serialize;
11
12use super::error::StudyError;
13use super::phase::{Phase, PhaseId, TaskKey, TaskMode};
14use super::renderer::{ProgressSummary, TaskExecutionSnapshot};
15use crate::clock::{duration_nanoseconds, utc_now_rfc3339};
16
17const STUDY_RECORD_FORMAT: &str = "scientific-workflow.study-record.v1";
18
19/// Durable lifecycle summary for one study execution.
20#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
21pub struct StudyRecord {
22    format: &'static str,
23    status: &'static str,
24    started_at_utc: String,
25    #[serde(skip_serializing_if = "Option::is_none")]
26    ended_at_utc: Option<String>,
27    #[serde(skip_serializing_if = "Option::is_none")]
28    duration_ns: Option<u64>,
29    phase_count: usize,
30    task_count: usize,
31    phases: Vec<PhaseRecord>,
32}
33
34/// Lifecycle facts for one selected phase.
35#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
36pub struct PhaseRecord {
37    id: u64,
38    label: String,
39    status: &'static str,
40    #[serde(skip_serializing_if = "Option::is_none")]
41    started_at_utc: Option<String>,
42    #[serde(skip_serializing_if = "Option::is_none")]
43    ended_at_utc: Option<String>,
44    #[serde(skip_serializing_if = "Option::is_none")]
45    duration_ns: Option<u64>,
46    progress: TaskCounts,
47    tasks: Vec<TaskRecord>,
48}
49
50/// Lifecycle facts for one selected task.
51#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
52pub struct TaskRecord {
53    id: String,
54    category: String,
55    label: String,
56    mode: &'static str,
57    status: &'static str,
58    #[serde(skip_serializing_if = "Option::is_none")]
59    started_at_utc: Option<String>,
60    #[serde(skip_serializing_if = "Option::is_none")]
61    ended_at_utc: Option<String>,
62    #[serde(skip_serializing_if = "Option::is_none")]
63    duration_ns: Option<u64>,
64    #[serde(skip_serializing_if = "Option::is_none")]
65    start_offset_ns: Option<u64>,
66    #[serde(skip_serializing_if = "Option::is_none")]
67    final_iteration: Option<u64>,
68    #[serde(skip_serializing_if = "Option::is_none")]
69    target_iteration: Option<u64>,
70}
71
72#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize)]
73struct TaskCounts {
74    total: u64,
75    completed: u64,
76    failed: u64,
77    cancelled: u64,
78    skipped: u64,
79}
80
81pub(crate) struct StudyRecorder {
82    inner: Arc<Mutex<RecorderState>>,
83}
84
85struct RecorderState {
86    path: PathBuf,
87    record: StudyRecord,
88    study_started: Instant,
89    phase_positions: HashMap<PhaseId, usize>,
90    task_positions: HashMap<TaskKey, (usize, usize)>,
91    phase_started: HashMap<PhaseId, Instant>,
92    finished: bool,
93}
94
95pub(crate) struct TaskTimer {
96    recorder: Arc<Mutex<RecorderState>>,
97    key: TaskKey,
98    started: Instant,
99}
100
101impl StudyRecorder {
102    pub(crate) fn start(path: PathBuf, phases: &[&Phase]) -> Result<Self, StudyError> {
103        let started_at_utc = timestamp("start study execution record")?;
104        let study_started = Instant::now();
105        let mut phase_positions = HashMap::with_capacity(phases.len());
106        let mut task_positions = HashMap::new();
107        let mut phase_records = Vec::with_capacity(phases.len());
108        let mut task_count = 0_usize;
109        for (phase_position, phase) in phases.iter().enumerate() {
110            phase_positions.insert(phase.id(), phase_position);
111            let mut tasks = Vec::with_capacity(phase.tasks().len());
112            for (task_position, task) in phase.tasks().iter().enumerate() {
113                task_positions.insert(task.key().clone(), (phase_position, task_position));
114                tasks.push(TaskRecord {
115                    id: task.id().to_string(),
116                    category: task.category_name().to_owned(),
117                    label: task.label().to_owned(),
118                    mode: match task.mode() {
119                        TaskMode::Progress => "progress",
120                        TaskMode::OneShot => "one-shot",
121                    },
122                    status: "pending",
123                    started_at_utc: None,
124                    ended_at_utc: None,
125                    duration_ns: None,
126                    start_offset_ns: None,
127                    final_iteration: None,
128                    target_iteration: None,
129                });
130            }
131            task_count += tasks.len();
132            phase_records.push(PhaseRecord {
133                id: phase.id().get(),
134                label: phase.label().to_owned(),
135                status: "pending",
136                started_at_utc: None,
137                ended_at_utc: None,
138                duration_ns: None,
139                progress: TaskCounts::default(),
140                tasks,
141            });
142        }
143        let recorder = Self {
144            inner: Arc::new(Mutex::new(RecorderState {
145                path,
146                record: StudyRecord {
147                    format: STUDY_RECORD_FORMAT,
148                    status: "running",
149                    started_at_utc,
150                    ended_at_utc: None,
151                    duration_ns: None,
152                    phase_count: phases.len(),
153                    task_count,
154                    phases: phase_records,
155                },
156                study_started,
157                phase_positions,
158                task_positions,
159                phase_started: HashMap::with_capacity(phases.len()),
160                finished: false,
161            })),
162        };
163        recorder.persist()?;
164        Ok(recorder)
165    }
166
167    pub(crate) fn phase_started(&self, id: PhaseId) -> Result<(), StudyError> {
168        let timestamp = timestamp("record phase start")?;
169        let mut state = lock(&self.inner);
170        let position = state.phase_positions[&id];
171        state.record.phases[position].status = "running";
172        state.record.phases[position].started_at_utc = Some(timestamp);
173        state.phase_started.insert(id, Instant::now());
174        persist_state(&state)
175    }
176
177    pub(crate) fn task_started(&self, key: &TaskKey) -> Result<TaskTimer, StudyError> {
178        let timestamp = timestamp("record task start")?;
179        let started = Instant::now();
180        let mut state = lock(&self.inner);
181        let (phase_position, task_position) = state.task_positions[key];
182        let phase_id = key.phase_id();
183        let offset = state
184            .phase_started
185            .get(&phase_id)
186            .map(|phase_started| nanoseconds(phase_started.elapsed()));
187        let task = &mut state.record.phases[phase_position].tasks[task_position];
188        task.status = "running";
189        task.started_at_utc = Some(timestamp);
190        task.start_offset_ns = offset;
191        Ok(TaskTimer {
192            recorder: Arc::clone(&self.inner),
193            key: key.clone(),
194            started,
195        })
196    }
197
198    pub(crate) fn phase_finished(
199        &self,
200        id: PhaseId,
201        success: bool,
202        progress: &ProgressSummary,
203        tasks: Vec<TaskExecutionSnapshot>,
204    ) -> Result<(), StudyError> {
205        let ended_at_utc = timestamp("record phase completion")?;
206        let mut state = lock(&self.inner);
207        for snapshot in tasks {
208            let (phase_position, task_position) = state.task_positions[&snapshot.key];
209            let task = &mut state.record.phases[phase_position].tasks[task_position];
210            task.status = snapshot.status.as_str();
211            task.final_iteration = snapshot.current_iteration;
212            task.target_iteration = snapshot.target_iteration;
213        }
214        let position = state.phase_positions[&id];
215        let duration_ns = state
216            .phase_started
217            .remove(&id)
218            .map(|start| nanoseconds(start.elapsed()));
219        let phase = &mut state.record.phases[position];
220        phase.status = if success { "completed" } else { "failed" };
221        phase.ended_at_utc = Some(ended_at_utc);
222        phase.duration_ns = duration_ns;
223        phase.progress = counts(progress);
224        persist_state(&state)
225    }
226
227    pub(crate) fn finish(&self, success: bool) -> Result<StudyRecord, StudyError> {
228        let ended_at_utc = timestamp("finish study execution record")?;
229        let mut state = lock(&self.inner);
230        state.record.status = if success { "completed" } else { "failed" };
231        state.record.ended_at_utc = Some(ended_at_utc);
232        state.record.duration_ns = Some(nanoseconds(state.study_started.elapsed()));
233        state.finished = true;
234        persist_state(&state)?;
235        Ok(state.record.clone())
236    }
237
238    fn persist(&self) -> Result<(), StudyError> {
239        persist_state(&lock(&self.inner))
240    }
241}
242
243impl Drop for StudyRecorder {
244    fn drop(&mut self) {
245        let mut state = lock(&self.inner);
246        if state.finished {
247            return;
248        }
249        state.record.status = "failed";
250        state.record.ended_at_utc = utc_now_rfc3339().ok();
251        state.record.duration_ns = Some(nanoseconds(state.study_started.elapsed()));
252        let _ = persist_state(&state);
253        state.finished = true;
254    }
255}
256
257impl Drop for TaskTimer {
258    fn drop(&mut self) {
259        let ended_at_utc = utc_now_rfc3339().ok();
260        let mut state = lock(&self.recorder);
261        let (phase_position, task_position) = state.task_positions[&self.key];
262        let task = &mut state.record.phases[phase_position].tasks[task_position];
263        task.ended_at_utc = ended_at_utc;
264        task.duration_ns = Some(nanoseconds(self.started.elapsed()));
265    }
266}
267
268fn counts(summary: &ProgressSummary) -> TaskCounts {
269    TaskCounts {
270        total: summary.total(),
271        completed: summary.completed(),
272        failed: summary.failed(),
273        cancelled: summary.cancelled(),
274        skipped: summary.skipped(),
275    }
276}
277
278fn timestamp(operation: &'static str) -> Result<String, StudyError> {
279    utc_now_rfc3339().map_err(|source| StudyError::StudyRecordTimestamp { operation, source })
280}
281
282fn nanoseconds(duration: std::time::Duration) -> u64 {
283    duration_nanoseconds(duration).unwrap_or(u64::MAX)
284}
285
286fn persist_state(state: &RecorderState) -> Result<(), StudyError> {
287    let path = &state.path;
288    if let Some(parent) = path
289        .parent()
290        .filter(|parent| !parent.as_os_str().is_empty())
291    {
292        fs::create_dir_all(parent).map_err(|source| StudyError::WriteStudyRecord {
293            path: path.clone(),
294            source,
295        })?;
296    }
297    let mut bytes = serde_json::to_vec_pretty(&state.record)
298        .map_err(|source| StudyError::SerializeStudyRecord { source })?;
299    bytes.push(b'\n');
300    let temporary = temporary_path(path);
301    let mut file = OpenOptions::new()
302        .write(true)
303        .create(true)
304        .truncate(true)
305        .open(&temporary)
306        .map_err(|source| StudyError::WriteStudyRecord {
307            path: temporary.clone(),
308            source,
309        })?;
310    file.write_all(&bytes)
311        .and_then(|()| file.sync_all())
312        .map_err(|source| StudyError::WriteStudyRecord {
313            path: temporary.clone(),
314            source,
315        })?;
316    fs::rename(&temporary, path).map_err(|source| StudyError::WriteStudyRecord {
317        path: path.clone(),
318        source,
319    })?;
320    if let Some(parent) = path
321        .parent()
322        .filter(|parent| !parent.as_os_str().is_empty())
323    {
324        File::open(parent)
325            .and_then(|directory| directory.sync_all())
326            .map_err(|source| StudyError::WriteStudyRecord {
327                path: parent.to_path_buf(),
328                source,
329            })?;
330    }
331    Ok(())
332}
333
334fn temporary_path(path: &Path) -> PathBuf {
335    let name = path
336        .file_name()
337        .and_then(|name| name.to_str())
338        .unwrap_or("study-record.json");
339    path.with_file_name(format!(".{name}.tmp"))
340}
341
342fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
343    mutex
344        .lock()
345        .unwrap_or_else(|poisoned| poisoned.into_inner())
346}
scientific_workflow/study/record.rs

scientific_workflow/study/
record.rs
