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/// Receives one callback per dispatched task.
52///
53/// Object-safe by construction — no generic methods, no associated types — so
54/// this is unrelated to the [`crate::AsyncFunctionHandler`] /
55/// `DynAsyncFunctionHandler` split and needs no `Dyn` sibling.
56/// `Arc<dyn ExecutionObserver>` works directly.
57///
58/// # Contract
59///
60/// `task_finished` is called **synchronously**, on the executor's thread,
61/// immediately after the task body returns and *before* the audit trail is
62/// written. On the sync-built-in path it runs inside the arena scope while the
63/// `!Send` arena borrow is live.
64///
65/// So an implementation must not block, must not re-enter the engine, and must
66/// not panic — a panic unwinds through the arena scope and out of
67/// `process_message`. It cannot `await`, since the method is synchronous. Push to
68/// a channel or bump an atomic and return.
69///
70/// A task whose condition evaluated false is **not** reported: it was never
71/// dispatched, so there is nothing to time. Tasks that fail *are* reported, with
72/// `status: Some(500)` — the event is emitted before the error propagates,
73/// because those are the tasks a host most wants timed.
74///
75/// # Example
76///
77/// ```
78/// use dataflow_rs::{ExecutionObserver, TaskEvent};
79/// use std::sync::atomic::{AtomicU64, Ordering};
80///
81/// #[derive(Default)]
82/// struct TotalMicros(AtomicU64);
83///
84/// impl ExecutionObserver for TotalMicros {
85///     fn task_finished(&self, event: &TaskEvent<'_>) {
86///         // Cheap and non-blocking, as the contract requires.
87///         self.0
88///             .fetch_add(event.duration.as_micros() as u64, Ordering::Relaxed);
89///     }
90/// }
91/// ```
92pub trait ExecutionObserver: Send + Sync + 'static {
93    /// Called once per dispatched task, immediately after its body returns.
94    fn task_finished(&self, event: &TaskEvent<'_>);
95}
dataflow_rs/engine/observer.rs

dataflow_rs/engine/
observer.rs
