Mail merge: cap the cells a recipient list keeps, not just its records and fields

Every record is padded to every field and the list is kept twice (as read,
and filtered and sorted), so a short CSV with a 1,000-field header and
100,000 one-field rows held 2 x 100 million empty cells, several gigabytes.

A list now keeps at most 4 million cells: wide lists keep fewer records (with
a warning), narrow ones keep the usual 100,000.

crates/engine/src/cmd/mailings/recipients.rs

@@ -72,6 +72,25 @@ const MAX_RECORDS: usize = 100_000;
 
 const MAX_FIELDS: usize = 1_000;
 
+/// Most cells (records × fields) a recipient list keeps. Every record is padded to every field
+/// and the list is kept twice (as read, and filtered and sorted), so a short file with a wide
+/// header and many one-field rows would otherwise claim gigabytes of empty cells.
+const MAX_CELLS: usize = 4_000_000;
+
+/// Most records a list with `fields` fields keeps: [`MAX_RECORDS`], fewer for a wide list.
+fn max_records(fields: usize) -> usize {
+    MAX_RECORDS.min(MAX_CELLS / fields.max(1))
+}
+
+/// Cut `rows` to what a list with `fields` fields keeps, saying so when it drops any.
+fn cap_records<T>(rows: &mut Vec<T>, fields: usize) {
+    let keep = max_records(fields);
+    if rows.len() > keep {
+        log::warn!("the recipient list keeps at most {keep} records with {fields} fields; the rest are left out");
+        rows.truncate(keep);
+    }
+}
+
 pub(super) fn cell(x: &Value) -> String {
     match x {
         Value::Null => String::new(),
@@ -102,7 +121,7 @@ fn table_from_rows(fields: Option<&Vec<Value>>, arr: &[Value]) -> Result<(Vec<St
     };
     let rows = arr
         .iter()
-        .take(MAX_RECORDS)
+        .take(max_records(headers.len()))
         .filter_map(|r| match r {
             Value::Array(cells) => Some((0..headers.len()).map(|i| cells.get(i).map(cell).unwrap_or_default()).collect()),
             Value::Object(o) => Some(headers.iter().map(|h| o.get(h).map(cell).unwrap_or_default()).collect()),
@@ -223,7 +242,8 @@ fn read_file(path: &str) -> Result<Vec<u8>, CmdError> {
 
 /// Field names and records from a table's rows: with `headers`, the first row names the fields
 /// (blank names become `Column N`, repeats are numbered); otherwise every row is a record and
-/// the fields are `Column 1`…`Column N`. Records are padded or cut to the fields, and capped.
+/// the fields are `Column 1`…`Column N`. Records are padded or cut to the fields, and capped
+/// ([`max_records`]).
 fn header_split(mut all: Vec<Vec<String>>, headers: bool) -> Result<(Vec<String>, Vec<Vec<String>>), CmdError> {
     if all.is_empty() {
         return Err(CmdError::Failed("the recipient list is empty".into()));
@@ -235,7 +255,7 @@ fn header_split(mut all: Vec<Vec<String>>, headers: bool) -> Result<(Vec<String>
         let width = all.iter().map(Vec::len).max().unwrap_or(0).min(MAX_FIELDS);
         (1..=width).map(|i| format!("Column {i}")).collect()
     };
-    all.truncate(MAX_RECORDS);
+    cap_records(&mut all, fields.len());
     for r in &mut all {
         r.resize(fields.len(), String::new());
     }

crates/engine/src/cmd/mailings/tests.rs

@@ -695,3 +695,21 @@ fn merge_uses_only_filtered_records_in_sort_order() {
     assert!(s.run("mailings.finish", &json!({})).is_err());
     assert!(s.run("mailings.editRecipients", &json!({})).is_ok());
 }
+
+/// A wide header and many one-field rows: every record is padded to every field (and the list
+/// kept twice), so the cells the list keeps are capped, not just its records and fields.
+#[test]
+fn a_wide_sparse_recipient_list_is_capped_by_cells() {
+    let header = (1..=1_000).map(|i| format!("F{i}")).collect::<Vec<_>>().join(",");
+    let csv = format!("{header}\n{}", "x\n".repeat(20_000));
+    let mut s = Session::new(Document::from_text("Hello"));
+    let r = s.run("mailings.recipients", &json!({"csv": csv})).unwrap();
+    assert_eq!(r["records"], 4_000);
+    let rows: Vec<Value> = (0..20_000).map(|_| json!({})).collect();
+    let fields: Vec<String> = (1..=1_000).map(|i| format!("F{i}")).collect();
+    let r = s.run("mailings.recipients", &json!({"rows": rows, "fields": fields})).unwrap();
+    assert_eq!(r["records"], 4_000);
+    // A narrow list keeps its usual number of records.
+    let r = s.run("mailings.recipients", &json!({"csv": format!("Name\n{}", "Ann\n".repeat(20_000))})).unwrap();
+    assert_eq!(r["records"], 20_000);
+}