prepress: bound flattener, tiling, image, raster grid and ink-set work (#545)

* prepress: bound flattener, tiling, image, raster grid and ink-set work

- compose: a group with more than MAX_GROUP_SEGMENTS (200k) path segments
  is rasterized whole (ComplexReason::TooManySegments) before its
  arrangement, whose crossings grow with the square of the segment count.
- geom: transparent_groups returns at once when nothing is transparent;
  overlap_groups folds the open boxes into one group past
  MAX_OVERLAP_COMPARISONS (4M) instead of comparing every pair.
- pdf: item_count charges path elements and glyphs, so the tiling budget
  bounds the geometry a cell expansion copies; Channels::apply refuses to
  spread samples past MAX_IMAGE_SAMPLES (256 MiB).
- shade: pixel_size refuses grids past MAX_GRID_PIXELS after rounding, so
  a hairline-thin area cannot make the other side unbounded.
- raster: Grid::within_budget rechecks the grid after scaling and scales
  by the pixel ratio when the square-root pass is not enough.
- ink: InkSet keeps a name index, so adding many spots is linear.

* address review: count image report only after channels apply

* address review: validate finiteness before segment-limit fallback, loop raster grid budget reduction

crates/prepress/src/flatten/compose.rs

@@ -105,8 +105,14 @@ pub enum ComplexReason {
     Degenerate,
     /// More regions than [`FlattenOptions::max_regions`].
     TooManyRegions(usize),
+    /// More path segments than [`MAX_GROUP_SEGMENTS`], so the arrangement was not built.
+    TooManySegments(usize),
 }
 
+/// Path segments (shapes and clips together) a group may hold before it is rasterized whole
+/// without building its arrangement, whose crossings grow with the square of the segment count.
+pub const MAX_GROUP_SEGMENTS: usize = 200_000;
+
 #[derive(Clone, Debug, PartialEq, thiserror::Error)]
 #[non_exhaustive]
 pub enum FlattenError {
@@ -180,6 +186,20 @@ fn flatten_group(
         out.complex.push(ComplexRegion { path, items: members.to_vec(), at: out.items.len(), whole: Some(reason) });
         out.report.complex_groups.push(ComplexGroup { items: members.to_vec(), reason });
     };
+    let segments =
+        b.shapes.iter().map(|s| s.path.elements().len()).chain(b.clips.iter().map(|c| c.path.elements().len())).fold(0_usize, usize::saturating_add);
+    if segments > MAX_GROUP_SEGMENTS {
+        geom::check_finite(&b.shapes, &b.clips)?;
+        let mut path = BezPath::new();
+        for &i in members {
+            if !bounds[i].is_nan() {
+                path.extend(bounds[i].abs().path_elements(0.1));
+            }
+        }
+        whole(out, path, ComplexReason::TooManySegments(segments));
+        out.report.flattened += members.len();
+        return Ok(());
+    }
     let regions = match geom::regions_with(&b.shapes, &b.clips, opts.sliver) {
         Ok(r) => r,
         Err(GeomError::Degenerate) => {
@@ -1089,6 +1109,45 @@ mod tests {
         assert_eq!(at(&f, 65.0, 50.0, &[C, M, K]).1, [0.5, 0.5, 0.0]);
     }
 
+    #[test]
+    fn a_group_with_too_many_segments_goes_to_complex_unbuilt() {
+        let mut path = BezPath::new();
+        path.move_to(Point::new(0.0, 0.0));
+        for i in 0..=MAX_GROUP_SEGMENTS {
+            path.line_to(Point::new((i % 100) as f64, (i / 100) as f64));
+        }
+        path.close_path();
+        let crowded = Item::Fill(FillItem {
+            path,
+            rule: FillRule::NonZero,
+            paint: Paint::Solid(InkColor::Tints(vec![(C, 1.0)])),
+            overprint: Overprint::default(),
+        });
+        let f = flat(&list(InkSet::new(), vec![Item::Group(group(0.5, BlendMode::Normal, vec![crowded]))]));
+        assert_eq!(f.report.complex_groups.len(), 1);
+        assert!(matches!(f.report.complex_groups[0].reason, ComplexReason::TooManySegments(n) if n > MAX_GROUP_SEGMENTS), "{:?}", f.report);
+        assert_eq!(f.report.regions, 0, "the arrangement was not built");
+    }
+
+    #[test]
+    fn a_group_with_too_many_segments_still_rejects_non_finite_points() {
+        let mut path = BezPath::new();
+        path.move_to(Point::new(0.0, 0.0));
+        for i in 0..=MAX_GROUP_SEGMENTS {
+            path.line_to(Point::new((i % 100) as f64, (i / 100) as f64));
+        }
+        path.line_to(Point::new(f64::NAN, 0.0));
+        path.close_path();
+        let crowded = Item::Fill(FillItem {
+            path,
+            rule: FillRule::NonZero,
+            paint: Paint::Solid(InkColor::Tints(vec![(C, 1.0)])),
+            overprint: Overprint::default(),
+        });
+        let l = list(InkSet::new(), vec![Item::Group(group(0.5, BlendMode::Normal, vec![crowded]))]);
+        assert!(flatten(&l, &FlattenOptions::default()).is_err());
+    }
+
     #[test]
     fn kept_isolated_and_knockout_groups_come_back_plain() {
         // A placed page: an isolated group of opaque art, away from any transparency.

crates/prepress/src/flatten/geom.rs

@@ -113,8 +113,24 @@ pub fn regions_with(shapes: &[Shape], clips: &[ClipPath], sliver: f64) -> Result
     Ok(Arrangement::new(shapes, clips, sliver)?.regions())
 }
 
+/// The finiteness check [`regions_with`] makes: every point of every shape path and of every clip
+/// path a shape names. For callers that skip the arrangement but must still reject bad input.
+pub fn check_finite(shapes: &[Shape], clips: &[ClipPath]) -> Result<(), GeomError> {
+    let used = shapes.iter().flat_map(|s| &s.clips).filter_map(|&c| clips.get(c)).map(|c| &c.path);
+    eps_for(shapes.iter().map(|s| &s.path).chain(used)).map(|_| ())
+}
+
+/// Box comparisons [`overlap_groups`] makes before it folds the boxes still open into one group:
+/// many boxes sharing a horizontal span would otherwise cost one comparison per pair.
+const MAX_OVERLAP_COMPARISONS: usize = 4_000_000;
+
 /// Indices of `bounds` grouped by overlap, transitively, each group in index order. Touching boxes
 /// overlap; NaN boxes overlap nothing.
+///
+/// Past [`MAX_OVERLAP_COMPARISONS`] the boxes still open at the sweep line are folded into one
+/// group under their joint box. Those groups are larger than the overlaps require, which is safe
+/// (a group only says which items are cut into regions together) and bounded by
+/// [`FlattenOptions::max_regions`](super::FlattenOptions::max_regions).
 pub fn overlap_groups(bounds: &[Rect]) -> Vec<Vec<usize>> {
     fn find(p: &mut [usize], mut i: usize) -> usize {
         while p[i] != i {
@@ -128,8 +144,10 @@ pub fn overlap_groups(bounds: &[Rect]) -> Vec<Vec<usize>> {
     order.sort_by(|a, b| a.1.x0.total_cmp(&b.1.x0));
     // Sweep left to right, comparing each box with those still open at its left edge.
     let mut open: Vec<(usize, Rect)> = Vec::new();
+    let mut comparisons = 0_usize;
     for (i, b) in order {
         open.retain(|(_, o)| o.x1 >= b.x0);
+        comparisons += open.len();
         for (j, o) in &open {
             if o.y0 <= b.y1 && b.y0 <= o.y1 {
                 let (a, c) = (find(&mut parent, i), find(&mut parent, *j));
@@ -137,6 +155,15 @@ pub fn overlap_groups(bounds: &[Rect]) -> Vec<Vec<usize>> {
             }
         }
         open.push((i, b));
+        if comparisons > MAX_OVERLAP_COMPARISONS && open.len() > 1 {
+            let (first, joint) = open[0];
+            let joint = open[1..].iter().fold(joint, |j, (k, o)| {
+                let (a, c) = (find(&mut parent, first), find(&mut parent, *k));
+                parent[a.max(c)] = a.min(c);
+                j.union(*o)
+            });
+            open = vec![(first, joint)];
+        }
     }
     let mut groups: Vec<Vec<usize>> = Vec::new();
     let mut slot: Vec<Option<usize>> = vec![None; bounds.len()];
@@ -157,6 +184,9 @@ pub fn overlap_groups(bounds: &[Rect]) -> Vec<Vec<usize>> {
 /// other item paints as it is. Groups are transitive so that an opaque item keeps its place in
 /// paint order above or below everything it overlaps.
 pub fn transparent_groups(bounds: &[Rect], transparent: &[bool]) -> Vec<Vec<usize>> {
+    if !transparent.iter().any(|t| *t) {
+        return Vec::new();
+    }
     let mut groups = overlap_groups(bounds);
     groups.retain(|g| g.iter().any(|&i| transparent.get(i).copied().unwrap_or(false)));
     groups
@@ -986,6 +1016,14 @@ mod tests {
         assert!(transparent_groups(&bounds, &[false; 5]).is_empty());
     }
 
+    #[test]
+    fn stacked_boxes_fold_into_one_group_past_the_comparison_budget() {
+        // Boxes one above the other, all sharing x 0–10: each stays open for every later one.
+        let stack = |n: usize| -> Vec<Rect> { (0..n).map(|i| Rect::new(0.0, 2.0 * i as f64, 10.0, 2.0 * i as f64 + 1.0)).collect() };
+        assert_eq!(overlap_groups(&stack(2_000)).len(), 2_000, "2 million comparisons: compared one by one");
+        assert_eq!(overlap_groups(&stack(4_000)).len(), 1, "8 million comparisons: folded");
+    }
+
     /// About 1,000 rectangles on a letter page. Run with `--ignored --nocapture`.
     #[test]
     #[ignore]

crates/prepress/src/flatten/pdf.rs

@@ -667,7 +667,7 @@ impl<'s, 'a> Reader<'s, 'a> {
         let space = cs.ink_space(simulation.as_ref());
         let (inks, channels) = self.space_inks(&space, cs)?;
         let samples: Vec<u8> = raster.values.iter().map(|v| (v.clamp(0.0, 1.0) * 255.0).round() as u8).collect();
-        let samples = channels.apply(samples, space.len());
+        let samples = channels.apply(samples, space.len())?;
         let (w, h) = (raster.width as usize, raster.height as usize);
         let smask = (!raster.alpha.iter().all(|a| *a == 255)).then(|| raster.alpha.into());
         self.shared.report.rasterized_shadings += 1;
@@ -768,12 +768,15 @@ impl<'s, 'a> Reader<'s, 'a> {
             return;
         };
         let Some((inks, channels)) = self.space_inks(&im.space, &im.color_space) else { return };
+        let (w, h) = (im.width as usize, im.height as usize);
+        let Some(samples) = channels.apply(im.samples, im.space.len()) else {
+            self.shared.report.unreadable_images += 1;
+            return;
+        };
         self.shared.report.images += 1;
         if matches!(im.space, InkSpace::Cmyk { converted: true }) {
             self.shared.report.converted_images += 1;
         }
-        let (w, h) = (im.width as usize, im.height as usize);
-        let samples = channels.apply(im.samples, im.space.len());
         let smask = im.alpha.map(|a| {
             if (a.width as usize, a.height as usize) == (w, h) {
                 a.data
@@ -1072,13 +1075,23 @@ enum Channels {
     Spread(usize),
 }
 
+/// The most samples an image may hold once its values are spread over the page's inks
+/// ([`Channels::Spread`]): the pixel and ink limits each hold, but their product need not.
+const MAX_IMAGE_SAMPLES: usize = 256 << 20;
+
 impl Channels {
-    fn apply(&self, samples: Vec<u8>, width: usize) -> Vec<u8> {
-        match self {
+    /// `None` when spreading would pass [`MAX_IMAGE_SAMPLES`].
+    fn apply(&self, samples: Vec<u8>, width: usize) -> Option<Vec<u8>> {
+        Some(match self {
             Self::All => samples,
             Self::Keep(keep) => samples.chunks_exact(width.max(1)).flat_map(|s| s.iter().zip(keep).filter(|(_, k)| **k).map(|(v, _)| *v)).collect(),
-            Self::Spread(n) => samples.iter().flat_map(|v| std::iter::repeat_n(*v, *n)).collect(),
-        }
+            Self::Spread(n) => {
+                if samples.len().checked_mul(*n)? > MAX_IMAGE_SAMPLES {
+                    return None;
+                }
+                samples.iter().flat_map(|v| std::iter::repeat_n(*v, *n)).collect()
+            }
+        })
     }
 
     fn apply_f32(&self, values: &[f32]) -> Vec<f32> {
@@ -1125,12 +1138,15 @@ fn limit_scale(scale: f64, area: Rect, pixels: f64) -> f64 {
     if have > pixels { scale * (pixels / have).sqrt() } else { scale }
 }
 
-/// `items` and every item inside their groups and masks.
+/// `items` and every item inside their groups and masks, plus the path elements and glyphs each
+/// owns: what a copy of them allocates and a translation visits.
 fn item_count(items: &[Item]) -> usize {
     items.iter().fold(0, |sum, item| {
         let inner = match item {
             Item::Group(g) => item_count(&g.children).saturating_add(g.mask.as_ref().map_or(0, |m| item_count(&m.group.children))),
-            Item::Fill(_) | Item::Text(_) | Item::Image(_) => 0,
+            Item::Fill(f) => f.path.elements().len(),
+            Item::Text(t) => t.outlines.elements().len().saturating_add(t.glyphs.len()),
+            Item::Image(_) => 0,
         };
         sum.saturating_add(1).saturating_add(inner)
     })
@@ -1332,4 +1348,25 @@ mod tests {
         // A CMYK page keeps its resolution.
         assert_eq!(tiling_scales(dpi, page, cell, 1.0, 4).0, dpi);
     }
+
+    #[test]
+    fn spreading_samples_over_inks_stays_within_the_sample_budget() {
+        assert_eq!(Channels::Spread(3).apply(vec![1, 2], 1), Some(vec![1, 1, 1, 2, 2, 2]));
+        let pixels = MAX_IMAGE_SAMPLES / 1024;
+        assert!(Channels::Spread(1024).apply(vec![0; pixels], 1).is_some(), "at the budget");
+        assert_eq!(Channels::Spread(1025).apply(vec![0; pixels], 1), None, "past it");
+        assert_eq!(Channels::Spread(usize::MAX).apply(vec![0; 2], 1), None, "overflow");
+    }
+
+    #[test]
+    fn item_count_charges_path_elements() {
+        let path = Rect::new(0.0, 0.0, 1.0, 1.0).to_path(0.1);
+        let fill = Item::Fill(FillItem {
+            path: path.clone(),
+            rule: FillRule::NonZero,
+            paint: Paint::Solid(InkColor::Tints(vec![])),
+            overprint: Overprint::default(),
+        });
+        assert_eq!(item_count(&[fill]), 1 + path.elements().len());
+    }
 }

crates/prepress/src/flatten/raster.rs

@@ -104,13 +104,10 @@ impl<'a> Rasterizer<'a> {
         if !(bounds.is_finite() && page.is_finite() && self.dpi.is_finite() && self.dpi > 0.0) {
             return Ok(None);
         }
-        let mut scale = f64::from(self.dpi) / 72.0;
-        let mut grid = Grid::new(bounds, page, scale);
-        let pixels = grid.pixels();
-        if pixels > MAX_RASTER_PIXELS {
-            scale *= (MAX_RASTER_PIXELS as f64 / pixels as f64).sqrt() * 0.999;
-            grid = Grid::new(bounds, page, scale);
-            let dpi = (scale * 72.0) as f32;
+        let scale = f64::from(self.dpi) / 72.0;
+        let Some((grid, lowered)) = Grid::within_budget(bounds, page, scale) else { return Ok(None) };
+        if lowered {
+            let dpi = (grid.scale * 72.0) as f32;
             self.report.lowest_dpi = Some(self.report.lowest_dpi.map_or(dpi, |d| d.min(dpi)));
         }
         if grid.width() <= 0 || grid.height() <= 0 {
@@ -318,6 +315,29 @@ impl Grid {
         }
     }
 
+    /// The grid at `scale`, or at a lower one that keeps it within [`MAX_RASTER_PIXELS`]
+    /// (`true` then), checked after rounding. `None` when no scale does.
+    fn within_budget(bounds: Rect, page: Rect, mut scale: f64) -> Option<(Self, bool)> {
+        let mut grid = Self::new(bounds, page, scale);
+        if grid.pixels() <= MAX_RASTER_PIXELS {
+            return Some((grid, false));
+        }
+        // Scaling both sides down once is not enough for a thin region, whose short side stays
+        // at its minimum pixels: later passes scale by the pixel ratio itself. A grid whose
+        // coordinates hit the clamp reports fewer pixels than it spans, so a few passes may be
+        // needed; each shrinks the scale, and 8 bound the work.
+        for pass in 0..8 {
+            let linear = pass > 0;
+            let ratio = MAX_RASTER_PIXELS as f64 / grid.pixels() as f64;
+            scale *= if linear { ratio } else { ratio.sqrt() } * 0.999;
+            grid = Self::new(bounds, page, scale);
+            if grid.pixels() <= MAX_RASTER_PIXELS {
+                return Some((grid, true));
+            }
+        }
+        None
+    }
+
     fn width(&self) -> i64 {
         self.x1 - self.x0
     }
@@ -391,3 +411,28 @@ pub fn flatten_page(pdf: &[u8], page: usize, opts: &FlattenPageOptions) -> Resul
     let (_, report) = report.pages.pop().ok_or(FlattenPageError::Read(ReadError::Interpret))?;
     Ok((bytes, report))
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn a_thin_region_still_fits_the_pixel_budget() {
+        // 1e7 × 1e-3 points at 600 dpi: 8.3e7 × 1 pixels before rounding, 3 rows after it.
+        let page = Rect::new(0.0, 0.0, 1e7, 1e7);
+        let (grid, lowered) = Grid::within_budget(Rect::new(0.0, 0.0, 1e7, 1e-3), page, 600.0 / 72.0).unwrap();
+        assert!(lowered);
+        assert!(grid.pixels() <= MAX_RASTER_PIXELS, "{} × {}", grid.width(), grid.height());
+        // The full page at that resolution: scaled down once, both ways.
+        let (grid, _) = Grid::within_budget(page, page, 600.0 / 72.0).unwrap();
+        assert!(grid.pixels() <= MAX_RASTER_PIXELS && grid.pixels() > MAX_RASTER_PIXELS / 2, "{} × {}", grid.width(), grid.height());
+    }
+
+    #[test]
+    fn a_clamped_grid_still_fits_the_pixel_budget() {
+        // Both sides past the coordinate clamp: the pixel count saturates, so one pass is not enough.
+        let page = Rect::new(0.0, 0.0, 1e15, 1e15);
+        let (grid, lowered) = Grid::within_budget(page, page, 600.0 / 72.0).unwrap();
+        assert!(lowered && grid.pixels() <= MAX_RASTER_PIXELS, "{} × {}", grid.width(), grid.height());
+    }
+}

crates/prepress/src/flatten/shade.rs

@@ -173,10 +173,16 @@ fn set(comps: &mut [f32], covered: &mut [bool], n: usize, i: usize, c: &[f32]) {
     covered[i] = true;
 }
 
-/// The pixel grid covering `area` at `scale`: its size, or `None` when empty.
+/// The most pixels [`pixel_size`] allows: twice `MAX_RASTER_PIXELS`, as `limit_scale` bounds the
+/// area before rounding and rounding adds a row and a column at most. A thinner area rounds to
+/// one pixel across and would otherwise let the other side grow without bound.
+const MAX_GRID_PIXELS: f64 = 2.0 * 16_777_216.0;
+
+/// The pixel grid covering `area` at `scale`: its size, or `None` when empty or past
+/// [`MAX_GRID_PIXELS`].
 pub(super) fn pixel_size(area: Rect, scale: f64) -> Option<(usize, usize)> {
     let (w, h) = ((area.width() * scale).ceil(), (area.height() * scale).ceil());
-    (w.is_finite() && h.is_finite() && w >= 1.0 && h >= 1.0).then_some((w as usize, h as usize))
+    (w.is_finite() && h.is_finite() && w >= 1.0 && h >= 1.0 && w * h <= MAX_GRID_PIXELS).then_some((w as usize, h as usize))
 }
 
 /// Mesh triangles drawn into component buffers, later triangles over earlier ones.

crates/prepress/src/ink.rs

@@ -1,7 +1,7 @@
 //! The ink model: which inks a job prints with, and colors as tints of those inks.
 
 use std::borrow::Cow;
-use std::collections::{BTreeMap, BTreeSet};
+use std::collections::{BTreeMap, BTreeSet, HashMap};
 
 use craft_cms::math;
 use craft_pdf::interpret::color::CmykConversion;
@@ -114,6 +114,8 @@ impl InkId {
 #[derive(Clone, Debug, PartialEq)]
 pub struct InkSet {
     inks: Vec<Ink>,
+    /// Spot ids by normalized name, so adding each of many spots does not rescan the set.
+    by_name: HashMap<String, InkId>,
     aliases: BTreeMap<InkId, InkId>,
     to_process: BTreeSet<InkId>,
     all_to_process: bool,
@@ -130,6 +132,7 @@ impl InkSet {
     pub fn new() -> Self {
         Self {
             inks: Process::ALL.into_iter().map(Ink::Process).collect(),
+            by_name: HashMap::new(),
             aliases: BTreeMap::new(),
             to_process: BTreeSet::new(),
             all_to_process: false,
@@ -160,6 +163,7 @@ impl InkSet {
             return Some(id);
         }
         let id = InkId(u16::try_from(self.inks.len()).ok()?);
+        self.by_name.insert(spot.name.clone(), id);
         self.inks.push(Ink::Spot(spot));
         Some(id)
     }
@@ -174,8 +178,7 @@ impl InkSet {
         if let Some(p) = Process::from_name(name) {
             return Some(InkId::process(p));
         }
-        let at = self.inks.iter().skip(4).position(|ink| ink.name() == name)?;
-        u16::try_from(at + 4).ok().map(InkId)
+        self.by_name.get(name).copied()
     }
 
     pub fn get(&self, id: InkId) -> Option<&Ink> {
@@ -529,6 +532,15 @@ mod tests {
         assert_eq!(set.spots().count(), 1);
     }
 
+    #[test]
+    fn thousands_of_spots_are_added_and_found_without_rescanning() {
+        let mut set = InkSet::new();
+        let ids: Vec<InkId> = (0..60_000).map(|i| set.add(spot(&format!("Spot {i}"))).unwrap()).collect();
+        assert_eq!(set.add(spot("Spot 59999")), Some(ids[59_999]), "added once");
+        assert_eq!(set.lookup("Spot#2042"), Some(ids[42]));
+        assert_eq!(set.len(), 60_004);
+    }
+
     fn close(got: Option<[f32; 4]>, expected: [f32; 4]) -> bool {
         got.is_some_and(|got| got.iter().zip(expected).all(|(g, e)| (g - e).abs() < 1e-4))
     }