scientific_workflow/runtime/
execution_record.rs

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

scientific_workflow/runtime/
execution_record.rs
