Skip to main content

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}