dataflow-rs 4.0.1

//! One fan-out: a task's host function called once per
//! element of an array, then folded into the run in element order.
//!
//! `over` is evaluated once, to an owned array, by a loop's rule. Each
//! element's call is a host call ([`invoke`]) over the run's one view, with
//! the element and its index bound beneath the call's own writes
//! ([`Binding`]). Nothing reaches the run until every call has finished, so
//! no call sees another's writes, however many run at once. The calls run
//! under `FuturesUnordered`, at most `max_concurrency` at once and with no
//! spawn, so they borrow the view; no new call starts once a finished one
//! halts or fails the task, and the calls in flight finish.
//!
//! The fold then goes in element order over the elements that finished, as
//! v3's `run_for_each` folds: the records the call added, its writes (an
//! `Ok`'s, whatever its outcome), its result in `into`, then the recorder,
//! stamped with the element. It stops after the first element that fails
//! the task or halts, so the elements after it contribute nothing, finished
//! or not: what makes `max_concurrency` a timing knob only. Each element is
//! decided under [`Policy::per_element`], which cannot halt; the task's own
//! `terminal`, `on_rejection: halt` and `on_failure: halt` apply once, after
//! the fold.
//!
//! Each element's call checks the run's limits as it starts and, natively,
//! races them; one that returns `Err` once a limit has passed is cut short
//! too, as a single call is. Once one is cut short, the calls in flight are
//! dropped and nothing is folded, the finished elements included. The
//! records each call that returned added are kept, in element order and
//! stamped with their element, as a single call cut short keeps its own;
//! then the task is recorded once, with no element, as a call cut short is,
//! and the run stops.
//!
//! In a timed run each element's call is timed as it runs. The task's one
//! record of a fan-out cut short takes the time the call cut short ran, and
//! of an `over` with nothing to call the time from its evaluation to its
//! record, as v3's.

use super::budget::{DispatchKind, OccupancyGuard};
use super::host::{CallResult, Invoked, Layers, Settled, call_start, invoke, keep_records, settle};
use super::interp::{Over, RunEnv, elements};
use super::policy::{self, Decision, Flow, Policy, decide};
use super::record::{self, Completed};
use crate::base::{Change, Error, ErrorKind, ErrorSite, SiteKind};
use crate::expr::path::MAX_PATH_SEGMENTS;
use crate::expr::{MAX_ARRAY_INDEX, Path, Seg};
use crate::function::{Binding, BoundCall, CallInfo, CallStart, Outcome, Overlay};
use crate::program::{CallSite, CompiledWorkflow, FanSite};
use crate::state::RunContext;
use datavalue::OwnedDataValue;
use futures_util::stream::{FuturesUnordered, StreamExt};
use std::fmt::Write as _;
use std::sync::Arc;

/// What one element's call left once its `Ctx` was settled: all the fold
/// needs, owned, so no element holds the view.
struct Element {
    invoked: Invoked,
    /// The call's decision under the per-element policy.
    decision: Decision,
}

impl Element {
    /// An element's call, once it returned without being cut short, decided
    /// under `policy`.
    fn new(
        mut invoked: Invoked,
        policy: Policy,
        task_id: &str,
        run_limited: bool,
        task_site: &ErrorSite,
    ) -> Self {
        let decision = match &mut invoked.result {
            CallResult::Returned(Ok(outcome)) => decide(policy, task_id, Ok(*outcome)),
            CallResult::Returned(Err(e)) => {
                let mut err = std::mem::replace(e, Error::service("", ""));
                if err.site().is_none() {
                    err = err.with_site(task_site.clone());
                }
                if *err.kind() == ErrorKind::Budget && run_limited {
                    policy::forced_abort(task_id, err)
                } else {
                    decide(policy, task_id, Err(err))
                }
            }
            CallResult::CutShort(..) => unreachable!("a call cut short breaks out before Element"),
        };
        Self { invoked, decision }
    }

    /// Whether the element fails the task or halts the workflow, under the
    /// per-element policy: no new call starts after it, and the fold stops
    /// after it.
    fn stops(&self) -> bool {
        matches!(
            self.decision.flow,
            Flow::Fail(_) | Flow::Abort(_) | Flow::Halt
        )
    }
}

/// One element's call: the task's `start` with the element bound and its
/// index in the call's info, timed when `timed`.
#[allow(
    clippy::too_many_arguments,
    reason = "one call site's per-element facts, passed straight to `invoke`"
)]
async fn call_element<'r>(
    start: CallStart<'r>,
    fan: &'r FanSite,
    index: usize,
    item: &'r OwnedDataValue,
    bound: &'r dyn BoundCall,
    layers: Layers<'r>,
    timed: bool,
    timers: bool,
    _guard: OccupancyGuard<'r>,
) -> (usize, Invoked) {
    let binding = Binding {
        item: &fan.r#as,
        value: item,
        index: &fan.index,
        index_value: OwnedDataValue::from(index),
    };
    let start = CallStart {
        binding: Some(&binding),
        info: CallInfo {
            element_index: Some(index),
            ..start.info
        },
        ..start
    };
    let started = timed.then(record::now);
    let mut invoked = invoke(start, bound, layers, fan.collect.as_ref(), timers).await;
    invoked.elapsed = started.map(record::since);
    (index, invoked)
}

/// `items`, unless `into` has no slot for one of them: an index past
/// [`MAX_ARRAY_INDEX`] is a key, so its result's path would name no element
/// and the trace's replay would put it nowhere.
fn fits_into<'e>(
    fan: &FanSite,
    task: &str,
    items: std::borrow::Cow<'e, [OwnedDataValue]>,
) -> Result<std::borrow::Cow<'e, [OwnedDataValue]>, Error> {
    if fan.into.is_none() || items.len() <= MAX_ARRAY_INDEX + 1 {
        return Ok(items);
    }
    Err(Error::new(
        ErrorKind::Eval,
        format!(
            "{} has {} elements, and into holds at most {}, one result per index up to {MAX_ARRAY_INDEX}",
            Over::FanOut(task),
            items.len(),
            MAX_ARRAY_INDEX + 1
        ),
    ))
}

/// `into`'s segments and one more for an element's slot, which each
/// element's fold sets to its index: no path is built per element.
fn slot_segs(into: &Path) -> Vec<Seg> {
    debug_assert!(
        into.segs().len() < MAX_PATH_SEGMENTS,
        "the compiler leaves room below `into` for an index"
    );
    let mut segs = Vec::with_capacity(into.segs().len() + 1);
    segs.extend_from_slice(into.segs());
    segs.push(Seg::index(0));
    segs
}

/// One `FanOut` op: `over`, the calls, `into` pre-filled, the fold in
/// element order, then the task's own policy; the flow.
///
/// An `over` that is not an array is one `Err` of the task, decided under
/// the task's policy and recorded with no element, and so is one with more
/// elements than `into` has slots for. An empty one writes
/// `into = []` with one `Change`, whose old value is `null` as v3's, and
/// records the task once, with no element.
pub(crate) async fn fan_out<'p>(
    rc: &mut RunContext<'_, 'p>,
    env: &mut RunEnv<'_, 'p>,
    wf: &'p CompiledWorkflow,
    site: &'p CallSite,
    fan: &'p FanSite,
    bound: &'p dyn BoundCall,
) -> Flow {
    let task_id = env.ids(site.task).task;
    let meta = &wf.tasks[site.task as usize];
    let policy = site.policy;
    let over_site = env.task_site(
        SiteKind::ForEachOver,
        site.task,
        meta.path.join("for_each").join("over"),
    );
    let task_site = ErrorSite {
        kind: SiteKind::Task,
        authored_path: Some(meta.path.clone()),
        ..over_site.clone()
    };
    let over = elements(
        rc,
        env.exprs,
        env.budget,
        &fan.over,
        Over::FanOut(task_id),
        over_site,
    )
    .and_then(|items| fits_into(fan, task_id, items));
    let items = match over {
        Ok(items) if !items.is_empty() => items,
        // Nothing to call: an empty array completes the task, anything else
        // fails it.
        no_calls => {
            let errors_before = env.errors.len();
            let started = env.watch.timed.then(record::now);
            let result = no_calls.map(|_| {
                if let Some(into) = &fan.into {
                    rc.set(into, OwnedDataValue::Array(Vec::new()), None);
                    if env.capture.captures() {
                        env.changes.push(Change::write(
                            Arc::clone(into.dotted()),
                            OwnedDataValue::Null,
                            OwnedDataValue::Array(Vec::new()),
                        ));
                    }
                }
                Outcome::Ok
            });
            let decision = match result {
                Err(e) if *e.kind() == ErrorKind::Budget && env.budget.is_run_limited() => {
                    policy::forced_abort(task_id, e)
                }
                other => decide(policy, task_id, other),
            };
            let done = Completed::new(
                env,
                site.task,
                None,
                started.map(record::since),
                errors_before,
                false,
            );
            return record::record(rc, env, wf, done, decision);
        }
    };

    let n = items.len();
    let per_element = policy.per_element();
    let run_limited = env.budget.is_run_limited();
    let mut finished: Vec<Option<Element>> = (0..n).map(|_| None).collect();
    // The first call the run's limits cut short: its index, the records it
    // added, its error and how long it ran. The fan-out folds nothing then.
    let mut cut = None;
    {
        // The reader is taken in its own statement and every call holds a
        // copy: nothing writes to the run until the last call has finished.
        let reader = rc.reader();
        let (start, layers) = call_start(env, reader, wf, site);
        let mut pending = items.iter().enumerate();
        let mut in_flight = FuturesUnordered::new();
        let mut stop = false;
        loop {
            while !stop && in_flight.len() < fan.max_concurrency {
                let Some((index, item)) = pending.next() else {
                    break;
                };
                let guard = match env.budget.charge_dispatch(DispatchKind::ForEachElement {
                    workflow_id: &wf.id,
                    task_id: &meta.id,
                    index,
                }) {
                    Ok(guard) => guard,
                    Err(e) => {
                        let e = e.with_site(task_site.clone());
                        finished[index] = Some(Element {
                            invoked: Invoked {
                                result: CallResult::Returned(Err(Error::service("", ""))),
                                overlay: Overlay::default(),
                                changes: Vec::new(),
                                errors: Vec::new(),
                                collected: None,
                                elapsed: None,
                            },
                            decision: policy::forced_abort(task_id, e),
                        });
                        stop = true;
                        break;
                    }
                };
                in_flight.push(call_element(
                    start,
                    fan,
                    index,
                    item,
                    bound,
                    layers,
                    env.watch.timed,
                    env.limits.timers,
                    guard,
                ));
            }
            let Some((index, invoked)) = in_flight.next().await else {
                break;
            };
            if let CallResult::CutShort(stop, error) = invoked.result {
                // Dropping `in_flight` drops the calls still running.
                cut = Some((index, invoked.errors, stop, error, invoked.elapsed));
                break;
            }
            let element = Element::new(invoked, per_element, task_id, run_limited, &task_site);
            stop |= element.stops();
            finished[index] = Some(element);
        }
    }
    drop(items);
    if let Some((cut_index, cut_errors, stop, e, elapsed)) = cut {
        let errors_before = env.errors.len();
        let ids = env.ids(site.task);
        let mut cut_errors = Some(cut_errors);
        for (index, slot) in finished.into_iter().enumerate() {
            let errors = if index == cut_index {
                cut_errors.take()
            } else {
                slot.map(|element| element.invoked.errors)
            };
            if let Some(errors) = errors {
                keep_records(ids, env.now, env.now_text, env.errors, errors, Some(index));
            }
        }
        let decision = policy::cut_short(task_id, stop, &e);
        let done = Completed::new(env, site.task, None, elapsed, errors_before, false);
        return record::record(rc, env, wf, done, decision);
    }

    // One slot per element, so an element that failed or never ran is a
    // `null` at its own index. Written by the engine: no `Change` for the
    // audit, and in a trace that keeps changes the first change of the first
    // folded element's step.
    if let Some(into) = &fan.into {
        let prefill = OwnedDataValue::Array(vec![OwnedDataValue::Null; n]);
        if let Some(trace) = env.watch.trace.as_deref_mut().filter(|t| t.changes()) {
            trace.push_prelude(Change::write(
                Arc::clone(into.dotted()),
                rc.get_or_null(into),
                prefill.clone(),
            ));
        }
        rc.set(into, prefill, None);
    }
    let mut slot = fan.into.as_ref().map(|into| {
        let at = format!("{into}.");
        let base = at.len();
        (slot_segs(into), at, base)
    });
    let mut halted = false;
    let mut rejected = false;
    let mut failed = false;
    for (index, element) in finished.into_iter().enumerate() {
        let Some(Element { invoked, decision }) = element else {
            continue;
        };
        let Settled {
            result,
            collected,
            elapsed,
            errors_before,
            marked,
            ..
        } = settle(rc, env, site.task, invoked, Some(index));
        let ok = result.is_ok();
        let (rejected_here, failed_here) = (decision.rejected(), decision.failed());
        if let Some((segs, at, base)) = slot.as_mut() {
            if let Some(last) = segs.last_mut() {
                last.set_index(index);
            }
            let value = match collected {
                Some(value) if !rejected_here && !failed_here => value,
                _ => OwnedDataValue::Null,
            };
            // An `Err` records no change, its result's included, as v3's.
            if ok && env.capture.captures() {
                // The dotted path `into.index_child(index)` would have.
                at.truncate(*base);
                let _ = write!(at, "{index}");
                env.changes.push(Change::write(
                    at.as_str(),
                    OwnedDataValue::Null,
                    value.clone(),
                ));
            }
            rc.set_segs(segs, value, None);
        }
        rejected |= rejected_here;
        failed |= failed_here;
        let done = Completed::new(env, site.task, Some(index), elapsed, errors_before, marked);
        let flow = record::record(rc, env, wf, done, decision);
        rc.tick();
        match flow {
            flow @ (Flow::Fail(_) | Flow::Abort(_) | Flow::Stop(_)) => return flow,
            Flow::Halt => {
                halted = true;
                break;
            }
            Flow::Next | Flow::Jump(_) => {}
        }
    }
    policy.after_fan_out(halted, rejected, failed)
}