dataflow_rs/run/
engine.rs

1//! The engine: a [`Program`] and its config, run over messages.
2
3use super::RunStatus;
4use super::budget::{ConditionErrorPolicy, TaskBudget};
5use super::builder::EngineBuilder;
6use super::cancel::{Limits, timers_available};
7use super::error_context::ErrorContext;
8use super::interp::{CaptureMode, Exit, Frame, RunEnv, Watch, run_workflow};
9use super::message::{Message, finish_audit, keeps_entries};
10use super::observers::Observers;
11use super::options::RunOptions;
12use super::record;
13use super::report::{Record, Run};
14use super::trace::Capture;
15use crate::def::Rollout;
16use crate::expr::Path;
17use crate::observe::{
18    Event, Observer, Selection, StepResult, TraceStep, WorkflowExit, WorkflowRecord,
19};
20use crate::program::{Compiler, Program};
21use crate::state::{RunContext, Tuning};
22use chrono::Utc;
23use datavalue::OwnedDataValue;
24use std::fmt;
25use std::sync::Arc;
26
27/// What an [`Engine`] is configured with beside its program. A reload,
28/// [`Engine::with_program`], keeps it.
29#[derive(Clone, Default)]
30#[non_exhaustive]
31pub struct EngineConfig {
32    /// The observer every run of the engine reports to, beside a run's own
33    /// ([`RunOptions::observer`](RunOptions::observer())). `None` by
34    /// default.
35    pub observer: Option<Arc<dyn Observer>>,
36    /// Where the records of each call that added any are mirrored in the
37    /// context, and how many are kept. `None` by default: nothing is
38    /// mirrored, and nothing is reset at run start.
39    pub error_context: Option<ErrorContext>,
40    /// Engine-wide default for how condition evaluation errors are handled
41    /// when [`RunOptions::condition_error_policy`] is `None`. Defaults to
42    /// [`ConditionErrorPolicy::Skip`].
43    pub condition_error_policy: ConditionErrorPolicy,
44}
45
46impl EngineConfig {
47    /// Reports every run of the engine to `o`.
48    #[must_use]
49    pub fn observer(mut self, o: Arc<dyn Observer>) -> Self {
50        self.observer = Some(o);
51        self
52    }
53
54    /// Mirrors every run's failure records by `ec`.
55    #[must_use]
56    pub fn error_context(mut self, ec: ErrorContext) -> Self {
57        self.error_context = Some(ec);
58        self
59    }
60
61    /// Sets the engine-wide default for how condition evaluation errors are
62    /// handled.
63    #[must_use]
64    pub fn condition_error_policy(mut self, policy: ConditionErrorPolicy) -> Self {
65        self.condition_error_policy = policy;
66        self
67    }
68
69    /// Alias for [`Self::condition_error_policy`].
70    #[must_use]
71    pub fn with_condition_error_policy(self, policy: ConditionErrorPolicy) -> Self {
72        self.condition_error_policy(policy)
73    }
74}
75
76impl fmt::Debug for EngineConfig {
77    /// The observer as a presence, and the error context:
78    /// `EngineConfig { observer: present, error_context: None, .. }`.
79    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
80        let observer = if self.observer.is_some() {
81            "present"
82        } else {
83            "absent"
84        };
85        f.debug_struct("EngineConfig")
86            .field("observer", &format_args!("{observer}"))
87            .field("error_context", &self.error_context)
88            .finish_non_exhaustive()
89    }
90}
91
92/// A program and its config: what runs messages.
93///
94/// An `Arc` inside, so cloning is cheap and a clone runs the same program.
95/// [`with_program`](Self::with_program) is a reload: the engine it returns
96/// runs the new program, and this one, with any run in flight, keeps the old.
97///
98/// ```
99/// # tokio::runtime::Builder::new_current_thread().build().unwrap().block_on(async {
100/// use dataflow_rs::datavalue::OwnedDataValue;
101/// use dataflow_rs::{Engine, Message};
102/// use serde_json::json;
103///
104/// let engine = Engine::builder()
105///     .workflow_json(&json!({"id": "w", "tasks": [
106///         {"id": "t", "function": {"name": "map", "input": {
107///             "mappings": [{"path": "data.greeting", "logic": "hello"}]}}}
108///     ]}))
109///     .build()
110///     .unwrap();
111/// let mut message = Message::from_json(&json!({}));
112/// let run = engine.run(&mut message).await;
113/// assert_eq!(run.matched(), 1);
114/// let greeting = OwnedDataValue::from(&json!({"greeting": "hello"}));
115/// assert_eq!(message.context().data(), &greeting);
116/// # });
117/// ```
118#[derive(Clone)]
119pub struct Engine {
120    inner: Arc<Inner>,
121}
122
123struct Inner {
124    program: Program,
125    config: EngineConfig,
126}
127
128/// The `metadata` keys the engine writes on every run: the progress after
129/// every call, which cross-workflow chaining reads, and the run's stamps,
130/// written before any op.
131pub(crate) mod slots {
132    pub(crate) const PROGRESS: &str = "progress";
133    pub(crate) const PROCESSED_AT: &str = "processed_at";
134    pub(crate) const ENGINE_VERSION: &str = "engine_version";
135    pub(crate) const CHANNEL: &str = "channel";
136    /// Every one of them.
137    pub(crate) const ALL: [&str; 4] = [PROGRESS, PROCESSED_AT, ENGINE_VERSION, CHANNEL];
138}
139
140/// `metadata.<key>`.
141pub(crate) fn metadata(key: &str) -> Path {
142    format!("metadata.{key}")
143        .parse()
144        .expect("a metadata key is a path")
145}
146
147impl Engine {
148    /// An engine running `program` with `cfg`.
149    pub fn new(program: Program, cfg: EngineConfig) -> Self {
150        Self {
151            inner: Arc::new(Inner {
152                program,
153                config: cfg,
154            }),
155        }
156    }
157
158    /// A builder starting from the default registry, for a host that does
159    /// not quarantine: its `build` fails if any workflow is rejected.
160    pub fn builder() -> EngineBuilder {
161        EngineBuilder::new()
162    }
163
164    /// Runs `msg` with [`RunOptions::default`].
165    pub async fn run(&self, msg: &mut Message) -> Run {
166        // Borrowed for the whole run: a temporary would make the run's
167        // future carry the options.
168        static DEFAULT: RunOptions<'static> = RunOptions::DEFAULT;
169        self.run_with(msg, &DEFAULT).await
170    }
171
172    /// Runs `msg` through the program's Active workflows, those of
173    /// `opts.channel` when it names one, in priority order.
174    ///
175    /// The run reads the clock once: `metadata.processed_at`, every audit
176    /// entry and every error record carry that reading. Only an observer
177    /// asking for [`TIMINGS`](crate::observe::Interest::TIMINGS), or a trace
178    /// asking for [`timings`](crate::observe::TraceOptions::timings()), has the
179    /// clock read again, for the elapsed times it is given. It writes
180    /// `metadata.engine_version`, and `metadata.channel` under a channel,
181    /// before any workflow runs, and then, with an [`ErrorContext`]
182    /// configured, removes its path's key: the records there are the run's
183    /// own. A workflow whose rollout does not admit `opts.routing_bucket` is
184    /// skipped; no bucket is admitted by every rollout. A workflow that stops
185    /// the run ends it, and the [`Run`] says how:
186    /// [`TimedOut`](RunStatus::TimedOut) once `opts.deadline` passed,
187    /// [`Cancelled`](RunStatus::Cancelled) once `opts.cancel` was cancelled,
188    /// and otherwise [`Stopped`](RunStatus::Stopped) with the error that
189    /// stopped it, whatever its kind: a child run's timeout that a function
190    /// returns before this run's own deadline passed is its task's failure,
191    /// not this run's. An `Err` a function returns once `opts.deadline` has
192    /// passed, or `opts.cancel` was cancelled, is the run's limit's.
193    ///
194    /// With [`opts.trace`](RunOptions::trace()) set, the [`Run`] carries the
195    /// run's [`Trace`](crate::observe::Trace): the context after the stamps
196    /// and the reset, then a step for every recorded call, skipped call or
197    /// workflow and sweep, the step that stopped the run included: the call
198    /// a limit cut short, or, for a stop as a sweep starts, the engine's
199    /// step for the counter it wrote. Its replay is the message's context on
200    /// every status.
201    pub async fn run_with(&self, msg: &mut Message, opts: &RunOptions<'_>) -> Run {
202        let program = &self.inner.program;
203        // Built in a block: a local the run borrowed would stay in its
204        // future beside the `Watch` it moved into.
205        let mut watch = {
206            let observers = Observers::attach(self.inner.config.observer.as_deref(), opts.observer);
207            Watch {
208                timed: observers.timed_tasks(),
209                observers,
210                trace: None,
211                error_context: self.inner.config.error_context.as_ref(),
212            }
213        };
214        let started = watch.observers.timed().then(record::now);
215        let now = Utc::now();
216        let now_str = now.to_rfc3339();
217        let mut now_text: Option<Arc<str>> = None;
218        let parts = msg.parts();
219        stamp(parts.context, now_str, opts.channel);
220        // A message brings no records in at the error context's path.
221        if let Some(ec) = &self.inner.config.error_context {
222            ec.reset(parts.context);
223        }
224        // `initial` is the context as the stamps and the reset left it.
225        if let Some(options) = &opts.trace {
226            let trace = Capture::new(options.clone(), parts.context);
227            watch.timed |= trace.timings();
228            watch.trace = Some(Box::new(trace));
229        }
230
231        let mut rc = RunContext::new(parts.context, Tuning::default());
232        let active = program.active(opts.channel);
233        watch.observers.emit(Event::RunStarted {
234            program: program.id(),
235            message_id: parts.id,
236            channel: opts.channel,
237            considered: active.len(),
238        });
239        // A call's changes are worth recording only in a mode that keeps
240        // an entry for them.
241        let capture_audit = opts.capture_changes && keeps_entries(opts.audit);
242        let mut records = Vec::with_capacity(active.len());
243        // Every 64th op of the run checks the limits, across workflows.
244        let mut ops = 0;
245        // Whether this run's runtime can race a deadline with a timer: the
246        // runtime context is the run's, so the answer is the run's.
247        let timers = opts.deadline.is_some() && timers_available();
248        // A bucket above 99 is served as 99, as in 3.x: the bucket space is
249        // `0..100`, and `[x, 100)` is the range that serves the rest.
250        let bucket = opts.routing_bucket.map(|b| b.min(Rollout::MAX_BUCKET));
251        let condition_error_policy = opts
252            .condition_error_policy
253            .unwrap_or(self.inner.config.condition_error_policy);
254        let budget = TaskBudget::new(opts.task_budget, opts.expr_ops_budget, opts.accountant);
255        // How the run ended, once a workflow stopped it: the loop's value,
256        // so the run's future holds no status across its workflows.
257        let stopped = 'run: {
258            for &ix in active {
259                let wf = program.compiled(ix);
260                if let (Some(rollout), Some(bucket)) = (wf.rollout, bucket)
261                    && !rollout.accepts(bucket)
262                {
263                    if let Some(trace) = watch.trace.as_deref_mut() {
264                        let (workflow_id, _) = program.ids(ix);
265                        trace.push(TraceStep::new(ix, workflow_id, StepResult::Skipped));
266                    }
267                    records.push(Record {
268                        ix,
269                        selection: Selection::SkippedByRollout,
270                        sweeps: 0,
271                        exit: None,
272                    });
273                    continue;
274                }
275                let (workflow_id, task_ids) = program.ids(ix);
276                let mut env = RunEnv {
277                    program: program.id(),
278                    workflow_ix: ix,
279                    task_base: program.task_base(ix),
280                    channel: opts.channel,
281                    routing_bucket: bucket,
282                    layers: opts.layers,
283                    limits: Limits {
284                        deadline: opts.deadline,
285                        cancel: opts.cancel,
286                        timers,
287                    },
288                    condition_error_policy,
289                    budget: &budget,
290                    ops,
291                    exprs: program.registry().exprs(),
292                    payload: parts.payload,
293                    errors: &mut *parts.errors,
294                    audit: &mut *parts.audit,
295                    mode: opts.audit,
296                    capture: CaptureMode::new(capture_audit, &watch),
297                    watch: &mut watch,
298                    now,
299                    now_text: &mut now_text,
300                    changes: Vec::new(),
301                    workflow_id,
302                    task_ids,
303                    failed: false,
304                    frame: Frame::default(),
305                };
306                let (exit, selection) = run_workflow(&mut rc, wf, &mut env).await;
307                ops = env.ops;
308                let ran = matches!(selection, Selection::Ran);
309                let (exit, stop) = match exit {
310                    Exit::Completed => (WorkflowExit::Completed, None),
311                    Exit::Halted => (WorkflowExit::Halted, None),
312                    Exit::Failed => (WorkflowExit::Failed, None),
313                    Exit::Stopped(e) => (WorkflowExit::Stopped, Some(RunStatus::Stopped(e))),
314                    // Only the run's own limits make it `TimedOut` or
315                    // `Cancelled`: a task's `Err` of either kind, a child run's
316                    // propagated, stops it as any failure does, unless one of
317                    // them had passed when the call returned it.
318                    Exit::Limited(stop) => (WorkflowExit::Stopped, Some(stop.status())),
319                };
320                let considered = Record {
321                    ix,
322                    selection,
323                    sweeps: env.frame.sweeps,
324                    exit: ran.then_some(exit),
325                };
326                let workflow_started = env.frame.started;
327                if ran && watch.observers.workflows() {
328                    let view = WorkflowRecord {
329                        id: workflow_id,
330                        index: ix,
331                        selection: &considered.selection,
332                        sweeps: considered.sweeps,
333                        exit: considered.exit,
334                    };
335                    watch.observers.emit(Event::WorkflowFinished {
336                        record: &view,
337                        elapsed: workflow_started.map(record::since),
338                    });
339                }
340                records.push(considered);
341                if let Some(status) = stop {
342                    break 'run Some(status);
343                }
344            }
345            None
346        };
347        let stats = rc.finish();
348        finish_audit(parts.audit, opts.audit);
349        let status = match stopped {
350            Some(status) => status,
351            None if !parts.errors.is_empty() => RunStatus::CompletedWithErrors,
352            None => RunStatus::Completed,
353        };
354        watch.observers.emit(Event::RunFinished {
355            message_id: parts.id,
356            status: &status,
357            errors: parts.errors.len(),
358            elapsed: started.map(record::since),
359        });
360        Run::new(
361            program.clone(),
362            status,
363            records,
364            stats,
365            watch.trace.map(|t| Box::new(t.finish())),
366        )
367    }
368
369    /// A compiler with the program's registry and options, whose workflows
370    /// link into a program this engine can run.
371    pub fn compiler(&self) -> Compiler {
372        let program = &self.inner.program;
373        program.registry().compiler(program.options().clone())
374    }
375
376    /// An engine running `program` with this engine's config: a reload. This
377    /// engine is unchanged.
378    #[must_use]
379    pub fn with_program(&self, program: Program) -> Self {
380        Self::new(program, self.inner.config.clone())
381    }
382
383    /// The program this engine runs.
384    pub fn program(&self) -> &Program {
385        &self.inner.program
386    }
387
388    /// The config this engine runs with.
389    #[doc(hidden)]
390    pub fn config(&self) -> &EngineConfig {
391        &self.inner.config
392    }
393}
394
395/// Writes the run's stamps, owned: no view exists yet. Each replaces its key
396/// in place or is appended, as v3's `set_processing_metadata`.
397fn stamp(context: &mut OwnedDataValue, now: String, channel: Option<&str>) {
398    let OwnedDataValue::Object(roots) = context else {
399        return;
400    };
401    let Some((_, OwnedDataValue::Object(meta))) = roots.iter_mut().find(|(k, _)| k == "metadata")
402    else {
403        return;
404    };
405    let need = 2 + usize::from(channel.is_some());
406    if meta.capacity() < meta.len() + need {
407        meta.reserve(need);
408    }
409    match meta.iter_mut().find(|(k, _)| k == slots::PROCESSED_AT) {
410        Some((_, slot)) => *slot = OwnedDataValue::String(now),
411        None => meta.push((slots::PROCESSED_AT.to_owned(), OwnedDataValue::String(now))),
412    }
413    put_meta_str(meta, slots::ENGINE_VERSION, crate::VERSION);
414    if let Some(channel) = channel {
415        put_meta_str(meta, slots::CHANNEL, channel);
416    }
417}
418
419fn put_meta_str(meta: &mut Vec<(String, OwnedDataValue)>, key: &str, text: &str) {
420    match meta.iter_mut().find(|(k, _)| k == key) {
421        Some((_, OwnedDataValue::String(s))) => {
422            if s != text {
423                s.clear();
424                s.push_str(text);
425            }
426        }
427        Some((_, slot)) => *slot = OwnedDataValue::String(text.to_owned()),
428        None => meta.push((key.to_owned(), OwnedDataValue::String(text.to_owned()))),
429    }
430}
431
432impl fmt::Debug for Engine {
433    /// The program and the config.
434    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
435        f.debug_struct("Engine")
436            .field("program", &self.inner.program)
437            .field("config", &self.inner.config)
438            .finish()
439    }
440}
dataflow_rs/run/engine.rs

dataflow_rs/run/
engine.rs
