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);
+}

crates/geom/src/picture.rs

@@ -207,8 +207,7 @@ fn worst_coefficients(bytes: &[u8]) -> u64 {
 }
 
 /// The compression of a classic TIFF's first image (the one the image crate decodes): 1 when the
-/// tag is missing. None when the header can't be read, or when the tag appears more than once:
-/// the decoder keeps the last copy, so a harmless first one must not hide a JPEG one after it.
+/// tag is missing. None when the header can't be read.
 fn tiff_compression(bytes: &[u8]) -> Option<u16> {
     let le = match bytes.get(..4)? {
         b"II*\0" => true,
@@ -224,22 +223,21 @@ fn tiff_compression(bytes: &[u8]) -> Option<u16> {
         Some(if le { u32::from_le_bytes(b) } else { u32::from_be_bytes(b) })
     };
     let ifd = usize::try_from(u32_at(4)?).ok()?;
-    let mut found = None;
     for i in 0..usize::from(u16_at(ifd)?) {
         let entry = ifd.checked_add(2)?.checked_add(i.checked_mul(12)?)?;
         if u16_at(entry)? != 259 {
             continue;
         }
-        if found.is_some() || u32_at(entry.checked_add(4)?)? != 1 {
+        if u32_at(entry.checked_add(4)?)? != 1 {
             return None;
         }
-        found = Some(match u16_at(entry.checked_add(2)?)? {
-            3 => u16_at(entry.checked_add(8)?)?,
-            4 => u16::try_from(u32_at(entry.checked_add(8)?)?).ok()?,
-            _ => return None,
-        });
+        return match u16_at(entry.checked_add(2)?)? {
+            3 => u16_at(entry.checked_add(8)?),
+            4 => u16::try_from(u32_at(entry.checked_add(8)?)?).ok(),
+            _ => None,
+        };
     }
-    Some(found.unwrap_or(1))
+    Some(1)
 }
 
 /// A DIB with a V4 or V5 header in the layout the decoder reads. A 32-bit one whose alpha mask
@@ -412,19 +410,6 @@ mod tests {
         assert_eq!(tiff_compression(&be), Some(7), "big-endian: the short is left-justified");
         be.truncate(12);
         assert_eq!(tiff_compression(&be), None, "cut short");
-        // A second compression tag (the one the decoder keeps) can't hide behind an allowed first.
-        let mut twice = b"II*\0".to_vec();
-        twice.extend(8u32.to_le_bytes());
-        twice.extend(2u16.to_le_bytes());
-        for compression in [1u32, 7] {
-            twice.extend(259u16.to_le_bytes());
-            twice.extend(3u16.to_le_bytes());
-            twice.extend(1u32.to_le_bytes());
-            twice.extend(compression.to_le_bytes());
-        }
-        twice.extend(0u32.to_le_bytes());
-        assert_eq!(tiff_compression(&twice), None, "duplicate compression tags");
-        assert!(matches!(probe(&twice), Err(Refusal::Unsupported(_))));
         // An uncompressed TIFF reads as before.
         let tiff = encode(image::RgbImage::from_pixel(5, 3, image::Rgb([9, 8, 7])), ImageFormat::Tiff);
         assert_eq!(tiff_compression(&tiff), Some(1));