dataflow_rs/function/ctx.rs
1//! `Ctx`: a function's view of one call, and the static facts about it.
2
3use super::erased::BoundCall;
4use super::overlay::Overlay;
5use crate::base::{
6 AuthoredPath, CancelToken, Change, Error, ErrorKind, ErrorRecord, Instant, PathSeg, ProgramId,
7 Result, TaskIx, WorkflowIx,
8};
9use crate::expr::arena::with_arena;
10use crate::expr::path;
11use crate::expr::{
12 CRATE, Expr, ExprEngine, ExprVisitor, Filled, FromValue, Interpolated, Output, OutputOverride,
13 Path, Rendered, ResolveExpr, Resolver, SealedResolver, Seg, reads_secret,
14};
15use crate::run::{RunAccountant, TaskBudget};
16use crate::state::{Lookup, Reader, splice};
17use bumpalo::Bump;
18use datavalue::{DataValue, OwnedDataValue};
19use serde_json::Value;
20use std::borrow::Cow;
21use std::fmt;
22use std::sync::Arc;
23
24/// Static facts about one call: where it sits in the program, and which
25/// iteration it is.
26#[derive(Clone, Copy, Debug)]
27#[non_exhaustive]
28pub struct CallInfo<'a> {
29 /// The program the call belongs to. It tells the indices of one build
30 /// from another's across a reload.
31 pub program: ProgramId,
32 /// The workflow's position in the program.
33 pub workflow: WorkflowIx,
34 /// The task's position in the program.
35 pub task: TaskIx,
36 /// The name the function was called by.
37 pub function: &'a str,
38 /// The workflow's id.
39 pub workflow_id: &'a str,
40 /// The task's id.
41 pub task_id: &'a str,
42 /// The channel the run was started on; `None` when the run takes every
43 /// active workflow.
44 pub channel: Option<&'a str>,
45 /// The run's routing bucket, `0..=99`; `None` when the run was given
46 /// none, which every rollout admits. A function that starts a child run
47 /// passes it on, so the child keeps the caller's rollout decision.
48 pub routing_bucket: Option<u8>,
49 /// The loop counter, inside a workflow's loop.
50 pub loop_counter: Option<i64>,
51 /// The element's index, inside a `for_each` fan-out.
52 pub element_index: Option<usize>,
53}
54
55/// A [`Function`](super::Function)'s view of one call.
56///
57/// It reads the run's single view of the context, so no view is built per
58/// call, and keeps the call's writes in an overlay of its own: a read
59/// sees them over the run's context, and they reach the message only when
60/// the call returns `Ok` (whatever the [`Outcome`](super::Outcome)). An `Err`
61/// drops them, and their `Change`s, and keeps the records the call added.
62///
63/// Once the call has written, each evaluation splices the overlay onto the
64/// view in an arena borrowed from a pool for that evaluation only. `Ctx`
65/// holds no arena, so it is `Send` and `Sync`. A host never builds one: the
66/// engine hands it to [`Function::call`](super::Function::call) and to each
67/// [`CallLayer`](super::CallLayer), and [`testing::call`](super::testing::call)
68/// to one call outside a run:
69///
70/// ```
71/// # tokio::runtime::Builder::new_current_thread().build().unwrap().block_on(async {
72/// use dataflow_rs::datavalue::OwnedDataValue;
73/// use dataflow_rs::function::testing::{self, Write};
74/// use dataflow_rs::run::Change;
75/// use dataflow_rs::{Ctx, Function, Message, Outcome, Path, Result};
76/// use serde_json::json;
77///
78/// /// Counts visits at `data.visits`, and drops the stale `data.cache`.
79/// struct Visit;
80///
81/// impl Function for Visit {
82/// type Config = ();
83///
84/// async fn call(&self, ctx: &mut Ctx<'_>, _cfg: &()) -> Result<Outcome> {
85/// let visits: Path = "data.visits".parse()?;
86/// let seen = ctx.get(&visits).and_then(|v| v.as_i64()).unwrap_or(0);
87/// ctx.set(&visits, OwnedDataValue::from(&json!(seen + 1)))?;
88/// // The call reads its own write at once; the message, once the
89/// // call returns `Ok`.
90/// assert_eq!(ctx.get(&visits).and_then(|v| v.as_i64()), Some(seen + 1));
91/// ctx.remove(&"data.cache".parse()?)?;
92/// Ok(Outcome::Ok)
93/// }
94/// }
95///
96/// let dv = |v: serde_json::Value| OwnedDataValue::from(&v);
97/// let data = dv(json!({"visits": 1, "cache": "stale"}));
98/// let called = testing::call(Visit, json!({}))
99/// .message(Message::from_parts(OwnedDataValue::Null, data, OwnedDataValue::Null))
100/// .await;
101/// let visits: Path = "data.visits".parse().unwrap();
102/// assert_eq!(called.writes[0], (visits, Write::Set(dv(json!(2)))));
103/// assert_eq!(called.writes[1].1, Write::Removed);
104/// assert_eq!(
105/// called.changes[0],
106/// Change::write("data.visits", dv(json!(1)), dv(json!(2)))
107/// );
108/// assert_eq!(called.message.context().data(), &dv(json!({"visits": 2})));
109/// # });
110/// ```
111pub struct Ctx<'r> {
112 /// What the runtime handed the call: the run's reader, a fan-out
113 /// element's bindings (under the call's own writes), the payload, the
114 /// engine, the call's facts, the run's limits, and whether `set` and
115 /// `remove` record `Change`s.
116 start: CallStart<'r>,
117 /// The function and its config: `emit_default` writes through its
118 /// primary output, and `input` resolves its config.
119 callee: &'r dyn BoundCall,
120 /// The call's writes, in order.
121 overlay: Overlay,
122 /// The call's `Change`s, in write order.
123 changes: Vec<Change>,
124 /// The records the call added.
125 errors: Vec<ErrorRecord>,
126}
127
128/// What the runtime hands a new [`Ctx`]. A fan-out builds one for the
129/// task and copies it for each element, with the element's binding and
130/// index.
131#[derive(Clone, Copy)]
132pub(crate) struct CallStart<'r> {
133 /// The run's reader, taken in its own statement: never a `&RunView`.
134 pub(crate) reader: Reader<'r>,
135 /// A fan-out element's bindings.
136 pub(crate) binding: Option<&'r Binding<'r>>,
137 /// The message's payload.
138 pub(crate) payload: &'r OwnedDataValue,
139 /// The engine the program was compiled with.
140 pub(crate) exprs: &'r ExprEngine,
141 /// The run's shared task and expression op budget.
142 pub(crate) budget: &'r TaskBudget<'r>,
143 /// The call's static facts.
144 pub(crate) info: CallInfo<'r>,
145 /// When the run must end.
146 pub(crate) deadline: Option<Instant>,
147 /// The run's cancellation token.
148 pub(crate) cancel: Option<&'r CancelToken>,
149 /// Whether the call records its `Change`s.
150 pub(crate) capture: bool,
151}
152
153/// A callee's primary output, as [`Ctx::emit_default`] writes through it.
154#[derive(Clone, Copy, Debug)]
155pub(crate) enum Primary<'r> {
156 /// The config declares none.
157 Undeclared,
158 /// The config's primary output, the field named, which the input left
159 /// unset: an `Option<Output>`.
160 Unset(&'r str),
161 /// The config's primary output, as the call site set it.
162 Set(&'r Output),
163}
164
165/// A fan-out element's bindings: `temp_data.<as>` holds the element and
166/// `temp_data.<as>_index` its index. They sit between the run's context and
167/// the call's own writes, and are never written to the run: the settle
168/// replays the call's writes at their paths, where the binding is not.
169pub(crate) struct Binding<'r> {
170 /// `temp_data.<as>`.
171 pub(crate) item: &'r Path,
172 /// The element.
173 pub(crate) value: &'r OwnedDataValue,
174 /// `temp_data.<as>_index`.
175 pub(crate) index: &'r Path,
176 /// The element's index.
177 pub(crate) index_value: OwnedDataValue,
178}
179
180impl Binding<'_> {
181 /// `root` with both bindings written, in `arena`.
182 fn splice_onto<'a>(&'a self, root: DataValue<'a>, arena: &'a Bump) -> DataValue<'a> {
183 let root =
184 splice::set(arena, root, self.item.segs(), self.value.view_in(arena)).unwrap_or(root);
185 splice::set(
186 arena,
187 root,
188 self.index.segs(),
189 self.index_value.view_in(arena),
190 )
191 .unwrap_or(root)
192 }
193
194 /// What the bindings say `path` holds, over the run's `root`: `Some` for
195 /// a path at, below or above one of them, `None` for any other.
196 fn get<'s>(
197 &'s self,
198 path: &Path,
199 root: DataValue<'_>,
200 ) -> Option<Option<Cow<'s, OwnedDataValue>>> {
201 for (at, value) in [(self.item, self.value), (self.index, &self.index_value)] {
202 if let Some(rest) = below(path, at) {
203 return Some(path::get(value, rest).map(Cow::Borrowed));
204 }
205 }
206 if [self.item, self.index]
207 .iter()
208 .any(|at| below(at, path).is_some())
209 {
210 return Some(with_arena(|arena| {
211 splice::get(self.splice_onto(root, arena), path.segs())
212 .map(|v| Cow::Owned(v.to_owned()))
213 }));
214 }
215 None
216 }
217}
218
219/// The segments of `path` below `prefix`, when `path` is `prefix` or below
220/// it. A binding's path is a root and a key, so segments compare by key.
221fn below<'p>(path: &'p Path, prefix: &Path) -> Option<&'p [Seg]> {
222 let (path, prefix) = (path.segs(), prefix.segs());
223 (path.len() >= prefix.len() && path.iter().zip(prefix).all(|(a, b)| a.key == b.key))
224 .then(|| &path[prefix.len()..])
225}
226
227/// `f` with the root a call's writes sit on, taking an arena only when there
228/// is a binding to write.
229fn with_base<R>(
230 reader: Reader<'_>,
231 binding: Option<&Binding<'_>>,
232 f: impl FnOnce(DataValue<'_>) -> R,
233) -> R {
234 match binding {
235 None => f(*reader.root),
236 Some(binding) => with_arena(|arena| f(binding.splice_onto(*reader.root, arena))),
237 }
238}
239
240impl<'r> Ctx<'r> {
241 /// A context for one call of `callee`.
242 pub(crate) fn new(start: CallStart<'r>, callee: &'r dyn BoundCall) -> Self {
243 Self {
244 start,
245 callee,
246 overlay: Overlay::default(),
247 changes: Vec::new(),
248 errors: Vec::new(),
249 }
250 }
251
252 /// The static facts about this call.
253 pub fn info(&self) -> &CallInfo<'r> {
254 &self.start.info
255 }
256
257 /// The function this call reaches and its config, bound at compile.
258 pub(crate) fn callee(&self) -> &'r dyn BoundCall {
259 self.callee
260 }
261
262 /// The callee's config resolved for this call: every field with a value,
263 /// set by the input or by its default, keyed by its authored key, in
264 /// declaration order, an `Option` left unset absent, as
265 /// [`FunctionConfig::resolve`](super::FunctionConfig::resolve) gives it
266 /// over the context as this call sees it, a fan-out element's binding
267 /// included. A hand-written config gives what its own `resolve` does.
268 /// A field of type `f64` writes a float, `2.0` for an authored `2` too,
269 /// where a [`Templated`](super::Templated) number field writes the
270 /// number as it was authored.
271 ///
272 /// A field whose expression reads a secret is `null`, the one redaction
273 /// the engine writes, as [`TraceOptions::redact`](crate::observe::TraceOptions::redact)
274 /// writes a redacted path's value, and it is never evaluated, so not
275 /// even its error can carry the value: a logging layer or a dry run
276 /// reads the input without reading a secret. The function itself
277 /// reads its own fields, secrets included, through [`Self::eval`].
278 ///
279 /// # Errors
280 ///
281 /// The first field's that fails: [`ErrorKind::Budget`] when the
282 /// operation ceiling is exceeded, [`ErrorKind::Eval`] for any other
283 /// evaluation failure, [`ErrorKind::Config`] for a serde field that does
284 /// not serialize.
285 ///
286 /// [`Self::input_without_primary_output`] and
287 /// [`Self::input_with_primary_output`] are this resolution with the
288 /// callee's primary output left out, or set to a path the caller resolved
289 /// itself, so a layer that owns the destination policy need not evaluate
290 /// it.
291 pub fn input(&self) -> Result<OwnedDataValue> {
292 self.input_with(None)
293 }
294
295 /// [`Self::input`] with the callee's primary output left out, as an
296 /// optional output the input left unset is. A config with no primary
297 /// output, or one left unset, resolves as [`Self::input`] does.
298 ///
299 /// The output's expression is never evaluated, so a computed
300 /// destination that yields `null` does not fail the rest of the input.
301 /// Every other field is resolved and redacted as [`Self::input`]
302 /// resolves it.
303 ///
304 /// # Errors
305 ///
306 /// [`Self::input`]'s, for every field but the primary output.
307 pub fn input_without_primary_output(&self) -> Result<OwnedDataValue> {
308 self.input_with(self.primary_output().map(|out| (out, Supplied::Omit)))
309 }
310
311 /// [`Self::input`] with the callee's primary output set to `path`, its
312 /// expression never evaluated. The field is the path's dotted text, as
313 /// [`Self::input`] writes it when the expression resolves.
314 ///
315 /// A layer that resolved the destination with a policy of its own, such
316 /// as a computed `null` falling back to `data`, reads the input here and
317 /// then writes the answer through the same path, without resolving the
318 /// output a second time.
319 ///
320 /// The secret redaction [`Self::input`] applies still applies: a field
321 /// that reads a secret is `null`, and the output is `null` too when its
322 /// expression reads one, since a destination derived from a secret is
323 /// never written to an input.
324 ///
325 /// # Errors
326 ///
327 /// [`Self::input`]'s, for every field but the primary output;
328 /// [`ErrorKind::Config`] when the callee has no primary output, or leaves
329 /// an optional one unset, which the error names.
330 ///
331 /// ```
332 /// # tokio::runtime::Builder::new_current_thread().build().unwrap().block_on(async {
333 /// use dataflow_rs::datavalue::OwnedDataValue;
334 /// use dataflow_rs::expr::Output;
335 /// use dataflow_rs::function::{BoxFuture, CallLayer, FunctionConfig, Next, testing};
336 /// use dataflow_rs::{Ctx, Expr, Function, Message, Outcome, Result};
337 /// use serde_json::json;
338 ///
339 /// #[derive(FunctionConfig)]
340 /// struct Fetch {
341 /// url: Expr<String>,
342 /// output: Output,
343 /// }
344 ///
345 /// struct FetchFn;
346 ///
347 /// impl Function for FetchFn {
348 /// type Config = Fetch;
349 ///
350 /// async fn call(&self, _ctx: &mut Ctx<'_>, _cfg: &Fetch) -> Result<Outcome> {
351 /// Ok(Outcome::Ok)
352 /// }
353 /// }
354 ///
355 /// /// Resolves the destination with a policy of its own, then logs the
356 /// /// redacted input with that destination.
357 /// struct Log;
358 ///
359 /// impl CallLayer for Log {
360 /// fn call<'a>(
361 /// &'a self,
362 /// ctx: &'a mut Ctx<'_>,
363 /// next: Next<'a>,
364 /// ) -> BoxFuture<'a, Result<Outcome>> {
365 /// Box::pin(async move {
366 /// let Some(out) = ctx.primary_output() else {
367 /// return next.run(ctx).await;
368 /// };
369 /// let path = ctx.eval(out.path())?.into_owned();
370 /// let input = ctx.input_with_primary_output(&path)?;
371 /// assert_eq!(
372 /// input.to_json_string(),
373 /// r#"{"url":"https://x","output":"data.response"}"#
374 /// );
375 /// next.run(ctx).await
376 /// })
377 /// }
378 /// }
379 ///
380 /// let called = testing::call(
381 /// FetchFn,
382 /// json!({"url": "https://x", "output": {"var": "data.where"}}),
383 /// )
384 /// .message(Message::from_parts(
385 /// OwnedDataValue::Null,
386 /// OwnedDataValue::from(&json!({"where": "data.response"})),
387 /// OwnedDataValue::Null,
388 /// ))
389 /// .layer(Log)
390 /// .await;
391 /// assert!(called.outcome.is_ok());
392 /// # });
393 /// ```
394 pub fn input_with_primary_output(&self, path: &Path) -> Result<OwnedDataValue> {
395 let out = self.require_primary()?;
396 self.input_with(Some((out, Supplied::At(path))))
397 }
398
399 /// [`Self::input`] with `primary` supplying the callee's primary output.
400 fn input_with(&self, primary: Option<(&'r Output, Supplied<'_>)>) -> Result<OwnedDataValue> {
401 let mut resolver = Redacted { ctx: self, primary };
402 let mut input = self.callee.resolve(&mut resolver)?;
403 // The walk the secret checks trust: whatever a field's resolve did,
404 // the value where its secret-reading expression sits is `null`.
405 self.callee.visit(&mut Redact(&mut input));
406 Ok(input)
407 }
408
409 /// `cfg` resolved for this call, unredacted:
410 /// [`FunctionConfig::resolve`](super::FunctionConfig::resolve) over the
411 /// context as this call sees it, a fan-out element's binding included,
412 /// with every expression evaluated, a secret-reading one too. It is
413 /// [`Self::input`]'s counterpart: `input` is what a layer or a dry run
414 /// reads, and this is the function's own view of its config, which it
415 /// may hand on whole.
416 ///
417 /// # Errors
418 ///
419 /// The first field's that fails: [`ErrorKind::Budget`] when the
420 /// operation ceiling is exceeded, [`ErrorKind::Eval`] for any other
421 /// evaluation failure, [`ErrorKind::Config`] for a serde field that does
422 /// not serialize.
423 pub fn resolve<C: super::FunctionConfig>(&self, cfg: &C) -> Result<OwnedDataValue> {
424 cfg.resolve(&mut &*self)
425 }
426
427 /// The message's payload, which no expression reads.
428 pub fn payload(&self) -> &OwnedDataValue {
429 self.start.payload
430 }
431
432 /// The engine the program was compiled with, for an evaluation over a
433 /// root of the function's own.
434 pub fn exprs(&self) -> &ExprEngine {
435 self.start.exprs
436 }
437
438 /// The secret at the dotted path `name`, from the engine's store. What a
439 /// function does with it is the function's: the engine never records it.
440 pub fn secret(&self, name: &str) -> Option<&OwnedDataValue> {
441 self.start.exprs.secrets().get(name)
442 }
443
444 /// When the run must end, if it has a deadline: a run this function
445 /// starts passes it on, or an earlier one. An `Err` this call returns
446 /// once the deadline has passed, such as that run's own `TimedOut`, ends
447 /// this run `TimedOut` too.
448 pub fn deadline(&self) -> Option<Instant> {
449 self.start.deadline
450 }
451
452 /// The run's cancellation token, if it has one: a child of it cancels a
453 /// run this function starts along with this one, so that run ends
454 /// `Cancelled` and reports it rather than being dropped. An `Err` this
455 /// call returns once the token is cancelled ends this run `Cancelled`.
456 pub fn cancel(&self) -> Option<&CancelToken> {
457 self.start.cancel
458 }
459
460 /// The host-owned [`RunAccountant`] attached to this run, if any, so a
461 /// function can charge domain-specific work units via
462 /// [`RunAccountant::charge_custom`].
463 pub fn accountant(&self) -> Option<&'r dyn RunAccountant> {
464 self.start.budget.accountant()
465 }
466
467 /// The value at `path` as this call sees it: its own writes, over a
468 /// fan-out element's bindings, over the run's context. Borrowed unless
469 /// it must be put together, when it is converted: a value only the run's
470 /// view knows, or one with this call's writes or a binding below `path`,
471 /// such as `temp_data` while `temp_data.p` is bound.
472 pub fn get(&self, path: &Path) -> Option<Cow<'_, OwnedDataValue>> {
473 self.find(path).map(Filled::into_cow)
474 }
475
476 /// Where the value at `path` is, as this call sees it, lent from
477 /// wherever it is: the call's writes, the binding, then the run's log,
478 /// owned tree or view. Only a value the call wrote below `path` is
479 /// converted, off the root `eval` sees.
480 fn find(&self, path: &Path) -> Option<Filled<'_>> {
481 match self.overlay.lookup(path) {
482 Lookup::Clean => {}
483 Lookup::Owned(v) => return v.map(Filled::lent),
484 Lookup::Dirty => {
485 return with_arena(|arena| {
486 splice::get(*self.root_in(arena), path.segs())
487 .map(|v| Filled::Owned(Cow::Owned(v.to_owned())))
488 });
489 }
490 }
491 if let Some(found) = self
492 .start
493 .binding
494 .and_then(|b| b.get(path, *self.start.reader.root))
495 {
496 return found.map(Filled::Owned);
497 }
498 self.start.reader.find(path)
499 }
500
501 /// `doc` with each hole looked up once, as [`Self::get`] would, and
502 /// lent where it was found; a hole that finds nothing yields its
503 /// default. No tree is built: the [`Rendered`] serializes the
504 /// document's skeleton with each value in place, so a consumer that is
505 /// itself a serializer pays for no intermediate value.
506 pub fn render<'s>(&'s self, doc: &'s Interpolated) -> Rendered<'s> {
507 doc.render_with(|hole| {
508 hole.path
509 .as_ref()
510 .and_then(|path| self.find(path))
511 .unwrap_or(Filled::Default(&hole.default))
512 })
513 }
514
515 /// Whether this call sees a value at `path`, converting none.
516 fn holds(&self, path: &Path) -> bool {
517 match self.overlay.lookup(path) {
518 Lookup::Owned(v) => v.is_some(),
519 Lookup::Dirty => {
520 with_arena(|arena| splice::get(*self.root_in(arena), path.segs()).is_some())
521 }
522 Lookup::Clean => with_base(self.start.reader, self.start.binding, |base| {
523 splice::get(base, path.segs()).is_some()
524 }),
525 }
526 }
527
528 /// Evaluates `expr` against the context as this call sees it. A folded
529 /// constant is lent from `expr` and takes no arena; anything else is
530 /// evaluated in a pooled arena and converted to `T` before the arena goes
531 /// back.
532 ///
533 /// # Errors
534 ///
535 /// [`ErrorKind::Budget`](crate::base::ErrorKind::Budget) when the
536 /// operation ceiling is exceeded; [`ErrorKind::Eval`](crate::base::ErrorKind::Eval)
537 /// for any other failure, including a result that is not a `T`.
538 pub fn eval<'e, T: FromValue>(&self, expr: &'e Expr<T>) -> Result<T::Out<'e>> {
539 if let Some(c) = expr.constant() {
540 return Ok(T::lend(c));
541 }
542 with_arena(|arena| {
543 let root = self.root_in(arena);
544 self.start
545 .budget
546 .eval_view(self.start.exprs, expr, root, arena)
547 })
548 }
549
550 /// `expr` evaluated as [`Self::eval`] evaluates it, as an owned value: a
551 /// folded constant converted, anything else evaluated in a pooled arena
552 /// and converted before the arena goes back.
553 fn eval_owned(&self, expr: &dyn ResolveExpr) -> Result<OwnedDataValue> {
554 if let Some(constant) = expr.constant_value(CRATE) {
555 return Ok(constant);
556 }
557 with_arena(|arena| {
558 let root = self.root_in(arena);
559 let (view, ops) = self.start.budget.eval_in(
560 self.start.exprs,
561 expr.logic(CRATE),
562 expr.reads_secret(CRATE),
563 root,
564 arena,
565 )?;
566 expr.value_of(view, ops, CRATE)
567 })
568 }
569
570 /// The root this call's evaluations see: the run's view, with the binding
571 /// and then the call's writes spliced on in `arena` when there are any.
572 fn root_in<'a>(&'a self, arena: &'a Bump) -> &'a DataValue<'a> {
573 if self.start.binding.is_none() && self.overlay.is_empty() {
574 return self.start.reader.root;
575 }
576 let base = match self.start.binding {
577 None => *self.start.reader.root,
578 Some(binding) => binding.splice_onto(*self.start.reader.root, arena),
579 };
580 if self.overlay.is_empty() {
581 return arena.alloc(base);
582 }
583 arena.alloc(self.overlay.splice_onto(base, arena))
584 }
585
586 /// Assigns `value` at `path`, by `Path::set`'s rules: a hop that cannot
587 /// continue writes nothing. The `Change` is recorded first, under
588 /// capture, with the value this call sees there now, so a second write to
589 /// one path leaves two `Change`s and one value. At a root the write
590 /// replaces the root, and its `Change` says so
591 /// ([`replaced`](Change::replaced)).
592 ///
593 /// # Errors
594 ///
595 /// [`ErrorKind::Config`](crate::base::ErrorKind::Config) when `path` is a
596 /// root and `value` is not an object.
597 pub fn set(&mut self, path: &Path, value: OwnedDataValue) -> Result<()> {
598 path.check_set(&value)?;
599 // A first write with no binding is resolved against the run's view
600 // once: the walk says whether the write lands, and whether anything
601 // is there to record as the old value.
602 let first = (self.overlay.is_empty() && self.start.binding.is_none())
603 .then(|| splice::resolve(*self.start.reader.root, path.segs()));
604 if self.start.capture {
605 let old_value = match first {
606 // Below a scalar, or creating the node at the path or above
607 // it: nothing is there, or a `null` element, read as `null`.
608 Some(None) => None,
609 Some(Some(r)) if r.created.is_some() => None,
610 Some(Some(r)) => self.start.reader.get_resolved(path, r),
611 None => self.get(path),
612 };
613 self.changes.push(Change::assign(
614 path,
615 old_value.map_or(OwnedDataValue::Null, Cow::into_owned),
616 value.clone(),
617 ));
618 }
619 if let Some(resolved) = first {
620 self.overlay.push(path, value, resolved.is_some());
621 return Ok(());
622 }
623 let Self { start, overlay, .. } = self;
624 with_base(start.reader, start.binding, |base| {
625 overlay.set(base, path, value)
626 });
627 Ok(())
628 }
629
630 /// Writes `value` through `out`, as [`Self::set`] does: at its path
631 /// when it is literal, else at the path it evaluates to now, against the
632 /// context as this call sees it.
633 ///
634 /// # Errors
635 ///
636 /// [`Self::eval`]'s, when `out` is computed and does not give a path;
637 /// [`Self::set`]'s.
638 pub fn emit(&mut self, out: &Output, value: OwnedDataValue) -> Result<()> {
639 let path = match out.constant() {
640 Some(path) => Cow::Borrowed(path),
641 None => self.eval(out.path())?,
642 };
643 self.set(&path, value)
644 }
645
646 /// The callee's primary output, the one [`Self::emit_default`] writes
647 /// through, as the call site set it: `None` when the config declares
648 /// none, or leaves an optional one unset. A layer reads it to know
649 /// whether, and where, an answer would be written. It borrows the
650 /// callee, not this context, so a layer can [`Self::emit`] through it.
651 pub fn primary_output(&self) -> Option<&'r Output> {
652 let callee: &'r dyn BoundCall = self.callee;
653 match callee.primary() {
654 Primary::Set(out) => Some(out),
655 Primary::Unset(_) | Primary::Undeclared => None,
656 }
657 }
658
659 /// [`Self::emit`] through the function's primary output, the one its
660 /// config's [`FunctionConfig::primary_output`](super::FunctionConfig::primary_output)
661 /// names.
662 ///
663 /// # Errors
664 ///
665 /// [`ErrorKind::Config`] when the config has no primary output, or
666 /// leaves an optional one unset, which the error names; [`Self::emit`]'s.
667 pub fn emit_default(&mut self, value: OwnedDataValue) -> Result<()> {
668 let out = self.require_primary()?;
669 self.emit(out, value)
670 }
671
672 /// The callee's primary output, or the `Config` error `emit_default`
673 /// gives when the config has none, or leaves an optional one unset.
674 fn require_primary(&self) -> Result<&'r Output> {
675 let function = self.start.info.function;
676 match self.callee.primary() {
677 Primary::Set(out) => Ok(out),
678 Primary::Unset(field) => Err(Error::new(
679 ErrorKind::Config,
680 format!("`{function}`'s primary output `{field}` is not set"),
681 )),
682 Primary::Undeclared => Err(Error::new(
683 ErrorKind::Config,
684 format!("`{function}` has no primary output"),
685 )),
686 }
687 }
688
689 /// Removes the value at `path`, recording the removal under capture.
690 /// Nothing there is nothing to do, and records nothing.
691 ///
692 /// # Errors
693 ///
694 /// [`ErrorKind::Config`](crate::base::ErrorKind::Config) when `path` is a
695 /// root.
696 pub fn remove(&mut self, path: &Path) -> Result<()> {
697 path.check_remove()?;
698 if self.start.capture {
699 let Some(old_value) = self.get(path).map(Cow::into_owned) else {
700 return Ok(());
701 };
702 self.changes
703 .push(Change::removal(Arc::clone(path.dotted()), old_value));
704 } else if !self.holds(path) {
705 return Ok(());
706 }
707 self.overlay.remove(path);
708 Ok(())
709 }
710
711 /// Adds a record to the message's errors, whatever the call returns. What
712 /// the record leaves unset the engine fills: the workflow's and the
713 /// task's ids, the run's timestamp, and in a fan-out the element's index.
714 pub fn record_error(&mut self, e: ErrorRecord) {
715 self.errors.push(e);
716 }
717
718 /// The call's writes, its `Change`s and the records it added.
719 pub(crate) fn settle(self) -> (Overlay, Vec<Change>, Vec<ErrorRecord>) {
720 (self.overlay, self.changes, self.errors)
721 }
722}
723
724/// A call's context resolves a config's fields against the context as the
725/// call sees it, as [`Ctx::eval`] and [`Ctx::render`] do, secrets read: a
726/// host's [`ConfigField::resolve`](crate::expr::ConfigField::resolve) calls
727/// it. A function resolves its whole config with [`Ctx::resolve`].
728impl SealedResolver for &Ctx<'_> {}
729
730impl Resolver for &Ctx<'_> {
731 fn eval(&mut self, expr: &dyn ResolveExpr) -> Result<OwnedDataValue> {
732 Ctx::eval_owned(self, expr)
733 }
734
735 fn render(&mut self, doc: &Interpolated) -> Result<OwnedDataValue> {
736 Ok(Ctx::render(self, doc).to_owned())
737 }
738}
739
740/// How a call's input resolves the callee's primary output, when the caller
741/// supplied it.
742#[derive(Clone, Copy)]
743enum Supplied<'p> {
744 /// Leave the field out.
745 Omit,
746 /// Resolve it to this path, its expression never evaluated.
747 At(&'p Path),
748}
749
750/// The resolver behind [`Ctx::input`]: an expression that reads a secret is
751/// `null`, and is never evaluated; anything else as the call's context
752/// resolves it, and the callee's primary output as the caller supplied it.
753struct Redacted<'c, 'r, 'p> {
754 ctx: &'c Ctx<'r>,
755 /// The callee's primary output and what to resolve it to, when the
756 /// caller supplied it.
757 primary: Option<(&'r Output, Supplied<'p>)>,
758}
759
760impl SealedResolver for Redacted<'_, '_, '_> {}
761
762impl Resolver for Redacted<'_, '_, '_> {
763 fn eval(&mut self, expr: &dyn ResolveExpr) -> Result<OwnedDataValue> {
764 if reads_secret(expr.raw()) {
765 return Ok(OwnedDataValue::Null);
766 }
767 self.ctx.eval_owned(expr)
768 }
769
770 fn render(&mut self, doc: &Interpolated) -> Result<OwnedDataValue> {
771 Ok(self.ctx.render(doc).to_owned())
772 }
773
774 fn output_override(&mut self, out: &Output) -> Option<OutputOverride> {
775 let (primary, supplied) = self.primary?;
776 if !std::ptr::eq(out, primary) {
777 return None;
778 }
779 Some(match supplied {
780 Supplied::Omit => OutputOverride::Omit,
781 Supplied::At(path) => OutputOverride::Value(OwnedDataValue::String(path.to_string())),
782 })
783 }
784}
785
786/// Sets the value of every expression that reads a secret to `null`, at the
787/// path the config's walk reports it, in a resolved config.
788struct Redact<'v>(&'v mut OwnedDataValue);
789
790impl ExprVisitor for Redact<'_> {
791 fn expr(&mut self, at: &AuthoredPath, raw: &Value, _records: bool) {
792 if reads_secret(raw)
793 && let Some(slot) = slot_at(self.0, at)
794 {
795 *slot = OwnedDataValue::Null;
796 }
797 }
798}
799
800/// The value at `at` in `value`, a key naming an object's entry and an
801/// index an array's element; `None` when nothing is there.
802fn slot_at<'v>(value: &'v mut OwnedDataValue, at: &AuthoredPath) -> Option<&'v mut OwnedDataValue> {
803 at.segs()
804 .iter()
805 .try_fold(value, |value, seg| match (value, seg) {
806 (OwnedDataValue::Object(entries), PathSeg::Key(key)) => entries
807 .iter_mut()
808 .find(|(k, _)| **k == **key)
809 .map(|(_, v)| v),
810 (OwnedDataValue::Array(items), PathSeg::Index(i)) => items.get_mut(*i),
811 _ => None,
812 })
813}
814
815impl fmt::Debug for Ctx<'_> {
816 /// The call's facts and whether it has written; never the context.
817 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
818 f.debug_struct("Ctx")
819 .field("info", &self.start.info)
820 .field("overlay_empty", &self.overlay.is_empty())
821 .finish_non_exhaustive()
822 }
823}
824
825/// A `Ctx` over a context of the tests' choosing.
826#[cfg(test)]
827pub(crate) mod testing {
828 use super::*;
829 use crate::function::{BoxFuture, Outcome};
830 use crate::state::{RunContext, Tuning};
831
832 /// A callee with no config: no primary output, and nothing to resolve.
833 pub(crate) struct NoCallee;
834
835 impl BoundCall for NoCallee {
836 fn call<'a>(&'a self, _ctx: &'a mut Ctx<'_>) -> BoxFuture<'a, Result<Outcome>> {
837 Box::pin(std::future::ready(Ok(Outcome::Ok)))
838 }
839
840 fn visit(&self, _v: &mut dyn ExprVisitor) {}
841
842 fn primary(&self) -> Primary<'_> {
843 Primary::Undeclared
844 }
845
846 fn resolve(&self, _r: &mut dyn Resolver) -> Result<OwnedDataValue> {
847 Ok(OwnedDataValue::Null)
848 }
849 }
850
851 /// Runs `f` with a call context over `context`, the call described by
852 /// `info`, with `binding` if any.
853 pub(crate) fn with_ctx<R>(
854 context: &serde_json::Value,
855 info: CallInfo<'_>,
856 binding: Option<&Binding<'_>>,
857 f: impl FnOnce(Ctx<'_>) -> R,
858 ) -> R {
859 let exprs = ExprEngine::default();
860 let budget = TaskBudget::unlimited();
861 let payload = OwnedDataValue::Null;
862 let mut slot = OwnedDataValue::from(context);
863 let mut rc = RunContext::new(&mut slot, Tuning::default());
864 let reader = rc.reader();
865 f(Ctx::new(
866 CallStart {
867 reader,
868 binding,
869 payload: &payload,
870 exprs: &exprs,
871 budget: &budget,
872 info,
873 deadline: None,
874 cancel: None,
875 capture: true,
876 },
877 &NoCallee,
878 ))
879 }
880
881 /// Runs `f` with a call context over `context` and `callee`, the call
882 /// described by `info`, as a registry binds one.
883 pub(crate) fn with_call<R>(
884 context: &serde_json::Value,
885 info: CallInfo<'_>,
886 callee: &dyn BoundCall,
887 f: impl FnOnce(Ctx<'_>) -> R,
888 ) -> R {
889 let exprs = ExprEngine::default();
890 let budget = TaskBudget::unlimited();
891 let payload = OwnedDataValue::Null;
892 let mut slot = OwnedDataValue::from(context);
893 let mut rc = RunContext::new(&mut slot, Tuning::default());
894 let reader = rc.reader();
895 f(Ctx::new(
896 CallStart {
897 reader,
898 binding: None,
899 payload: &payload,
900 exprs: &exprs,
901 budget: &budget,
902 info,
903 deadline: None,
904 cancel: None,
905 capture: true,
906 },
907 callee,
908 ))
909 }
910}
911
912#[cfg(test)]
913mod tests {
914 use super::testing::{NoCallee, with_call, with_ctx};
915 use super::*;
916 use crate::state::{RunContext, Tuning};
917 use serde_json::{Value, json};
918
919 fn send_sync<T: Send + Sync>() {}
920
921 #[test]
922 fn ctx_is_send_and_sync() {
923 send_sync::<Ctx<'static>>();
924 }
925
926 fn info() -> CallInfo<'static> {
927 CallInfo {
928 program: ProgramId::next(),
929 workflow: WorkflowIx::new(0),
930 task: TaskIx::new(0),
931 function: "f",
932 workflow_id: "w",
933 task_id: "t",
934 channel: None,
935 routing_bucket: None,
936 loop_counter: None,
937 element_index: None,
938 }
939 }
940
941 fn parse(dotted: &str) -> Path {
942 dotted.parse().unwrap()
943 }
944
945 fn get(ctx: &Ctx<'_>, dotted: &str) -> Option<Value> {
946 ctx.get(&parse(dotted)).map(|v| Value::from(&*v))
947 }
948
949 fn eval(ctx: &Ctx<'_>, raw: Value) -> Value {
950 let expr: Expr = ExprEngine::default().compile(&raw).unwrap();
951 Value::from(&*ctx.eval(&expr).unwrap())
952 }
953
954 #[test]
955 fn the_binding_sits_under_the_calls_writes() {
956 let item = parse("temp_data.item");
957 let index = parse("temp_data.item_index");
958 let value = OwnedDataValue::from(&json!({"sku": "A", "qty": 2}));
959 let binding = Binding {
960 item: &item,
961 value: &value,
962 index: &index,
963 index_value: OwnedDataValue::from(3),
964 };
965 let context = json!({"data": {}, "metadata": {}, "temp_data": {"kept": true}});
966 with_ctx(&context, info(), Some(&binding), |mut ctx| {
967 assert_eq!(get(&ctx, "temp_data.item.sku"), Some(json!("A")));
968 assert_eq!(get(&ctx, "temp_data.item_index"), Some(json!(3)));
969 assert_eq!(
970 get(&ctx, "temp_data"),
971 Some(json!({"kept": true, "item": {"sku": "A", "qty": 2}, "item_index": 3}))
972 );
973 assert_eq!(eval(&ctx, json!({"var": "temp_data.item.qty"})), json!(2));
974 // A write below a binding is seen over it, by `get` and `eval`.
975 ctx.set(&parse("temp_data.item.qty"), OwnedDataValue::from(5))
976 .unwrap();
977 assert_eq!(
978 get(&ctx, "temp_data.item"),
979 Some(json!({"sku": "A", "qty": 5}))
980 );
981 assert_eq!(eval(&ctx, json!({"var": "temp_data.item.qty"})), json!(5));
982 assert_eq!(get(&ctx, "temp_data.kept"), Some(json!(true)));
983 });
984 }
985
986 #[test]
987 fn a_write_that_cannot_land_records_its_change_and_writes_nothing() {
988 // v3's `TaskContext::set` records before it writes. Removing nothing
989 // records nothing.
990 let context = json!({"data": {"s": "scalar"}, "metadata": {}, "temp_data": {}});
991 let (overlay, changes, errors) = with_ctx(&context, info(), None, |mut ctx| {
992 ctx.set(&parse("data.s.below"), OwnedDataValue::from(1))
993 .unwrap();
994 ctx.remove(&parse("data.absent")).unwrap();
995 assert_eq!(get(&ctx, "data.s"), Some(json!("scalar")));
996 ctx.settle()
997 });
998 assert!(errors.is_empty());
999 // The write is kept for the settle, which replays it where the
1000 // context may take it, as v3 replays a `Change`; here it cannot.
1001 let mut slot = OwnedDataValue::from(&context);
1002 let mut rc = RunContext::new(&mut slot, Tuning::default());
1003 overlay.apply(&mut rc);
1004 rc.finish();
1005 assert_eq!(Value::from(&slot), context);
1006 assert_eq!(
1007 serde_json::to_value(&changes).unwrap(),
1008 json!([{"path": "data.s.below", "old_value": null, "new_value": 1}])
1009 );
1010 }
1011
1012 #[test]
1013 fn a_removal_without_capture_removes_what_the_call_sees_and_records_nothing() {
1014 let context = json!({"data": {"a": 1, "list": [1, 2]}, "metadata": {}, "temp_data": {}});
1015 let exprs = ExprEngine::default();
1016 let budget = TaskBudget::unlimited();
1017 let payload = OwnedDataValue::Null;
1018 let mut slot = OwnedDataValue::from(&context);
1019 let mut rc = RunContext::new(&mut slot, Tuning::default());
1020 let mut ctx = Ctx::new(
1021 CallStart {
1022 reader: rc.reader(),
1023 binding: None,
1024 payload: &payload,
1025 exprs: &exprs,
1026 budget: &budget,
1027 info: info(),
1028 deadline: None,
1029 cancel: None,
1030 capture: false,
1031 },
1032 &NoCallee,
1033 );
1034 // A value of the run's, one the call wrote, nothing, and an element
1035 // the call's own removal moved: after it, `list.1` holds nothing.
1036 for (dotted, write) in [
1037 ("data.a", None),
1038 ("data.b", Some(2)),
1039 ("data.b", None),
1040 ("data.absent", None),
1041 ("data.list.0", None),
1042 ("data.list.1", None),
1043 ] {
1044 match write {
1045 Some(n) => ctx.set(&parse(dotted), OwnedDataValue::from(n)),
1046 None => ctx.remove(&parse(dotted)),
1047 }
1048 .unwrap();
1049 }
1050 assert_eq!(get(&ctx, "data"), Some(json!({"list": [2]})));
1051 let (overlay, changes, errors) = ctx.settle();
1052 assert!(changes.is_empty() && errors.is_empty());
1053 overlay.apply(&mut rc);
1054 rc.finish();
1055 assert_eq!(Value::from(&slot)["data"], json!({"list": [2]}));
1056 }
1057
1058 /// What each call recorded and wrote, when each of `writes` is made by a
1059 /// call of its own, as its first write or, with `after_another`, after
1060 /// a write elsewhere: the two ways `Ctx::set` reads the old value. The
1061 /// run has written `data.run` into its view, so its log holds it.
1062 fn first_writes(writes: &[&str], after_another: bool) -> (Vec<Value>, String) {
1063 let context = json!({"data": {"kept": {"k": 1}, "list": [1, null], "s": "scalar"},
1064 "metadata": {}, "temp_data": {}});
1065 let exprs = ExprEngine::default();
1066 let budget = TaskBudget::unlimited();
1067 let payload = OwnedDataValue::Null;
1068 let mut slot = OwnedDataValue::from(&context);
1069 let mut rc = RunContext::new(&mut slot, Tuning::default());
1070 rc.read(|_, _| ());
1071 rc.set(&parse("data.run"), OwnedDataValue::from(7), None);
1072 let mut olds = Vec::new();
1073 let mut overlays = Vec::new();
1074 for dotted in writes {
1075 let mut ctx = Ctx::new(
1076 CallStart {
1077 reader: rc.reader(),
1078 binding: None,
1079 payload: &payload,
1080 exprs: &exprs,
1081 budget: &budget,
1082 info: info(),
1083 deadline: None,
1084 cancel: None,
1085 capture: true,
1086 },
1087 &NoCallee,
1088 );
1089 if after_another {
1090 ctx.set(&parse("temp_data.other"), OwnedDataValue::from(0))
1091 .unwrap();
1092 }
1093 ctx.set(&parse(dotted), OwnedDataValue::from("w")).unwrap();
1094 let (overlay, changes, _) = ctx.settle();
1095 olds.push(serde_json::to_value(&changes.last().unwrap().old_value).unwrap());
1096 overlays.push(overlay);
1097 }
1098 for overlay in overlays {
1099 overlay.apply(&mut rc);
1100 }
1101 rc.finish();
1102 let data = parse("data").get(&slot).map(serde_json::to_string);
1103 (olds, data.unwrap().unwrap())
1104 }
1105
1106 #[test]
1107 fn a_first_write_reads_what_was_there_as_a_later_write_does() {
1108 // In the run's log, in the owned tree, missing below a missing key, a
1109 // `null` element and below it, and below a scalar, where nothing lands.
1110 // Settled in order, the write below `list.1` finds the string the
1111 // write before it put there, and lands nowhere.
1112 let writes = [
1113 "data.run",
1114 "data.kept.k",
1115 "data.absent.x",
1116 "data.list.1",
1117 "data.list.1.x",
1118 "data.s.below",
1119 ];
1120 let (first, data) = first_writes(&writes, false);
1121 assert_eq!(
1122 first,
1123 vec![
1124 json!(7),
1125 json!(1),
1126 json!(null),
1127 json!(null),
1128 json!(null),
1129 json!(null)
1130 ]
1131 );
1132 assert_eq!(first_writes(&writes, true), (first, data.clone()));
1133 assert_eq!(
1134 data,
1135 r#"{"kept":{"k":"w"},"list":[1,"w"],"s":"scalar","run":"w","absent":{"x":"w"}}"#
1136 );
1137 }
1138
1139 /// One call's writes: `(path, Some(value))`, or `(path, None)` for a
1140 /// removal.
1141 type Call<'a> = Vec<(&'a str, Option<Value>)>;
1142
1143 /// The run context over `context` after each call's writes are made
1144 /// through a `Ctx` of its own over the run's one view, and the overlays
1145 /// applied in call order, as a fan-out's elements settle. Wire text.
1146 fn settled(context: &Value, calls: &[Call<'_>]) -> String {
1147 let exprs = ExprEngine::default();
1148 let budget = TaskBudget::unlimited();
1149 let payload = OwnedDataValue::Null;
1150 let mut slot = OwnedDataValue::from(context);
1151 let mut rc = RunContext::new(&mut slot, Tuning::default());
1152 let mut overlays = Vec::new();
1153 for writes in calls {
1154 let mut ctx = Ctx::new(
1155 CallStart {
1156 reader: rc.reader(),
1157 binding: None,
1158 payload: &payload,
1159 exprs: &exprs,
1160 budget: &budget,
1161 info: info(),
1162 deadline: None,
1163 cancel: None,
1164 capture: true,
1165 },
1166 &NoCallee,
1167 );
1168 for (dotted, value) in writes {
1169 match value {
1170 Some(value) => ctx.set(&parse(dotted), OwnedDataValue::from(value)),
1171 None => ctx.remove(&parse(dotted)),
1172 }
1173 .unwrap();
1174 }
1175 overlays.push(ctx.settle().0);
1176 }
1177 for overlay in overlays {
1178 overlay.apply(&mut rc);
1179 }
1180 rc.finish();
1181 serde_json::to_string(&slot).unwrap()
1182 }
1183
1184 /// Calls planned over one base, then settled in order, end as one call
1185 /// making all their writes in that order would: each write lands at its
1186 /// own path on the writes before it. Two keys under a parent neither
1187 /// call's base has, then the three cases a write of a whole container
1188 /// would lose: padding, an element's removal, a key written after its
1189 /// removal.
1190 #[test]
1191 fn calls_settled_in_order_end_as_one_call_making_their_writes() {
1192 // The data root before, each call's writes, and the data root after
1193 // as wire text: key order is part of what is checked.
1194 let cases: [(Value, Vec<Call<'_>>, &str); 4] = [
1195 (
1196 json!({}),
1197 vec![
1198 vec![("data.prices.a", Some(json!(1)))],
1199 vec![("data.prices.b", Some(json!(2)))],
1200 ],
1201 r#"{"prices":{"a":1,"b":2}}"#,
1202 ),
1203 (
1204 json!({"list": [1, 2]}),
1205 vec![
1206 vec![("data.list.4", Some(json!(5)))],
1207 vec![("data.list.3", Some(json!(4)))],
1208 ],
1209 r#"{"list":[1,2,null,4,5]}"#,
1210 ),
1211 (
1212 json!({"list": [1, 2, 3]}),
1213 vec![vec![("data.list.0", None)], vec![("data.list.2", None)]],
1214 r#"{"list":[2,3]}"#,
1215 ),
1216 (
1217 json!({"a": 1, "b": 2}),
1218 vec![
1219 vec![("data.b", Some(json!(5)))],
1220 vec![("data.a", None), ("data.a", Some(json!(3)))],
1221 ],
1222 r#"{"b":5,"a":3}"#,
1223 ),
1224 ];
1225 for (data, calls, want) in cases {
1226 let context = json!({"data": data, "metadata": {}, "temp_data": {}});
1227 let want = format!(r#"{{"data":{want},"metadata":{{}},"temp_data":{{}}}}"#);
1228 let one_call = vec![calls.concat()];
1229 assert_eq!(settled(&context, &one_call), want, "one call: {calls:?}");
1230 assert_eq!(
1231 settled(&context, &calls),
1232 want,
1233 "settled in order: {calls:?}"
1234 );
1235 }
1236 }
1237
1238 /// A fan-out's calls, each over the run's one view with its own
1239 /// element bound, each writing under its binding and under `data`,
1240 /// then settled in element order onto a run where no binding is: each
1241 /// call reads its writes over its element, and each write lands at its
1242 /// own path on the writes settled before it, as v3 replays an element's
1243 /// `Change`s onto a message that holds no binding. A write below a
1244 /// scalar element is blocked in its call and lands at the settle.
1245 #[test]
1246 fn calls_with_bindings_settled_in_order_replay_each_write_at_its_path() {
1247 let context = json!({"data": {}, "metadata": {}, "temp_data": {}});
1248 let item = parse("temp_data.p");
1249 let index = parse("temp_data.p_index");
1250 // Each element, its writes, what the call then sees at
1251 // `temp_data.p`, and the old values its `Change`s record.
1252 let elements: [(Value, Call<'_>, Value, [Value; 2]); 3] = [
1253 (
1254 json!({"id": "a"}),
1255 vec![
1256 ("temp_data.p.x", Some(json!(1))),
1257 ("data.a", Some(json!(1))),
1258 ],
1259 json!({"id": "a", "x": 1}),
1260 [json!(null), json!(null)],
1261 ),
1262 (
1263 json!(5),
1264 vec![
1265 ("temp_data.p.y", Some(json!(2))),
1266 ("data.b", Some(json!(2))),
1267 ],
1268 json!(5),
1269 [json!(null), json!(null)],
1270 ),
1271 (
1272 json!({"id": "c"}),
1273 vec![
1274 ("temp_data.p.id", Some(json!("C"))),
1275 ("data.a", Some(json!(3))),
1276 ],
1277 json!({"id": "C"}),
1278 [json!("c"), json!(null)],
1279 ),
1280 ];
1281 let exprs = ExprEngine::default();
1282 let budget = TaskBudget::unlimited();
1283 let payload = OwnedDataValue::Null;
1284 let mut slot = OwnedDataValue::from(&context);
1285 let mut rc = RunContext::new(&mut slot, Tuning::default());
1286 let mut overlays = Vec::new();
1287 let reader = rc.reader();
1288 for (i, (element, writes, seen, olds)) in elements.iter().enumerate() {
1289 let value = OwnedDataValue::from(element);
1290 let binding = Binding {
1291 item: &item,
1292 value: &value,
1293 index: &index,
1294 index_value: OwnedDataValue::from(i),
1295 };
1296 let mut ctx = Ctx::new(
1297 CallStart {
1298 reader,
1299 binding: Some(&binding),
1300 payload: &payload,
1301 exprs: &exprs,
1302 budget: &budget,
1303 info: info(),
1304 deadline: None,
1305 cancel: None,
1306 capture: true,
1307 },
1308 &NoCallee,
1309 );
1310 for (dotted, value) in writes {
1311 let value = value.as_ref().map(OwnedDataValue::from).unwrap();
1312 ctx.set(&parse(dotted), value).unwrap();
1313 }
1314 assert_eq!(get(&ctx, "temp_data.p").as_ref(), Some(seen), "element {i}");
1315 assert_eq!(
1316 eval(&ctx, json!({"var": "temp_data.p"})),
1317 *seen,
1318 "element {i}"
1319 );
1320 assert_eq!(get(&ctx, "temp_data.p_index"), Some(json!(i)));
1321 let (overlay, changes, _) = ctx.settle();
1322 let recorded: Vec<Value> = changes.iter().map(|c| Value::from(&c.old_value)).collect();
1323 assert_eq!(recorded, olds, "element {i}");
1324 overlays.push(overlay);
1325 }
1326 for overlay in overlays {
1327 overlay.apply(&mut rc);
1328 }
1329 rc.finish();
1330 assert_eq!(
1331 serde_json::to_string(&slot).unwrap(),
1332 r#"{"data":{"a":3,"b":2},"metadata":{},"temp_data":{"p":{"x":1,"y":2,"id":"C"}}}"#
1333 );
1334 }
1335
1336 #[test]
1337 fn render_lends_each_hole_from_where_it_is_and_builds_no_tree() {
1338 // Only a value the call wrote below a hole is converted.
1339 let context = json!({"data": {"order": {"customer": "C", "qty": 3}},
1340 "metadata": {}, "temp_data": {}});
1341 let exprs = ExprEngine::default();
1342 let budget = TaskBudget::unlimited();
1343 let payload = OwnedDataValue::Null;
1344 let mut slot = OwnedDataValue::from(&context);
1345 let mut rc = RunContext::new(&mut slot, Tuning::default());
1346 rc.read(|_, _| ());
1347 // The run writes one value with its owned copy, as `map` does under
1348 // capture, and one into the view only, as the progress write does.
1349 rc.set(
1350 &parse("data.order.since"),
1351 OwnedDataValue::from("2026"),
1352 None,
1353 );
1354 let progress = parse("metadata.progress");
1355 rc.edit(|e| {
1356 e.set(
1357 &progress,
1358 OwnedDataValue::from("p").to_arena(e.arena()),
1359 None,
1360 )
1361 });
1362 let mut ctx = Ctx::new(
1363 CallStart {
1364 reader: rc.reader(),
1365 binding: None,
1366 payload: &payload,
1367 exprs: &exprs,
1368 budget: &budget,
1369 info: info(),
1370 deadline: None,
1371 cancel: None,
1372 capture: true,
1373 },
1374 &NoCallee,
1375 );
1376 ctx.set(&parse("temp_data.mine"), OwnedDataValue::from(1))
1377 .unwrap();
1378 ctx.set(&parse("temp_data.box.k"), OwnedDataValue::from(2))
1379 .unwrap();
1380 let doc = exprs
1381 .interpolated(&json!([
1382 {"var": "temp_data.mine"},
1383 {"var": "temp_data.box"},
1384 {"var": "data.order.since"},
1385 {"var": "metadata.progress"},
1386 {"var": "data.order"},
1387 {"var": "data.order.customer"},
1388 {"var": ["data.absent", "d"]},
1389 {"var": ["payload.x", "p"]}
1390 ]))
1391 .unwrap();
1392 let rendered = ctx.render(&doc);
1393 let kinds: Vec<&str> = rendered
1394 .filled()
1395 .iter()
1396 .map(|filled| match filled {
1397 Filled::Owned(Cow::Borrowed(_)) => "lent",
1398 Filled::Owned(Cow::Owned(_)) => "converted",
1399 Filled::View(_) => "view",
1400 Filled::Default(_) => "default",
1401 })
1402 .collect();
1403 assert_eq!(
1404 kinds,
1405 [
1406 "lent", // the call's own write
1407 "converted", // written below by the call
1408 "lent", // the run's log, its owned copy
1409 "view", // the run's log, the view alone
1410 "view", // above a write in the run's log
1411 "lent", // the owned tree
1412 "default", // nothing there
1413 "default" // the payload, never in the context
1414 ]
1415 );
1416 assert_eq!(
1417 serde_json::to_value(&rendered).unwrap(),
1418 json!([1, {"k": 2}, "2026", "p",
1419 {"customer": "C", "qty": 3, "since": "2026"}, "C", "d", "p"])
1420 );
1421 }
1422
1423 /// A callee whose resolve puts a value where its walk reports a
1424 /// secret-reading expression, as a field resolving its secret some other
1425 /// way than through the resolver would.
1426 struct Leaky;
1427
1428 impl BoundCall for Leaky {
1429 fn call<'a>(
1430 &'a self,
1431 _ctx: &'a mut Ctx<'_>,
1432 ) -> crate::function::BoxFuture<'a, Result<crate::function::Outcome>> {
1433 Box::pin(std::future::ready(Ok(crate::function::Outcome::Ok)))
1434 }
1435
1436 fn visit(&self, v: &mut dyn ExprVisitor) {
1437 let at = |text: &str| text.parse::<AuthoredPath>().unwrap();
1438 v.expr(&at("a.b[1]"), &json!({"secret": "k"}), false);
1439 v.expr(
1440 &at("a.c"),
1441 &json!({"cat": ["x", {"secret": {"var": "data.k"}}]}),
1442 false,
1443 );
1444 v.expr(&at("kept"), &json!({"var": "data.k"}), false);
1445 // Nothing is there: nothing is written.
1446 v.expr(&at("absent.x"), &json!({"secret": "k"}), false);
1447 }
1448
1449 fn primary(&self) -> Primary<'_> {
1450 Primary::Undeclared
1451 }
1452
1453 fn resolve(&self, _r: &mut dyn Resolver) -> Result<OwnedDataValue> {
1454 Ok(OwnedDataValue::from(&json!({
1455 "a": {"b": [1, "leaked"], "c": "leaked too"},
1456 "kept": "v",
1457 })))
1458 }
1459 }
1460
1461 #[test]
1462 fn input_nulls_every_value_the_walk_reports_reading_a_secret() {
1463 let context = json!({"data": {"k": "v"}, "metadata": {}, "temp_data": {}});
1464 let exprs = ExprEngine::default();
1465 let budget = TaskBudget::unlimited();
1466 let payload = OwnedDataValue::Null;
1467 let mut slot = OwnedDataValue::from(&context);
1468 let mut rc = RunContext::new(&mut slot, Tuning::default());
1469 let ctx = Ctx::new(
1470 CallStart {
1471 reader: rc.reader(),
1472 binding: None,
1473 payload: &payload,
1474 exprs: &exprs,
1475 budget: &budget,
1476 info: info(),
1477 deadline: None,
1478 cancel: None,
1479 capture: true,
1480 },
1481 &Leaky,
1482 );
1483 assert_eq!(
1484 ctx.input().unwrap().to_json_string(),
1485 r#"{"a":{"b":[1,null],"c":null},"kept":"v"}"#
1486 );
1487 }
1488
1489 #[test]
1490 fn debug_shows_the_call_and_never_the_context() {
1491 let context = json!({"data": {"secret_looking": "value"}});
1492 with_ctx(&context, info(), None, |ctx| {
1493 let text = format!("{ctx:?}");
1494 assert!(text.contains("task_id: \"t\"") && text.contains("overlay_empty: true"));
1495 assert!(!text.contains("secret_looking"), "{text}");
1496 assert!(text.ends_with(".. }"), "{text}");
1497 });
1498 }
1499
1500 /// A callee with one primary output and a fixed field, resolving the
1501 /// output through the config's own `Output`, as a derived config does.
1502 struct Computed {
1503 output: Output,
1504 }
1505
1506 impl Computed {
1507 fn new(raw: Value) -> Self {
1508 let engine = ExprEngine::default();
1509 let mut issues = Vec::new();
1510 let mut cx = crate::expr::CompileCx::new(&engine, &mut issues);
1511 let output = <Output as crate::expr::ConfigField>::compile(&raw, &mut cx)
1512 .expect("the output compiles");
1513 drop(cx);
1514 assert!(issues.is_empty(), "{issues:?}");
1515 Self { output }
1516 }
1517 }
1518
1519 impl BoundCall for Computed {
1520 fn call<'a>(
1521 &'a self,
1522 _ctx: &'a mut Ctx<'_>,
1523 ) -> crate::function::BoxFuture<'a, Result<crate::function::Outcome>> {
1524 Box::pin(std::future::ready(Ok(crate::function::Outcome::Ok)))
1525 }
1526
1527 fn visit(&self, v: &mut dyn ExprVisitor) {
1528 v.expr(&"out".parse().unwrap(), self.output.path().raw(), true);
1529 }
1530
1531 fn primary(&self) -> Primary<'_> {
1532 Primary::Set(&self.output)
1533 }
1534
1535 fn resolve(&self, r: &mut dyn Resolver) -> Result<OwnedDataValue> {
1536 let mut entries = Vec::new();
1537 if let Some(value) = crate::expr::ConfigField::resolve(&self.output, r)? {
1538 entries.push(("out".to_owned(), value));
1539 }
1540 entries.push(("kept".to_owned(), OwnedDataValue::from(7)));
1541 Ok(OwnedDataValue::Object(entries))
1542 }
1543 }
1544
1545 #[test]
1546 fn an_input_can_skip_or_supply_the_primary_output() {
1547 let callee = Computed::new(json!({"var": "data.missing"}));
1548 let context = json!({"data": {"k": 1}, "metadata": {}, "temp_data": {}});
1549 with_call(&context, info(), &callee, |ctx| {
1550 // Strict by default: a computed `null` is not a path.
1551 assert!(ctx.input().is_err());
1552 assert_eq!(
1553 ctx.input_without_primary_output().unwrap().to_json_string(),
1554 r#"{"kept":7}"#
1555 );
1556 assert_eq!(
1557 ctx.input_with_primary_output(&parse("data.fallback"))
1558 .unwrap()
1559 .to_json_string(),
1560 r#"{"out":"data.fallback","kept":7}"#
1561 );
1562 });
1563 }
1564
1565 #[test]
1566 fn a_valid_computed_output_resolves_as_usual_and_can_be_supplied() {
1567 let callee = Computed::new(json!({"var": "data.where"}));
1568 let context = json!({"data": {"where": "data.answer"}, "metadata": {}, "temp_data": {}});
1569 with_call(&context, info(), &callee, |ctx| {
1570 assert_eq!(
1571 ctx.input().unwrap().to_json_string(),
1572 r#"{"out":"data.answer","kept":7}"#
1573 );
1574 assert_eq!(
1575 ctx.input_with_primary_output(&parse("data.other"))
1576 .unwrap()
1577 .to_json_string(),
1578 r#"{"out":"data.other","kept":7}"#
1579 );
1580 });
1581 }
1582
1583 #[test]
1584 fn supplying_a_primary_output_a_config_lacks_is_a_config_error() {
1585 let context = json!({"data": {}, "metadata": {}, "temp_data": {}});
1586 with_ctx(&context, info(), None, |ctx| {
1587 let error = ctx.input_with_primary_output(&parse("data.x")).unwrap_err();
1588 assert_eq!(error.kind(), &ErrorKind::Config);
1589 assert!(error.message().contains("has no primary output"), "{error}");
1590 // Nothing to leave out: the omit form is `input`.
1591 assert_eq!(
1592 ctx.input_without_primary_output().unwrap(),
1593 OwnedDataValue::Null
1594 );
1595 });
1596 }
1597}