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.
112 pub sweeps: u32,
113 /// Whether the workflow ended by halting rather than running out of tasks.
114 pub halted: bool,
115}
116
117/// Receives one callback per dispatched task.
118///
119/// Object-safe by construction — no generic methods, no associated types — so
120/// this is unrelated to the [`crate::AsyncFunctionHandler`] /
121/// `DynAsyncFunctionHandler` split and needs no `Dyn` sibling.
122/// `Arc<dyn ExecutionObserver>` works directly.
123///
124/// # Contract
125///
126/// `task_finished` is called **synchronously**, on the executor's thread,
127/// immediately after the task body returns and *before* the audit trail is
128/// written. On the sync-built-in path it runs inside the arena scope while the
129/// `!Send` arena borrow is live.
130///
131/// So an implementation must not block, must not re-enter the engine, and must
132/// not panic — a panic unwinds through the arena scope and out of
133/// `process_message`. It cannot `await`, since the method is synchronous. Push to
134/// a channel or bump an atomic and return.
135///
136/// A task whose condition evaluated false is **not** reported: it was never
137/// dispatched, so there is nothing to time. Tasks that fail *are* reported, with
138/// `status: Some(500)` — the event is emitted before the error propagates,
139/// because those are the tasks a host most wants timed.
140///
141/// # Example
142///
143/// ```
144/// use dataflow_rs::{ExecutionObserver, TaskEvent};
145/// use std::sync::atomic::{AtomicU64, Ordering};
146///
147/// #[derive(Default)]
148/// struct TotalMicros(AtomicU64);
149///
150/// impl ExecutionObserver for TotalMicros {
151/// fn task_finished(&self, event: &TaskEvent<'_>) {
152/// // Cheap and non-blocking, as the contract requires.
153/// self.0
154/// .fetch_add(event.duration.as_micros() as u64, Ordering::Relaxed);
155/// }
156/// }
157/// ```
158pub trait ExecutionObserver: Send + Sync + 'static {
159 /// A message run is beginning. Defaulted to a no-op.
160 fn message_started(&self, _event: &MessageStarted<'_>) {}
161
162 /// A message run has finished. Defaulted to a no-op.
163 fn message_finished(&self, _event: &MessageFinished<'_>) {}
164
165 /// A workflow admitted by its gates is beginning. Defaulted to a no-op.
166 fn workflow_started(&self, _event: &WorkflowStarted<'_>) {}
167
168 /// A workflow has finished. Defaulted to a no-op.
169 fn workflow_finished(&self, _event: &WorkflowFinished<'_>) {}
170
171 /// Called once per dispatched task, immediately after its body returns.
172 fn task_finished(&self, event: &TaskEvent<'_>);
173}