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

scientific_workflow/study/
record.rs
