functions: one work budget across every call of a lifted function

A function lifted over a computed array runs once per element, and only the
six lookups counted their repeated work against LIFTED_SCAN_BUDGET. Criteria
aggregates rescanned their ranges per criterion, CUMIPMT/CUMPRINC stepped
through up to 1e7 periods per element, and array-producing functions such as
MUNIT built a whole result per element before keeping its first cell.

The budget now covers all of it: COUNTIF/SUMIF/AVERAGEIF(S)/MAXIFS/MINIFS
charge their range cells per call, the period loops of every element are
added up before any runs, and array results charge their cells as they are
built. Past the budget the call is #NUM!, as lifted lookups already are.

crates/calc/src/eval/array.rs

@@ -135,41 +135,10 @@ pub(super) fn single(v: Value) -> Option<Value> {
     }
 }
 
-/// Text bytes that one `&` over arrays may produce in all. Each piece is capped on its own, but a
-/// compact formula can broadcast a long text across millions of cells
-/// (`SEQUENCE(4000,4000)&REPT("x",30000)` is 480 GB): past this the whole result is `#NUM!`, the
-/// error for a result too large to hold, before the rest is built.
-pub(super) const MAX_CONCAT_ARRAY_BYTES: usize = 512 * 1024 * 1024;
-
 pub fn binary_values(op: BinOp, x: &Value, y: &Value, sys: gridcraft_core::DateSystem) -> Value {
-    if op == BinOp::Concat && (matches!(x, Value::Array(_)) || matches!(y, Value::Array(_))) {
-        return concat_arrays(x, y, sys, MAX_CONCAT_ARRAY_BYTES);
-    }
     zip2(x, y, |a, b| binary_scalar(op, a, b, sys))
 }
 
-/// `x & y` with at least one array operand, giving `#NUM!` once the texts built so far pass `budget`
-/// bytes.
-fn concat_arrays(x: &Value, y: &Value, sys: gridcraft_core::DateSystem, budget: usize) -> Value {
-    let used = std::cell::Cell::new(0usize);
-    let over = std::cell::Cell::new(false);
-    let r = zip2(x, y, |a, b| {
-        if over.get() {
-            return Value::Empty;
-        }
-        let v = binary_scalar(BinOp::Concat, a, b, sys);
-        if let Value::Text(t) = &v {
-            used.set(used.get().saturating_add(t.len()));
-            if used.get() > budget {
-                over.set(true);
-                return Value::Empty;
-            }
-        }
-        v
-    });
-    if over.get() { Value::Error(CellError::Num) } else { r }
-}
-
 fn binary_scalar(op: BinOp, a: &Value, b: &Value, sys: gridcraft_core::DateSystem) -> Value {
     if let Value::Error(e) = a {
         return Value::Error(*e);
@@ -218,22 +187,7 @@ fn binary_scalar(op: BinOp, a: &Value, b: &Value, sys: gridcraft_core::DateSyste
 
 #[cfg(test)]
 mod tests {
-    use super::{Array, BinOp, CellError, MAX_CELLS, Value, broadcast_len, concat_arrays};
-
-    #[test]
-    fn concatenating_arrays_stops_at_the_byte_budget() {
-        let sys = gridcraft_core::DateSystem::default();
-        let col = Value::from(Array::column((0..100).map(|i| Value::number(i as f64)).collect()));
-        let long = Value::text("x".repeat(100));
-        // 100 cells of 101 or 102 bytes each: about 10 KB in all.
-        let r = concat_arrays(&col, &long, sys, 20_000);
-        assert!(matches!(&r, Value::Array(a) if a.data.len() == 100), "{r:?}");
-        assert_eq!(concat_arrays(&col, &long, sys, 5_000), Value::Error(CellError::Num));
-        assert_eq!(concat_arrays(&long, &col, sys, 5_000), Value::Error(CellError::Num));
-        // Through the operator, a small result is unaffected.
-        let r = super::binary_values(BinOp::Concat, &col, &Value::text("a"), sys);
-        assert!(matches!(&r, Value::Array(a) if a.data.first() == Some(&Value::text("0a"))), "{r:?}");
-    }
+    use super::{MAX_CELLS, broadcast_len};
 
     #[test]
     fn broadcast_len_caps_at_max_cells() {

crates/functions/src/lift.rs

@@ -68,26 +68,70 @@ const SCANS: &[&str] = &["VLOOKUP", "HLOOKUP", "LOOKUP", "MATCH", "XLOOKUP", "XM
 /// `XMATCH(SEQUENCE(1048576), -SEQUENCE(1048576))` runs for hours and hangs opening the file.
 /// Past the budget the call is `#NUM!`, as MMULT is past 2e9 multiply-adds. 5e8 keeps the worst
 /// case to a few seconds and still admits 5,000 needles over a 100,000-row table.
+///
+/// The same budget covers all the work of a lifted call added up across its elements: criteria
+/// aggregates rescanning their ranges, period loops, and array results built per element.
 const LIFTED_SCAN_BUDGET: usize = 500_000_000;
 
-/// The length of the longest vector a lifted call of `spec` searches per needle: 0 unless `spec`
-/// is one of the [`SCANS`] lookups and a non-lifted argument is an array.
+/// Criteria aggregates: every call tests each cell of its ranges against its criteria, so a call
+/// lifted over `n` criteria does `n` x (range cells) work, like the [`SCANS`] lookups.
+const CRITERIA: &[&str] = &["COUNTIF", "COUNTIFS", "SUMIF", "SUMIFS", "AVERAGEIF", "AVERAGEIFS", "MAXIFS", "MINIFS"];
+
+/// Cumulative TVM functions that step through every period from `start_period` (argument 3) to
+/// `end_period` (argument 4) on each call.
+const PERIOD_LOOPS: &[&str] = &["CUMIPMT", "CUMPRINC"];
+
+/// The work every lifted call of `spec` repeats over its non-lifted arguments: the longest vector
+/// a [`SCANS`] lookup searches, or every cell a [`CRITERIA`] aggregate tests. 0 for other
+/// functions.
 fn scan_len(spec: &FnSpec, args: &[Arg]) -> usize {
-    if !SCANS.contains(&spec.name) {
+    let scans = SCANS.contains(&spec.name);
+    if !scans && !CRITERIA.contains(&spec.name) {
         return 0;
     }
-    args.iter()
-        .enumerate()
-        .filter(|(i, _)| !spec.is_scalar_arg(*i))
-        .filter_map(|(_, a)| match &a.value {
-            Value::Array(arr) => Some(arr.rows.max(arr.cols)),
-            _ => None,
-        })
-        .max()
-        .unwrap_or(0)
+    let lens = args.iter().enumerate().filter(|(i, _)| !spec.is_scalar_arg(*i)).filter_map(|(_, a)| match &a.value {
+        Value::Array(arr) => Some(if scans { arr.rows.max(arr.cols) } else { arr.data.len() }),
+        _ => None,
+    });
+    if scans { lens.max().unwrap_or(0) } else { lens.fold(0, usize::saturating_add) }
+}
+
+/// The work of one lifted call that depends on its own (lifted) scalar arguments: the number of
+/// periods a [`PERIOD_LOOPS`] function steps through. 0 for other functions.
+fn call_cost(spec: &FnSpec, cell_args: &[Arg]) -> usize {
+    if !PERIOD_LOOPS.contains(&spec.name) {
+        return 0;
+    }
+    let period = |i: usize| match cell_args.get(i).map(|a| &a.value) {
+        Some(Value::Number(n)) if n.is_finite() => Some(n.trunc()),
+        _ => None,
+    };
+    match (period(3), period(4)) {
+        // The function itself refuses spans past 1e7 periods; anything larger costs the cap.
+        (Some(start), Some(end)) if end >= start => ((end - start) + 1.0).min(1e8) as usize,
+        _ => 0,
+    }
 }
 
 pub(crate) fn call_lifted(spec: &FnSpec, args: &[Arg], ctx: &mut dyn Ctx) -> Value {
+    call_lifted_within(spec, args, ctx, LIFTED_SCAN_BUDGET)
+}
+
+/// Sets the lifted arguments of `cell_args` to their elements at (`r`, `c`).
+fn element_args(spec: &FnSpec, args: &[Arg], r: usize, c: usize, cell_args: &mut [Arg]) {
+    for (i, a) in args.iter().enumerate() {
+        if spec.is_scalar_arg(i)
+            && let Value::Array(arr) = &a.value
+            && let Some(slot) = cell_args.get_mut(i)
+        {
+            let v = arr.get_broadcast(r, c);
+            slot.value = if let Value::Array(inner) = &v { inner.data.first().cloned().unwrap_or(Value::Empty) } else { v };
+        }
+    }
+}
+
+/// [`call_lifted`] with every call's work added up against `budget` (see [`LIFTED_SCAN_BUDGET`]).
+fn call_lifted_within(spec: &FnSpec, args: &[Arg], ctx: &mut dyn Ctx, budget: usize) -> Value {
     let rejected = if spec.value_args.is_empty() { None } else { reject_multi_cell_refs(spec, args) };
     let args = rejected.as_deref().unwrap_or(args);
     let mut rows = 0usize;
@@ -106,25 +150,42 @@ pub(crate) fn call_lifted(spec: &FnSpec, args: &[Arg], ctx: &mut dyn Ctx) -> Val
         return (spec.imp)(args, ctx);
     }
     let needles = rows.saturating_mul(cols);
-    if needles > MAX_CELLS || needles.saturating_mul(scan_len(spec, args)) > LIFTED_SCAN_BUDGET {
+    // Every call is charged its repeated scan (at least 1), its own period loop, and the cells of
+    // an array result (built in full, though only its first cell is kept), so lifting a scalar
+    // argument can't multiply what one call may do into hours of work:
+    // `COUNTIF(big_range, SEQUENCE(1e4))`, `CUMIPMT(0.1, 1e7, 1, 1, SEQUENCE(1e4,,1e7,0), 0)`,
+    // `MUNIT(SEQUENCE(1e4,,3000,0))`.
+    let per_call = scan_len(spec, args).max(1);
+    let mut cell_args: Vec<Arg> = args.to_vec();
+    let mut spent = needles.saturating_mul(per_call);
+    if needles > MAX_CELLS || spent > budget {
         return Value::Error(CellError::Num);
     }
-    let mut cell_args: Vec<Arg> = args.to_vec();
+    // Period loops are known before any call runs, so a call over budget does none of them.
+    if PERIOD_LOOPS.contains(&spec.name) {
+        for r in 0..rows {
+            for c in 0..cols {
+                element_args(spec, args, r, c, &mut cell_args);
+                spent = spent.saturating_add(call_cost(spec, &cell_args));
+                if spent > budget {
+                    return Value::Error(CellError::Num);
+                }
+            }
+        }
+    }
     let mut data = Vec::with_capacity(rows * cols);
     for r in 0..rows {
         for c in 0..cols {
-            for (i, a) in args.iter().enumerate() {
-                if spec.is_scalar_arg(i)
-                    && let Value::Array(arr) = &a.value
-                    && let Some(slot) = cell_args.get_mut(i)
-                {
-                    let v = arr.get_broadcast(r, c);
-                    slot.value = if let Value::Array(inner) = &v { inner.data.first().cloned().unwrap_or(Value::Empty) } else { v };
-                }
-            }
+            element_args(spec, args, r, c, &mut cell_args);
             let out = (spec.imp)(&cell_args, ctx);
             data.push(match out {
-                Value::Array(a) => a.data.first().cloned().unwrap_or(Value::Empty),
+                Value::Array(a) => {
+                    spent = spent.saturating_add(a.data.len());
+                    if spent > budget {
+                        return Value::Error(CellError::Num);
+                    }
+                    a.data.first().cloned().unwrap_or(Value::Empty)
+                }
                 v => v,
             });
         }
@@ -209,6 +270,52 @@ mod tests {
         assert_eq!(rows_of(&out), vec![vec![nv(-1.0)]; 1000]);
     }
 
+    /// Security: lifting a criteria aggregate rescans its range for every criterion, so criteria
+    /// x range cells past the budget is `#NUM!` before any scan runs.
+    #[test]
+    fn lifted_criteria_aggregates_stop_past_their_work_budget() {
+        use gridcraft_core::Array;
+        let tall = |n: usize| av(Array::filled(n, 1, nv(-1.0)).into());
+        is_err(ev("COUNTIF", vec![tall(1_000_000), tall(501)]), CellError::Num);
+        is_err(ev("SUMIF", vec![tall(1_000_000), tall(501)]), CellError::Num);
+        is_err(ev("COUNTIFS", vec![tall(1_000_000), tall(501)]), CellError::Num);
+        // Two ranges of 250,000 cells each count together.
+        is_err(ev("SUMIFS", vec![tall(250_000), tall(250_000), tall(1001)]), CellError::Num);
+        let counts = ev("COUNTIF", vec![av(col(&[1.0, 2.0, 2.0])), av(col(&[2.0, 1.0, 5.0]))]);
+        assert_eq!(rows_of(&counts), vec![vec![nv(2.0)], vec![nv(1.0)], vec![nv(0.0)]]);
+        for name in super::CRITERIA.iter().chain(super::PERIOD_LOOPS) {
+            assert!(crate::lookup(name).is_some(), "{name} isn't registered");
+        }
+    }
+
+    /// Security: CUMIPMT and CUMPRINC loop over their periods once per lifted element; the
+    /// periods of all elements are added up first, so an over-budget call runs none of them.
+    #[test]
+    fn lifted_period_loops_stop_past_their_work_budget() {
+        use gridcraft_core::Array;
+        let ends = |len: usize, end: f64| av(Array::filled(len, 1, nv(end)).into());
+        let args = |len: usize, end: f64| vec![n(0.01), n(1e7), n(1000.0), n(1.0), ends(len, end), n(0.0)];
+        // 51 elements of 1e7 periods each is past 5e8.
+        is_err(ev("CUMIPMT", args(51, 1e7)), CellError::Num);
+        is_err(ev("CUMPRINC", args(51, 1e7)), CellError::Num);
+        let out = ev("CUMIPMT", args(2, 12.0));
+        assert_eq!(rows_of(&out).len(), 2);
+        assert!(matches!(rows_of(&out).first().and_then(|r| r.first()), Some(Value::Number(x)) if *x < 0.0));
+    }
+
+    /// Security: an array-producing function lifted over a computed array builds each element's
+    /// whole result before keeping one cell; their cells count against the budget.
+    #[test]
+    fn lifted_array_results_count_against_the_budget() {
+        use gridcraft_core::Array;
+        let spec = crate::lookup("MUNIT").unwrap_or_else(|| panic!("MUNIT isn't registered"));
+        let dims = |n: usize| [av(Array::filled(n, 1, nv(10.0)).into())];
+        // Ten 10x10 identity matrices are 1,000 cells.
+        let out = super::call_lifted_within(spec, &dims(10), &mut TestCtx::default(), 1_010);
+        assert_eq!(rows_of(&out), vec![vec![nv(1.0)]; 10]);
+        is_err(super::call_lifted_within(spec, &dims(11), &mut TestCtx::default(), 1_010), CellError::Num);
+    }
+
     #[test]
     fn value_params_reject_multi_cell_references() {
         let s = FnSpec { value_args: &[true, false], ..spec() };