dataflow_rs/engine/task_context.rs
1//! # Task context
2//!
3//! Wraps the per-call state passed to every `AsyncFunctionHandler::execute`
4//! call: the message under processing, a handle to the shared datalogic
5//! engine, and an audit-trail accumulator. Exposes typed helpers so handlers
6//! don't have to reach into `crate::engine::utils::{get,set}_nested_value`
7//! or hand-build `Change` entries.
8//!
9//! Custom handlers should treat `TaskContext` as their *only* mutation surface
10//! for `message.context`: the `set` family records a `Change` automatically
11//! when `message.capture_changes` is true, keeping the audit trail in sync
12//! with the data without per-handler boilerplate.
13
14use crate::engine::error::{ErrorInfo, Result};
15use crate::engine::message::{Change, Message};
16use crate::engine::secrets::{self, Secrets};
17use crate::engine::utils::{get_nested_value, set_nested_value};
18use bumpalo::Bump;
19use datalogic_rs::{Engine as DatalogicEngine, Logic};
20use datavalue::{DataValue, OwnedDataValue};
21use serde_json::Value as JsonValue;
22use std::sync::Arc;
23
24/// Per-call execution context handed to `AsyncFunctionHandler::execute`.
25///
26/// Borrows the message and datalogic engine for the duration of the handler
27/// call; collects `Change` entries that the workflow executor folds into the
28/// audit trail when the handler returns. Drop semantics are trivial — there
29/// is nothing to flush; the executor extracts the buffered changes via
30/// `into_changes()`.
31pub struct TaskContext<'a> {
32 message: &'a mut Message,
33 datalogic: &'a Arc<DatalogicEngine>,
34 /// Changes accumulated through the `set*` family. Only populated when
35 /// `message.capture_changes` is true; otherwise pushes are no-ops to
36 /// keep the bulk-pipeline fast path allocation-free.
37 changes: Vec<Change>,
38 /// Who is executing, when the engine built this context.
39 ///
40 /// Borrowed rather than `Arc`-cloned: the ids live on the `Workflow` and
41 /// `Task`, both of which outlive the dispatch call, and the accessors hand
42 /// back `&str` either way — so a refcount bump per task would buy nothing.
43 identity: Option<TaskIdentity<'a>>,
44 /// Sweep index of the enclosing looping workflow, if any.
45 loop_counter: Option<i64>,
46 /// Fan-out element this call runs for, if the task carries `for_each`.
47 element_index: Option<usize>,
48 /// The engine's secret store — empty for a context built with
49 /// [`Self::new`], for the same reason `identity` is `None` there.
50 secrets: &'a Secrets,
51}
52
53/// Where a handler call sits in a run: the loop sweep and the fan-out
54/// element, either of which may be absent.
55///
56/// One parameter rather than two, so the dispatch path stays under clippy's
57/// argument-count limit.
58#[derive(Debug, Clone, Copy, Default)]
59pub(crate) struct CallStamp {
60 pub loop_counter: Option<i64>,
61 pub element_index: Option<usize>,
62}
63
64/// Which task, in which workflow, the engine is currently running.
65///
66/// All-or-nothing by construction: the engine executing a task inside a
67/// workflow knows both ids, and every other path knows neither. Two separate
68/// `Option<&str>` fields would additionally allow "workflow known, task
69/// unknown" — a state that never occurs — and would be silently swappable at
70/// the call site, being adjacent and identically typed.
71#[derive(Debug, Clone, Copy)]
72pub(crate) struct TaskIdentity<'a> {
73 pub workflow_id: &'a str,
74 pub task_id: &'a str,
75}
76
77impl<'a> TaskContext<'a> {
78 /// Construct a new context. Mostly engine-internal — handlers receive a
79 /// pre-built `&mut TaskContext` from the executor — but exposed `pub` so
80 /// tests and benchmarks can drive `AsyncFunctionHandler::execute`
81 /// directly without going through `Engine::process_message`.
82 /// A context built this way reports `None` from [`Self::workflow_id`],
83 /// [`Self::task_id`] and [`Self::loop_counter`] — there is no workflow run
84 /// to describe, and inventing ids would be worse than admitting their
85 /// absence.
86 pub fn new(message: &'a mut Message, datalogic: &'a Arc<DatalogicEngine>) -> Self {
87 Self {
88 message,
89 datalogic,
90 changes: Vec::new(),
91 identity: None,
92 loop_counter: None,
93 element_index: None,
94 secrets: &secrets::EMPTY,
95 }
96 }
97
98 /// As [`Self::new`], with the identity of the executing task.
99 ///
100 /// Used by the task executor on the dispatch path. A separate constructor
101 /// rather than setters, so there is no window in which a context exists
102 /// with half its identity filled in.
103 pub(crate) fn with_identity(
104 message: &'a mut Message,
105 datalogic: &'a Arc<DatalogicEngine>,
106 identity: Option<TaskIdentity<'a>>,
107 stamp: CallStamp,
108 secrets: &'a Secrets,
109 ) -> Self {
110 Self {
111 message,
112 datalogic,
113 changes: Vec::new(),
114 identity,
115 loop_counter: stamp.loop_counter,
116 element_index: stamp.element_index,
117 secrets,
118 }
119 }
120
121 /// Id of the workflow being executed, when the engine built this context.
122 ///
123 /// `None` for a context built with [`Self::new`] — a test or benchmark
124 /// driving a handler directly is not inside a workflow run.
125 ///
126 /// ```
127 /// # use dataflow_rs::{TaskContext, engine::message::Message};
128 /// # use serde_json::json;
129 /// # let datalogic = std::sync::Arc::new(datalogic_rs::Engine::new());
130 /// # let mut message = Message::from_value(&json!({}));
131 /// let ctx = TaskContext::new(&mut message, &datalogic);
132 /// assert_eq!(ctx.workflow_id(), None);
133 /// ```
134 #[inline]
135 pub fn workflow_id(&self) -> Option<&str> {
136 self.identity.map(|i| i.workflow_id)
137 }
138
139 /// Id of the task being executed, when the engine built this context.
140 ///
141 /// Always a **leaf** task's id. Handlers run only on leaf tasks — a task
142 /// group is span bookkeeping recorded on the task that opens it, never a
143 /// dispatch target — so a group id can never appear here.
144 ///
145 /// `None` for a context built with [`Self::new`].
146 #[inline]
147 pub fn task_id(&self) -> Option<&str> {
148 self.identity.map(|i| i.task_id)
149 }
150
151 /// Sweep index of the enclosing looping workflow, or `None` when the
152 /// workflow does not carry a `loop`.
153 ///
154 /// This is a different fact from identity being unknown: a handler in a
155 /// non-looping workflow has both ids and no counter.
156 ///
157 /// Worth preferring over reading the counter out of `temp_data`, which
158 /// only works when the host gave `LoopConfig` a `counter` name and the
159 /// handler hardcodes that path. A loop with no named counter writes to no
160 /// path at all, and its sweep index is reachable no other way.
161 #[inline]
162 pub fn loop_counter(&self) -> Option<i64> {
163 self.loop_counter
164 }
165
166 /// Index of the element this call runs for, when the task carries a
167 /// [`for_each`](crate::engine::for_each::ForEach); `None` otherwise.
168 ///
169 /// The same value the engine binds at `temp_data.<as>_index` in this
170 /// call's view of the message, without the handler having to know the
171 /// `as` name.
172 ///
173 /// ```
174 /// # use dataflow_rs::{TaskContext, engine::message::Message};
175 /// # use serde_json::json;
176 /// # let datalogic = std::sync::Arc::new(datalogic_rs::Engine::new());
177 /// # let mut message = Message::from_value(&json!({}));
178 /// let ctx = TaskContext::new(&mut message, &datalogic);
179 /// assert_eq!(ctx.element_index(), None);
180 /// ```
181 #[inline]
182 pub fn element_index(&self) -> Option<usize> {
183 self.element_index
184 }
185
186 /// A secret by dotted name, from the store the host configured through
187 /// [`crate::EngineBuilder::with_secrets`].
188 ///
189 /// For handlers whose config names a key (`"key_name": "partner_hmac"`)
190 /// rather than embedding a [`crate::Template`] — a `Template` field can
191 /// simply read `{"secret": "partner_hmac"}` and needs nothing here.
192 ///
193 /// `None` when the key is not declared, and always `None` for a context
194 /// built with [`Self::new`]. The contract for what you do with the value is
195 /// one line: **a handler must not write a secret-derived value into the
196 /// message.** Nothing in the engine records what this returns; whether it
197 /// stays unrecorded is the handler's business from here on.
198 #[inline]
199 pub fn secret(&self, name: &str) -> Option<&OwnedDataValue> {
200 self.secrets.get(name)
201 }
202
203 /// Borrow the message under processing. Use this when you need to inspect
204 /// the message id, payload, or audit trail; for reading and mutating the
205 /// `data` / `metadata` / `temp_data` context, prefer the typed helpers on
206 /// `TaskContext` itself.
207 ///
208 /// Inside a [`for_each`](crate::ForEach) call this is the call's own copy,
209 /// whose audit trail and errors start empty: it carries the context, id
210 /// and payload, not the message's history.
211 #[inline]
212 pub fn message(&self) -> &Message {
213 self.message
214 }
215
216 /// Mutable access to the message. Prefer the typed helpers (`set`,
217 /// `add_error`) over poking at `message.context` directly — direct
218 /// mutations bypass the audit trail.
219 #[inline]
220 pub fn message_mut(&mut self) -> &mut Message {
221 self.message
222 }
223
224 /// Shared datalogic engine, in case the handler needs to evaluate ad-hoc
225 /// JSONLogic. Most handlers can ignore this argument.
226 #[inline]
227 pub fn datalogic(&self) -> &Arc<DatalogicEngine> {
228 self.datalogic
229 }
230
231 /// Read-only view of `data`. Returns `&OwnedDataValue::Null` if missing
232 /// (mirrors the `Index` fallback semantics of `serde_json::Value`).
233 #[inline]
234 pub fn data(&self) -> &OwnedDataValue {
235 self.message.data()
236 }
237
238 /// Read-only view of `metadata`.
239 #[inline]
240 pub fn metadata(&self) -> &OwnedDataValue {
241 self.message.metadata()
242 }
243
244 /// Read-only view of `temp_data`.
245 #[inline]
246 pub fn temp_data(&self) -> &OwnedDataValue {
247 self.message.temp_data()
248 }
249
250 /// The full `{data, metadata, temp_data}` tree — the root every workflow
251 /// JSONLogic expression is written against.
252 ///
253 /// [`Self::data`] / [`Self::metadata`] / [`Self::temp_data`] expose the three
254 /// slots individually; this is the whole-context accessor, so handlers do not
255 /// have to reach through `ctx.message().context`.
256 ///
257 /// Note `payload` is **not** part of this tree, and therefore not part of the
258 /// JSONLogic evaluation context — `{"var": "payload.foo"}` resolves to
259 /// nothing. Parse the payload into `data` first.
260 #[inline]
261 pub fn context(&self) -> &OwnedDataValue {
262 &self.message.context
263 }
264
265 /// Evaluate a pre-compiled expression against the message context, on the
266 /// worker thread's pooled arena.
267 ///
268 /// The same path [`crate::engine::executor::evaluate_condition`] takes, but
269 /// returning the value instead of collapsing it to a bool, and surfacing
270 /// evaluation failures as `Err` instead of `false`. That difference is
271 /// deliberate: a condition that fails to evaluate should not run its task,
272 /// whereas a handler reading a config value needs to know the read failed.
273 ///
274 /// # Errors
275 ///
276 /// [`crate::DataflowError::LogicEvaluation`] if the expression fails to
277 /// evaluate, or [`crate::DataflowError::BudgetExceeded`] if it was aborted
278 /// for crossing the ceiling set by
279 /// [`crate::EngineBuilder::with_ops_budget`].
280 pub fn eval(&self, logic: &Logic) -> Result<OwnedDataValue> {
281 crate::engine::executor::eval_to_owned(self.datalogic, logic, &self.message.context)
282 .map_err(|e| crate::engine::error::from_datalogic_eval(&e))
283 }
284
285 /// As [`Self::eval`], projected straight from the arena to
286 /// `serde_json::Value` in one walk — no `OwnedDataValue` intermediate and no
287 /// `serde_json::from_value` rebuild.
288 pub fn eval_json(&self, logic: &Logic) -> Result<JsonValue> {
289 crate::engine::executor::eval_to_json(self.datalogic, logic, &self.message.context)
290 .map_err(|e| crate::engine::error::from_datalogic_eval(&e))
291 }
292
293 /// As [`Self::eval`], coerced to a *plain* string: a JSON string result
294 /// yields its contents, anything else its compact JSON form.
295 ///
296 /// # This disagrees with datalogic-rs on purpose
297 ///
298 /// datalogic-rs's `String: FromDataValue` — and therefore
299 /// `Session::eval_str` — keeps the JSON quoting, so a string result comes
300 /// back from it as `"\"abc\""`. This method returns `abc`.
301 ///
302 /// The name says `plain_string` rather than `to_string` precisely so the
303 /// difference is visible at the call site: two string semantics in one
304 /// ecosystem is a footgun, and these values end up in URL paths and message
305 /// keys. A test pins both sides, so it fails if either changes.
306 pub fn eval_to_plain_string(&self, logic: &Logic) -> Result<String> {
307 crate::engine::executor::eval_to_plain_string(self.datalogic, logic, &self.message.context)
308 .map_err(|e| crate::engine::error::from_datalogic_eval(&e))
309 }
310
311 /// Run `f` against one arena view of the message context, so several
312 /// expressions cost one view of the context instead of one each.
313 ///
314 /// Every [`Self::eval`] (and every [`Template::resolve`] on a field that
315 /// reads the message) builds an arena view of the *whole* context before it
316 /// evaluates, so a step costs time in proportion to the number of arrays
317 /// and objects in the message, not to what it reads. A handler resolving k
318 /// templated fields pays for k views. Inside `with_view` it pays for one:
319 /// resolve the fields with [`Template::resolve_in`] and friends, or
320 /// evaluate directly with [`ContextView::eval`].
321 ///
322 /// Strings, keys and tensor buffers are borrowed rather than copied
323 /// ([`datavalue::OwnedDataValue::view_in`]), on this path and on the
324 /// per-call one, so what a view saves is the spine walk, not the bytes.
325 ///
326 /// The view borrows `self`, so nothing can write to the message while it
327 /// is alive, and the closure is synchronous, so the view cannot be held
328 /// across an `.await`. Resolve what the call needs up front, then do the
329 /// I/O:
330 ///
331 /// ```rust
332 /// # use dataflow_rs::{Result, TaskContext, Template};
333 /// # fn run(ctx: &mut TaskContext<'_>, url: &Template, timeout: &Template) -> Result<()> {
334 /// let (url, timeout_ms) = ctx.with_view(|view| {
335 /// Ok::<_, dataflow_rs::DataflowError>((
336 /// url.resolve_string_in(view)?,
337 /// timeout.resolve_u64_in(view, "timeout_ms")?,
338 /// ))
339 /// })?;
340 /// // ... await the call with `url` and `timeout_ms` ...
341 /// # let _ = (url, timeout_ms);
342 /// # Ok(())
343 /// # }
344 /// ```
345 ///
346 /// [`Template::resolve`]: crate::Template::resolve
347 /// [`Template::resolve_in`]: crate::Template::resolve_in
348 pub fn with_view<R>(&self, f: impl FnOnce(&ContextView<'_>) -> R) -> R {
349 crate::engine::executor::with_eval_arena(|arena| {
350 let data = arena.alloc(self.message.context.view_in(arena));
351 f(&ContextView {
352 datalogic: self.datalogic,
353 arena,
354 data,
355 })
356 })
357 }
358
359 /// Look up a value by dot-path against the full context tree (rooted at
360 /// the unified `{data, metadata, temp_data}` object). Returns `None` if
361 /// the path doesn't resolve.
362 ///
363 /// Use the same path syntax as JSONLogic: `"data.user.name"`,
364 /// `"temp_data.items.0"`, `"metadata.progress.status_code"`.
365 #[inline]
366 pub fn get(&self, path: &str) -> Option<&OwnedDataValue> {
367 get_nested_value(&self.message.context, path)
368 }
369
370 /// Set a value at a dot-path on the context. Records a `Change` on the
371 /// audit trail when `message.capture_changes` is true; otherwise the
372 /// write happens but no audit entry is buffered.
373 ///
374 /// Intermediate objects/arrays are created on demand; see
375 /// [`crate::engine::utils::set_nested_value`] for the exact semantics
376 /// (numeric segments → arrays, `#` prefix → escaped object key, etc.).
377 pub fn set(&mut self, path: &str, value: OwnedDataValue) {
378 if self.message.capture_changes {
379 let old_value = get_nested_value(&self.message.context, path)
380 .cloned()
381 .unwrap_or(OwnedDataValue::Null);
382 let new_value = value.clone();
383 self.changes.push(Change {
384 path: Arc::from(path),
385 old_value,
386 new_value,
387 removed: false,
388 });
389 }
390 set_nested_value(&mut self.message.context, path, value);
391 }
392
393 /// Same as [`Self::set`] but accepts a `serde_json::Value` (bridges
394 /// through `OwnedDataValue::from`). Convenience for handlers that
395 /// already speak `serde_json::Value`.
396 #[inline]
397 pub fn set_json(&mut self, path: &str, value: &JsonValue) {
398 self.set(path, OwnedDataValue::from(value));
399 }
400
401 /// Append an error to `message.errors`. Convenience for
402 /// `ctx.message_mut().add_error(...)`.
403 #[inline]
404 pub fn add_error(&mut self, error: ErrorInfo) {
405 self.message.add_error(error);
406 }
407
408 /// Drain the accumulated changes. The workflow executor calls this after
409 /// the handler returns to fold them into the audit trail; tests and
410 /// benchmarks driving the trait directly can use it to inspect what the
411 /// handler buffered.
412 #[inline]
413 pub fn into_changes(self) -> Vec<Change> {
414 self.changes
415 }
416}
417
418/// The message context converted into the arena once, for evaluating several
419/// expressions against it. Built by [`TaskContext::with_view`].
420///
421/// Each method evaluates exactly as its [`TaskContext`] namesake does, with
422/// the same errors, but against this one conversion instead of a fresh one.
423pub struct ContextView<'v> {
424 datalogic: &'v DatalogicEngine,
425 arena: &'v Bump,
426 data: &'v DataValue<'v>,
427}
428
429impl ContextView<'_> {
430 /// As [`TaskContext::eval`], against this view.
431 ///
432 /// # Errors
433 ///
434 /// As [`TaskContext::eval`].
435 pub fn eval(&self, logic: &Logic) -> Result<OwnedDataValue> {
436 self.evaluate(logic).map(|v| v.to_owned())
437 }
438
439 /// As [`TaskContext::eval_json`], against this view.
440 ///
441 /// # Errors
442 ///
443 /// As [`TaskContext::eval`].
444 pub fn eval_json(&self, logic: &Logic) -> Result<JsonValue> {
445 self.evaluate(logic).map(|v| v.to_serde_value())
446 }
447
448 /// As [`TaskContext::eval_to_plain_string`], against this view: a string
449 /// result yields its contents, anything else its compact JSON form.
450 ///
451 /// # Errors
452 ///
453 /// As [`TaskContext::eval`].
454 pub fn eval_to_plain_string(&self, logic: &Logic) -> Result<String> {
455 self.evaluate(logic).map(|v| match v {
456 DataValue::String(s) => s.to_string(),
457 other => other.to_string(),
458 })
459 }
460
461 fn evaluate<'r>(&'r self, logic: &'r Logic) -> Result<&'r DataValue<'r>> {
462 self.datalogic
463 .evaluate(logic, self.data, self.arena)
464 .map_err(|e| crate::engine::error::from_datalogic_eval(&e))
465 }
466}
467
468#[cfg(test)]
469mod tests {
470 use super::*;
471 use crate::engine::error::DataflowError;
472 use crate::engine::executor::with_arena;
473 use crate::engine::utils::set_nested_value;
474 use serde_json::json;
475
476 fn dv(v: serde_json::Value) -> OwnedDataValue {
477 OwnedDataValue::from(&v)
478 }
479
480 fn engine() -> Arc<DatalogicEngine> {
481 Arc::new(crate::engine::compiler::datalogic_engine_builder().build())
482 }
483
484 /// A message with one key in each of the three context slots.
485 fn populated() -> Message {
486 let mut m = Message::from_value(&json!({"payload_key": "payload_value"}));
487 set_nested_value(&mut m.context, "data.x", dv(json!("dx")));
488 set_nested_value(&mut m.context, "metadata.x", dv(json!("mx")));
489 set_nested_value(&mut m.context, "temp_data.x", dv(json!("tx")));
490 m
491 }
492
493 #[test]
494 fn context_matches_the_three_slot_accessors() {
495 let mut m = populated();
496 let dl = engine();
497 let ctx = TaskContext::new(&mut m, &dl);
498
499 let whole = ctx.context();
500 assert_eq!(&whole["data"], ctx.data());
501 assert_eq!(&whole["metadata"], ctx.metadata());
502 assert_eq!(&whole["temp_data"], ctx.temp_data());
503 assert_eq!(whole, &ctx.message().context);
504 }
505
506 #[test]
507 fn eval_roots_at_the_unified_context_not_data_alone() {
508 let mut m = populated();
509 let dl = engine();
510 let ctx = TaskContext::new(&mut m, &dl);
511
512 for (path, expected) in [
513 ("data.x", "dx"),
514 ("metadata.x", "mx"),
515 ("temp_data.x", "tx"),
516 ] {
517 let logic = dl.compile_arc(&json!({"var": path})).unwrap();
518 assert_eq!(ctx.eval(&logic).unwrap(), dv(json!(expected)));
519 assert_eq!(ctx.eval_json(&logic).unwrap(), json!(expected));
520 assert_eq!(ctx.eval_to_plain_string(&logic).unwrap(), expected);
521 }
522 }
523
524 #[test]
525 fn payload_is_not_in_the_eval_context() {
526 // Stays true through the new surface: `payload` is a separate field on
527 // Message and never part of the JSONLogic root.
528 let mut m = populated();
529 let dl = engine();
530 let ctx = TaskContext::new(&mut m, &dl);
531
532 let logic = dl
533 .compile_arc(&json!({"var": "payload.payload_key"}))
534 .unwrap();
535 assert_eq!(ctx.eval(&logic).unwrap(), OwnedDataValue::Null);
536 assert_eq!(ctx.eval_json(&logic).unwrap(), serde_json::Value::Null);
537 assert_eq!(ctx.eval_to_plain_string(&logic).unwrap(), "null");
538 }
539
540 #[test]
541 fn eval_json_covers_every_result_kind() {
542 let mut m = Message::from_value(&json!({}));
543 let dl = engine();
544 let ctx = TaskContext::new(&mut m, &dl);
545
546 for expected in [
547 json!(null),
548 json!(true),
549 json!(42),
550 json!(1.5),
551 json!("abc"),
552 json!([1, 2]),
553 json!({"a": 1}),
554 json!({"a": [1, {"b": "c"}], "d": {"e": [true, null]}}),
555 ] {
556 let logic = dl.compile_arc(&expected).unwrap();
557 assert_eq!(
558 ctx.eval_json(&logic).unwrap(),
559 expected,
560 "round-trip for {expected}"
561 );
562 // The owned and JSON projections agree on the same result.
563 assert_eq!(
564 serde_json::Value::from(&ctx.eval(&logic).unwrap()),
565 expected
566 );
567 }
568 }
569
570 #[test]
571 fn eval_to_plain_string_unquotes_strings_and_compacts_the_rest() {
572 let mut m = Message::from_value(&json!({}));
573 let dl = engine();
574 let ctx = TaskContext::new(&mut m, &dl);
575
576 let cases = [
577 (json!("abc"), "abc"),
578 (json!(""), ""),
579 (json!(null), "null"),
580 (json!(true), "true"),
581 (json!(42), "42"),
582 (json!({"a": 1}), "{\"a\":1}"),
583 (json!([1, 2]), "[1,2]"),
584 ];
585 for (input, expected) in cases {
586 let logic = dl.compile_arc(&input).unwrap();
587 assert_eq!(
588 ctx.eval_to_plain_string(&logic).unwrap(),
589 expected,
590 "for {input}"
591 );
592 }
593 }
594
595 #[test]
596 fn eval_to_plain_string_diverges_from_datalogics_own_string_projection() {
597 // This test IS the documentation of the divergence — it must fail if
598 // either side changes. datalogic-rs's `String: FromDataValue` keeps the
599 // JSON quoting; ours does not.
600 let mut m = Message::from_value(&json!({}));
601 let dl = engine();
602 let ctx = TaskContext::new(&mut m, &dl);
603
604 // Non-ASCII plus an embedded quote, to cover escaping too.
605 let raw = "héllo \"world\" 世界";
606 let logic = dl.compile_arc(&json!(raw)).unwrap();
607
608 // Ours: contents, byte-identical.
609 assert_eq!(ctx.eval_to_plain_string(&logic).unwrap(), raw);
610
611 // datalogic's: JSON-quoted and escaped.
612 let via_session = dl.session().eval_str(&logic, &m.context).unwrap();
613 assert_ne!(
614 via_session, raw,
615 "if these agree, the divergence this method exists for is gone"
616 );
617 assert!(
618 via_session.starts_with('"') && via_session.contains("\\\""),
619 "datalogic keeps the quoting and escaping, got: {via_session}"
620 );
621 }
622
623 #[test]
624 fn eval_surfaces_an_error_where_evaluate_condition_returns_false() {
625 // Asserted together so the difference is intentional and visible: a
626 // condition that fails should not run its task; a handler reading a
627 // config value needs to know the read failed.
628 let mut m = Message::from_value(&json!({}));
629 let dl = engine();
630
631 // `+` over a non-numeric operand fails to evaluate.
632 let bad = dl.compile_arc(&json!({"+": ["abc", 1]})).unwrap();
633
634 let condition_result =
635 crate::engine::executor::evaluate_condition(&dl, Some(&bad), &m.context);
636
637 let ctx = TaskContext::new(&mut m, &dl);
638 let eval_result = ctx.eval(&bad);
639
640 match (&condition_result, &eval_result) {
641 (Ok(false), Err(DataflowError::LogicEvaluation(msg))) => {
642 assert!(!msg.is_empty(), "the error message must not be empty");
643 }
644 other => panic!(
645 "expected evaluate_condition Ok(false) alongside eval Err(LogicEvaluation), got {other:?}"
646 ),
647 }
648 }
649
650 #[test]
651 fn consecutive_evals_and_interleaved_sets_both_work() {
652 // The arena is rewound between calls, not corrupted; and an eval between
653 // two `set`s leaves the buffered Changes intact.
654 let mut m = populated();
655 let dl = engine();
656 let first = dl.compile_arc(&json!({"var": "data.x"})).unwrap();
657 let second = dl.compile_arc(&json!({"var": "metadata.x"})).unwrap();
658
659 let mut ctx = TaskContext::new(&mut m, &dl);
660
661 assert_eq!(ctx.eval_json(&first).unwrap(), json!("dx"));
662 assert_eq!(ctx.eval_json(&second).unwrap(), json!("mx"));
663 assert_eq!(ctx.eval_json(&first).unwrap(), json!("dx"));
664
665 ctx.set("data.written", dv(json!(1)));
666 assert_eq!(ctx.eval_json(&first).unwrap(), json!("dx"));
667 ctx.set("data.written2", dv(json!(2)));
668
669 let changes = ctx.into_changes();
670 let paths: Vec<&str> = changes.iter().map(|c| &*c.path).collect();
671 assert_eq!(paths, vec!["data.written", "data.written2"]);
672 }
673
674 #[test]
675 fn eval_inside_a_with_arena_scope_falls_back_instead_of_panicking() {
676 // `TaskContext::new` is pub so a test or bench can construct one inside a
677 // `with_arena` closure. The `try_borrow_mut` fallback makes that return
678 // `Ok` on a fresh Bump rather than panicking out of the arena scope.
679 let mut m = populated();
680 let dl = engine();
681 let logic = dl.compile_arc(&json!({"var": "data.x"})).unwrap();
682
683 let got = with_arena(|_| {
684 let ctx = TaskContext::new(&mut m, &dl);
685 ctx.eval_json(&logic)
686 });
687
688 assert_eq!(got.unwrap(), json!("dx"));
689 }
690}