dataflow_rs/engine/observer.rs
1//! # Execution Observer Module
2//!
3//! An always-on, per-task callback for aggregation — counters, histograms,
4//! spans — as distinct from [`crate::ExecutionTrace`], which is a per-request
5//! allocation you persist.
6//!
7//! This exists because the eight sync built-ins (`map`, `validation`/`validate`,
8//! `parse_json`, `parse_xml`, `publish_json`, `publish_xml`, `filter`, `log`)
9//! are dispatched inside a private method on the workflow executor and never
10//! reach the function registry. A host can time its own registered handlers by
11//! wrapping their bodies, but it cannot time those eight at any price, and so
12//! cannot tell how much of a message's wall clock was spent inside the engine
13//! versus inside its own handlers.
14
15use core::time::Duration;
16
17/// One finished task.
18///
19/// Borrowed for the duration of the callback; an observer must copy out anything
20/// it needs to keep.
21#[derive(Debug)]
22pub struct TaskEvent<'a> {
23 /// `Workflow::id` of the workflow the task belongs to.
24 pub workflow_id: &'a str,
25 /// `Task::id`.
26 pub task_id: &'a str,
27 /// The function name — one of the built-in names, or a `Custom` handler's
28 /// registered name.
29 ///
30 /// Note this reports `"validate"` for both `validation` and `validate`
31 /// configs: they share a single `FunctionConfig::Validation` variant, and
32 /// this is that variant's canonical name.
33 pub function: &'a str,
34 /// `TaskOutcome::audit_status()` for a successful dispatch, `Some(500)` when
35 /// the task returned `Err`.
36 ///
37 /// `None` means the handler returned `TaskOutcome::Skip` — the body ran, but
38 /// no audit entry was recorded for it.
39 pub status: Option<u16>,
40 /// Wall-clock duration of the task **body only**: the dispatch call, not the
41 /// condition evaluation, the audit-trail push, or the `metadata.progress`
42 /// write.
43 ///
44 /// Derived from two `Utc::now()` reads rather than a monotonic clock, because
45 /// `std::time::Instant::now()` panics on `wasm32-unknown-unknown` and the
46 /// wasm bindings route through these instrumentation points. A backward clock
47 /// step clamps to zero rather than wrapping.
48 pub duration: Duration,
49}
50
51/// A message run beginning.
52#[derive(Debug)]
53#[non_exhaustive]
54pub struct MessageStarted<'a> {
55 /// The message's id.
56 pub message_id: &'a str,
57 /// How many workflows are about to be considered. Not how many will run —
58 /// conditions and rollout gates have not been evaluated yet.
59 ///
60 /// How many actually ran is deliberately absent here and from
61 /// [`MessageFinished`]: it is exactly the number of
62 /// [`ExecutionObserver::workflow_started`] callbacks between the two, so
63 /// carrying it as well would duplicate the event stream and cost a counter
64 /// threaded through the execution path.
65 pub workflows_considered: usize,
66}
67
68/// A message run finishing, whether it completed or stopped early.
69#[derive(Debug)]
70#[non_exhaustive]
71pub struct MessageFinished<'a> {
72 /// The message's id.
73 pub message_id: &'a str,
74 /// Wall-clock duration of the whole run, from the first workflow gate to
75 /// the last. Subtracting the workflow durations inside it gives the
76 /// engine's own between-workflow cost.
77 pub duration: Duration,
78 /// `message.errors().len()` at the end of the run.
79 pub errors: usize,
80 /// Whether the run stopped early with an `Err`.
81 pub stopped_early: bool,
82}
83
84/// A workflow beginning to run.
85///
86/// Fires only for a workflow that a rollout gate **and** its condition both
87/// admitted — a skipped workflow never starts, mirroring how a skipped task is
88/// not reported today.
89#[derive(Debug)]
90#[non_exhaustive]
91pub struct WorkflowStarted<'a> {
92 /// `Workflow::id`.
93 pub workflow_id: &'a str,
94}
95
96/// A workflow finishing.
97#[derive(Debug)]
98#[non_exhaustive]
99pub struct WorkflowFinished<'a> {
100 /// `Workflow::id`.
101 pub workflow_id: &'a str,
102 /// Wall-clock duration of the whole workflow, task bodies included.
103 ///
104 /// `duration - Σ task durations` for this workflow is the engine's own
105 /// overhead: condition evaluation, group gating, loop bookkeeping,
106 /// audit-trail writes and arena management. That figure was previously
107 /// only reachable as a whole-message residual.
108 pub duration: Duration,
109 /// Sweeps run. `1` for a workflow with no `loop`; a looping workflow
110 /// reports one event for the whole loop, with the count here — per-sweep
111 /// events would explode cardinality. Iterations only: a loop's `setup`
112 /// pass is not counted, so a loop over an empty array reports `0`.
113 pub sweeps: u32,
114 /// Whether the workflow ended by halting rather than running out of tasks.
115 pub halted: bool,
116}
117
118/// Receives one callback per dispatched task.
119///
120/// Object-safe by construction — no generic methods, no associated types — so
121/// this is unrelated to the [`crate::AsyncFunctionHandler`] /
122/// `DynAsyncFunctionHandler` split and needs no `Dyn` sibling.
123/// `Arc<dyn ExecutionObserver>` works directly.
124///
125/// # Contract
126///
127/// `task_finished` is called **synchronously**, on the executor's thread,
128/// immediately after the task body returns and *before* the audit trail is
129/// written. On the sync-built-in path it runs inside the arena scope while the
130/// `!Send` arena borrow is live.
131///
132/// So an implementation must not block, must not re-enter the engine, and must
133/// not panic — a panic unwinds through the arena scope and out of
134/// `process_message`. It cannot `await`, since the method is synchronous. Push to
135/// a channel or bump an atomic and return.
136///
137/// A task whose condition evaluated false is **not** reported: it was never
138/// dispatched, so there is nothing to time. Tasks that fail *are* reported, with
139/// `status: Some(500)` — the event is emitted before the error propagates,
140/// because those are the tasks a host most wants timed.
141///
142/// # Example
143///
144/// ```
145/// use dataflow_rs::{ExecutionObserver, TaskEvent};
146/// use std::sync::atomic::{AtomicU64, Ordering};
147///
148/// #[derive(Default)]
149/// struct TotalMicros(AtomicU64);
150///
151/// impl ExecutionObserver for TotalMicros {
152/// fn task_finished(&self, event: &TaskEvent<'_>) {
153/// // Cheap and non-blocking, as the contract requires.
154/// self.0
155/// .fetch_add(event.duration.as_micros() as u64, Ordering::Relaxed);
156/// }
157/// }
158/// ```
159pub trait ExecutionObserver: Send + Sync + 'static {
160 /// A message run is beginning. Defaulted to a no-op.
161 fn message_started(&self, _event: &MessageStarted<'_>) {}
162
163 /// A message run has finished. Defaulted to a no-op.
164 fn message_finished(&self, _event: &MessageFinished<'_>) {}
165
166 /// A workflow admitted by its gates is beginning. Defaulted to a no-op.
167 fn workflow_started(&self, _event: &WorkflowStarted<'_>) {}
168
169 /// A workflow has finished. Defaulted to a no-op.
170 fn workflow_finished(&self, _event: &WorkflowFinished<'_>) {}
171
172 /// Called once per dispatched task, immediately after its body returns.
173 fn task_finished(&self, event: &TaskEvent<'_>);
174}