oximedia_workflow/
workflow_log.rs

1//! Structured workflow event log for `oximedia-workflow`.
2//!
3//! [`WorkflowLog`] stores an in-memory sequence of [`WorkflowEvent`]s and
4//! provides filtering helpers such as [`WorkflowLog::recent_errors`] so that
5//! operators can quickly surface problems without trawling the full history.
6
7#![allow(dead_code)]
8
9use serde::{Deserialize, Serialize};
10use std::time::{Duration, SystemTime, UNIX_EPOCH};
11
12// ── Event type ────────────────────────────────────────────────────────────────
13
14/// Discriminates the kind of event stored in a [`WorkflowEvent`].
15#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
16pub enum WorkflowEventType {
17    /// The workflow was submitted to the execution queue.
18    Submitted,
19    /// A task within the workflow started executing.
20    TaskStarted,
21    /// A task completed successfully.
22    TaskCompleted,
23    /// A task failed and will (or will not) be retried.
24    TaskFailed,
25    /// The workflow completed without any failures.
26    WorkflowCompleted,
27    /// The workflow was cancelled before completion.
28    WorkflowCancelled,
29    /// A non-fatal warning was recorded (e.g. resource limit approached).
30    Warning,
31    /// A fatal error occurred; the workflow cannot continue.
32    Error,
33}
34
35impl WorkflowEventType {
36    /// Returns `true` when this event type indicates a failure or error.
37    #[must_use]
38    pub fn is_error(self) -> bool {
39        matches!(self, Self::TaskFailed | Self::Error)
40    }
41
42    /// Returns `true` when this event type signals successful completion.
43    #[must_use]
44    pub fn is_success(self) -> bool {
45        matches!(self, Self::TaskCompleted | Self::WorkflowCompleted)
46    }
47
48    /// Returns a short label suitable for log output.
49    #[must_use]
50    pub fn label(self) -> &'static str {
51        match self {
52            Self::Submitted => "SUBMITTED",
53            Self::TaskStarted => "TASK_START",
54            Self::TaskCompleted => "TASK_OK",
55            Self::TaskFailed => "TASK_FAIL",
56            Self::WorkflowCompleted => "WF_OK",
57            Self::WorkflowCancelled => "WF_CANCEL",
58            Self::Warning => "WARN",
59            Self::Error => "ERROR",
60        }
61    }
62}
63
64impl std::fmt::Display for WorkflowEventType {
65    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
66        write!(f, "{}", self.label())
67    }
68}
69
70// ── Event ─────────────────────────────────────────────────────────────────────
71
72/// A single structured log entry within a [`WorkflowLog`].
73#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
74pub struct WorkflowEvent {
75    /// Monotonically increasing sequence number within the log.
76    pub seq: u64,
77    /// Seconds since the UNIX epoch when this event was recorded.
78    pub timestamp_secs: u64,
79    /// The kind of event.
80    pub event_type: WorkflowEventType,
81    /// Identifier of the workflow this event belongs to.
82    pub workflow_id: String,
83    /// Optional identifier of the specific task (if task-scoped).
84    pub task_id: Option<String>,
85    /// Human-readable message providing additional context.
86    pub message: String,
87}
88
89impl WorkflowEvent {
90    /// Creates a workflow-level event using the current system time.
91    #[must_use]
92    pub fn workflow_event(
93        seq: u64,
94        event_type: WorkflowEventType,
95        workflow_id: impl Into<String>,
96        message: impl Into<String>,
97    ) -> Self {
98        Self {
99            seq,
100            timestamp_secs: now_secs(),
101            event_type,
102            workflow_id: workflow_id.into(),
103            task_id: None,
104            message: message.into(),
105        }
106    }
107
108    /// Creates a task-scoped event using the current system time.
109    #[must_use]
110    pub fn task_event(
111        seq: u64,
112        event_type: WorkflowEventType,
113        workflow_id: impl Into<String>,
114        task_id: impl Into<String>,
115        message: impl Into<String>,
116    ) -> Self {
117        Self {
118            seq,
119            timestamp_secs: now_secs(),
120            event_type,
121            workflow_id: workflow_id.into(),
122            task_id: Some(task_id.into()),
123            message: message.into(),
124        }
125    }
126
127    /// Returns `true` when this event represents a failure or error.
128    #[must_use]
129    pub fn is_error(&self) -> bool {
130        self.event_type.is_error()
131    }
132}
133
134/// Returns seconds since UNIX epoch (best-effort; returns 0 on overflow).
135fn now_secs() -> u64 {
136    SystemTime::now()
137        .duration_since(UNIX_EPOCH)
138        .unwrap_or(Duration::ZERO)
139        .as_secs()
140}
141
142// ── Log ───────────────────────────────────────────────────────────────────────
143
144/// An in-memory log that accumulates [`WorkflowEvent`]s for one or more
145/// workflows and exposes query helpers.
146#[derive(Debug, Default, Clone)]
147pub struct WorkflowLog {
148    events: Vec<WorkflowEvent>,
149    next_seq: u64,
150}
151
152impl WorkflowLog {
153    /// Creates a new, empty log.
154    #[must_use]
155    pub fn new() -> Self {
156        Self::default()
157    }
158
159    /// Appends an event to the log, assigning it the next sequence number.
160    pub fn append(
161        &mut self,
162        event_type: WorkflowEventType,
163        workflow_id: impl Into<String>,
164        message: impl Into<String>,
165    ) {
166        let ev = WorkflowEvent::workflow_event(self.next_seq, event_type, workflow_id, message);
167        self.next_seq += 1;
168        self.events.push(ev);
169    }
170
171    /// Appends a task-scoped event to the log.
172    pub fn append_task(
173        &mut self,
174        event_type: WorkflowEventType,
175        workflow_id: impl Into<String>,
176        task_id: impl Into<String>,
177        message: impl Into<String>,
178    ) {
179        let ev =
180            WorkflowEvent::task_event(self.next_seq, event_type, workflow_id, task_id, message);
181        self.next_seq += 1;
182        self.events.push(ev);
183    }
184
185    /// Total number of events in the log.
186    #[must_use]
187    pub fn len(&self) -> usize {
188        self.events.len()
189    }
190
191    /// Returns `true` when the log contains no events.
192    #[must_use]
193    pub fn is_empty(&self) -> bool {
194        self.events.is_empty()
195    }
196
197    /// Returns a slice of all events in insertion order.
198    #[must_use]
199    pub fn all(&self) -> &[WorkflowEvent] {
200        &self.events
201    }
202
203    /// Returns the `n` most recently appended error events (`TaskFailed` or
204    /// Error), newest first.
205    #[must_use]
206    pub fn recent_errors(&self, n: usize) -> Vec<&WorkflowEvent> {
207        self.events
208            .iter()
209            .rev()
210            .filter(|e| e.is_error())
211            .take(n)
212            .collect()
213    }
214
215    /// Returns all events for the given workflow ID.
216    #[must_use]
217    pub fn events_for_workflow<'a>(&'a self, workflow_id: &str) -> Vec<&'a WorkflowEvent> {
218        self.events
219            .iter()
220            .filter(|e| e.workflow_id == workflow_id)
221            .collect()
222    }
223
224    /// Returns all events matching the given event type.
225    #[must_use]
226    pub fn events_of_type(&self, event_type: WorkflowEventType) -> Vec<&WorkflowEvent> {
227        self.events
228            .iter()
229            .filter(|e| e.event_type == event_type)
230            .collect()
231    }
232
233    /// Returns the number of error-type events in the log.
234    #[must_use]
235    pub fn error_count(&self) -> usize {
236        self.events.iter().filter(|e| e.is_error()).count()
237    }
238
239    /// Clears all events from the log and resets the sequence counter.
240    pub fn clear(&mut self) {
241        self.events.clear();
242        self.next_seq = 0;
243    }
244}
245
246// ── Tests ─────────────────────────────────────────────────────────────────────
247
248#[cfg(test)]
249mod tests {
250    use super::*;
251
252    fn make_log() -> WorkflowLog {
253        let mut log = WorkflowLog::new();
254        log.append(WorkflowEventType::Submitted, "wf-1", "Workflow submitted");
255        log.append_task(
256            WorkflowEventType::TaskStarted,
257            "wf-1",
258            "task-a",
259            "Started task-a",
260        );
261        log.append_task(
262            WorkflowEventType::TaskFailed,
263            "wf-1",
264            "task-a",
265            "task-a failed: timeout",
266        );
267        log.append(WorkflowEventType::Error, "wf-1", "Unrecoverable error");
268        log.append_task(
269            WorkflowEventType::TaskCompleted,
270            "wf-2",
271            "task-b",
272            "task-b done",
273        );
274        log
275    }
276
277    #[test]
278    fn test_new_log_empty() {
279        let log = WorkflowLog::new();
280        assert!(log.is_empty());
281        assert_eq!(log.len(), 0);
282    }
283
284    #[test]
285    fn test_append_increments_len() {
286        let mut log = WorkflowLog::new();
287        log.append(WorkflowEventType::Submitted, "wf-x", "msg");
288        assert_eq!(log.len(), 1);
289    }
290
291    #[test]
292    fn test_sequence_numbers_monotonic() {
293        let log = make_log();
294        for (i, ev) in log.all().iter().enumerate() {
295            assert_eq!(ev.seq, i as u64);
296        }
297    }
298
299    #[test]
300    fn test_recent_errors_count() {
301        let log = make_log();
302        // Two error events: TaskFailed and Error
303        let errors = log.recent_errors(10);
304        assert_eq!(errors.len(), 2);
305    }
306
307    #[test]
308    fn test_recent_errors_newest_first() {
309        let log = make_log();
310        let errors = log.recent_errors(10);
311        // The Error event was appended after TaskFailed, so it has higher seq
312        assert!(errors[0].seq > errors[1].seq);
313    }
314
315    #[test]
316    fn test_recent_errors_limit() {
317        let log = make_log();
318        let errors = log.recent_errors(1);
319        assert_eq!(errors.len(), 1);
320    }
321
322    #[test]
323    fn test_events_for_workflow() {
324        let log = make_log();
325        let wf1 = log.events_for_workflow("wf-1");
326        assert_eq!(wf1.len(), 4);
327        let wf2 = log.events_for_workflow("wf-2");
328        assert_eq!(wf2.len(), 1);
329    }
330
331    #[test]
332    fn test_events_of_type() {
333        let log = make_log();
334        let starts = log.events_of_type(WorkflowEventType::TaskStarted);
335        assert_eq!(starts.len(), 1);
336    }
337
338    #[test]
339    fn test_error_count() {
340        let log = make_log();
341        assert_eq!(log.error_count(), 2);
342    }
343
344    #[test]
345    fn test_clear_resets_log() {
346        let mut log = make_log();
347        log.clear();
348        assert!(log.is_empty());
349        assert_eq!(log.error_count(), 0);
350    }
351
352    #[test]
353    fn test_clear_resets_sequence() {
354        let mut log = make_log();
355        log.clear();
356        log.append(WorkflowEventType::Submitted, "wf-new", "msg");
357        assert_eq!(log.all()[0].seq, 0);
358    }
359
360    #[test]
361    fn test_event_type_is_error() {
362        assert!(WorkflowEventType::TaskFailed.is_error());
363        assert!(WorkflowEventType::Error.is_error());
364        assert!(!WorkflowEventType::TaskCompleted.is_error());
365        assert!(!WorkflowEventType::Submitted.is_error());
366    }
367
368    #[test]
369    fn test_event_type_is_success() {
370        assert!(WorkflowEventType::TaskCompleted.is_success());
371        assert!(WorkflowEventType::WorkflowCompleted.is_success());
372        assert!(!WorkflowEventType::TaskFailed.is_success());
373    }
374
375    #[test]
376    fn test_event_type_label() {
377        assert_eq!(WorkflowEventType::Error.label(), "ERROR");
378        assert_eq!(WorkflowEventType::Submitted.label(), "SUBMITTED");
379    }
380
381    #[test]
382    fn test_event_type_display() {
383        assert_eq!(format!("{}", WorkflowEventType::Warning), "WARN");
384    }
385
386    #[test]
387    fn test_task_event_has_task_id() {
388        let log = make_log();
389        let task_ev = log
390            .events_of_type(WorkflowEventType::TaskStarted)
391            .into_iter()
392            .next()
393            .expect("should succeed in test");
394        assert_eq!(task_ev.task_id.as_deref(), Some("task-a"));
395    }
396
397    #[test]
398    fn test_workflow_event_no_task_id() {
399        let log = make_log();
400        let submitted = log
401            .events_of_type(WorkflowEventType::Submitted)
402            .into_iter()
403            .next()
404            .expect("should succeed in test");
405        assert!(submitted.task_id.is_none());
406    }
407}
oximedia_workflow/workflow_log.rs

oximedia_workflow/
workflow_log.rs
