Bound the commentary-index parse, and stop proving it with a clock

The gate caught `commentary_index_lookup_is_not_quadratic` failing, then
passing on a re-run. Measured both ways: 1.62s alone, OVER 10s against
its own 10s deadline while the suite's other 3,347 tests ran
concurrently. A 6x margin against a shared CPU is not a margin, and a
test that gets re-run until it passes is not a test.

It was also measuring the wrong thing. Replacing the linear scan with a
HashSet bounded the LOOKUP; the set was still built from every entry the
disc declared. `playlists.xml` is attacker-controlled and has no length
of its own, so a hostile disc could still force an unbounded allocation
before any lookup happened — the parse, not the query, was the exposure.
The code's own comment said "unbounded parsed input" and only fixed half
of it.

`MAX_COM_INDICES` bounds both halves, at the one value that cannot
change behaviour: an index at or beyond `u16::MAX` can never match a
cell, because the labelling loops break at `u16::try_from(i + 1)`. Real
authoring is nowhere near it — the BD STN table admits 32 streams.

The parse moved into a `com_indices` helper so the bound is OBSERVABLE.
Through `labels_from_feature` it is not: a HashSet collapses repeated
values, and an out-of-range index changes no label, so the obvious test
passes with or without the cap — an assertion that cannot fail, which is
what the first draft of this fix shipped. The test now hands in 50,000
DISTINCT unaddressable indices and asserts they are refused. Proven red
with the bound removed (50,000 kept), green with it.

Applies to both `sub_com1_idx` and `aud_com1_idx`.

NOT done here: `labels/mod.rs:1421`'s `jar_inventory_dedup_is_not_quadratic`
is the same wall-clock shape and has the same flakiness. Named so it is
not lost.

src/labels/paramount.rs

@@ -98,6 +98,40 @@ enum ForcedSub {
     ForcedNarrative,
 }
 
+/// The most `*_com1_idx` entries worth parsing from one playlist.
+///
+/// These indices address CSV cell positions, and a cell is only addressable
+/// while its 1-based number fits a `u16` — the loops below `break` at
+/// `u16::try_from(i + 1)`. So an index at or beyond `u16::MAX` can never match
+/// a cell, and more than that many entries cannot describe anything new.
+///
+/// The bound is what makes the parse safe on hostile input, not merely fast.
+/// The `HashSet` that replaced a linear scan fixed the LOOKUP cost, but the
+/// set is still built from an attribute with no length limit: a disc
+/// declaring half a billion indices allocates half a billion entries before
+/// any lookup happens. Real authoring is nowhere near this — the BD STN table
+/// admits at most 32 streams per playlist — so nothing legitimate is lost.
+const MAX_COM_INDICES: usize = u16::MAX as usize;
+
+/// Parse a `*_com1_idx` attribute into the set the labelling loops query.
+///
+/// Extracted so the BOUND is observable. Asserting it through
+/// `labels_from_feature` is not possible: a `HashSet` collapses repeated
+/// values, and an out-of-range index changes no label either way, so such a
+/// test passes whether or not the cap exists — an assertion that cannot fail.
+/// Returning the set lets a test hand in tens of thousands of DISTINCT
+/// unaddressable indices and see them refused.
+fn com_indices(attr: Option<String>) -> HashSet<usize> {
+    attr.map(|s| {
+        s.split(',')
+            .take(MAX_COM_INDICES)
+            .filter_map(|i| i.trim().parse().ok())
+            .filter(|&i| i < MAX_COM_INDICES)
+            .collect()
+    })
+    .unwrap_or_default()
+}
+
 fn forced_sub_cell(cell: &str) -> ForcedSub {
     match cell.trim() {
         "1" => ForcedSub::ContainsForcedSegments,
@@ -123,9 +157,7 @@ fn labels_from_feature(feature: &str) -> Vec<StreamLabel> {
         // linearly once per stream, so `aud="..."` and `aud_com1_idx="..."`
         // both grown large make this quadratic in the size of one XML file.
         // Membership is the only operation performed on it.
-        let com_indices: HashSet<usize> = xml::attr(feature, "aud_com1_idx")
+        let com_indices = com_indices(xml::attr(feature, "aud_com1_idx"));
-            .map(|s| s.split(',').filter_map(|i| i.trim().parse().ok()).collect())
-            .unwrap_or_default();
 
         // The CSV *is* the STN list: one cell per stream, in stream order,
         // and `aud_com1_idx` is a 0-based index into those same cells. So
@@ -178,9 +210,7 @@ fn labels_from_feature(feature: &str) -> Vec<StreamLabel> {
 
         // HashSet for the same reason as the audio side above: unbounded
         // parsed input, membership-only use, linear scan once per stream.
-        let com_indices: HashSet<usize> = xml::attr(feature, "sub_com1_idx")
+        let com_indices = com_indices(xml::attr(feature, "sub_com1_idx"));
-            .map(|s| s.split(',').filter_map(|i| i.trim().parse().ok()).collect())
-            .unwrap_or_default();
 
         // As with audio: the cell position IS the STN slot. `forced_sub` and
         // `sub_com1_idx` are indexed against those same cells, so an empty
@@ -315,61 +345,91 @@ mod tests {
         );
     }
 
-    /// `sub_com1_idx` is an unbounded index list parsed straight out of the
+    /// `sub_com1_idx` is parsed straight out of the disc's `playlists.xml`,
-    /// disc's `playlists.xml` and was membership-tested with a linear
+    /// which is attacker-controlled and has no length bound of its own.
-    /// `Vec::contains` once per subtitle stream — quadratic in the size of a
-    /// single attacker-supplied file.
     ///
-    /// Proof is by deadline rather than micro-benchmark. With the linear scan
+    /// This replaces a WALL-CLOCK test. That one built a 200 000 x 1 000 001
-    /// the original fixture (200 000 streams x 1 000 001 indices) measured
+    /// fixture and failed if it took over 10 s, to prove the membership test
-    /// 31 s in a release build and far longer in debug; with a set it measured
+    /// was a set rather than a linear scan. Measured on the machine that
-    /// 0.03 s release / 0.56 s debug. A 10 s deadline sits ~18x above the
+    /// wrote this: 1.62 s alone, and OVER 10 s — a real failure — when the
-    /// slowest passing measurement and ~3x below the fastest failing one, and
+    /// suite's other 3 347 tests were running concurrently. A 6x margin
-    /// makes a regression fail fast instead of hanging CI.
+    /// against a shared CPU is not a margin; it is a CI failure that looks
+    /// like a flake and gets re-run until it passes.
     ///
-    /// The CSV now stops at the end of the 1-based `u16` stream-numbering
+    /// It also measured the wrong thing. Making the lookup O(1) bounded the
-    /// space, so only the first 65 535 cells are scanned. `INDICES` is raised
+    /// QUERY, not the PARSE: the set was still built from every entry the
-    /// to keep the linear-scan work product (`cells x indices`) at or above
+    /// disc declared, so a hostile playlist could still force an unbounded
-    /// the original fixture's, preserving that deadline margin.
+    /// allocation before any lookup happened. `MAX_COM_INDICES` bounds that,
-    ///
+    /// and this test asserts the bound directly — an equality check with no
-    /// Correctness is pinned on fixture-derived literals: indices 0, 2 and 4
+    /// clock in it, which cannot flake under any load.
-    /// are the commentary tracks, 1 and 3 are not.
     #[test]
-    fn commentary_index_lookup_is_not_quadratic() {
+    fn a_hostile_commentary_index_list_is_bounded_not_merely_fast() {
-        /// Cells offered. Everything past `u16::MAX` is unnumberable and the
+        // Three real indices, then far more entries than can address a cell.
-        /// parser stops there, so the scanned prefix is 65 535 cells.
+        const OVERSIZED: usize = MAX_COM_INDICES + 10_000;
-        const STREAMS: usize = 200_000;
-        const SCANNED: usize = u16::MAX as usize;
-        const INDICES: usize = 3_100_000;
-        let (tx, rx) = std::sync::mpsc::channel();
-        let worker = std::thread::spawn(move || {
         let mut feature = String::from(r#"<playlist name="Feature" sub=""#);
-            feature.push_str(&"eng,".repeat(STREAMS));
+        feature.push_str(&"eng,".repeat(8));
         feature.pop();
-            // Three real commentary indices, then a long run of one
-            // out-of-range value: nothing here is bounded by the stream count.
         feature.push_str(r#"" sub_com1_idx="0,2,4,"#);
-            feature.push_str(&"9999999,".repeat(INDICES));
+        feature.push_str(&"9999999,".repeat(OVERSIZED));
         feature.pop();
         feature.push_str(r#"" />"#);
-            let _ = tx.send(labels_from_feature(&feature));
+
-        });
+        let labels = labels_from_feature(&feature);
-        match rx.recv_timeout(std::time::Duration::from_secs(10)) {
+
-            Ok(labels) => {
+        // The fixture's real indices still decide the purposes: bounding the
-                worker.join().expect("worker panicked");
+        // parse must not change what a legitimate playlist means.
-                assert_eq!(labels.len(), SCANNED);
+        assert_eq!(labels.len(), 8);
         assert_eq!(labels[0].purpose, LabelPurpose::Commentary);
         assert_eq!(labels[1].purpose, LabelPurpose::Normal);
         assert_eq!(labels[2].purpose, LabelPurpose::Commentary);
         assert_eq!(labels[3].purpose, LabelPurpose::Normal);
         assert_eq!(labels[4].purpose, LabelPurpose::Commentary);
     }
-            Err(_) => panic!(
+
-                "labels_from_feature did not finish {STREAMS} streams x \
+    /// The set REFUSES unaddressable indices, so a hostile playlist cannot
-                 {INDICES} commentary indices within 10s — the membership \
+    /// inflate it. DISTINCT values on purpose: a `HashSet` collapses repeats,
-                 test is still a linear scan"
+    /// so a million copies of one index costs one entry and would prove
-            ),
+    /// nothing. Fifty thousand distinct out-of-range indices cost fifty
+    /// thousand entries without the filter, and none with it — so this test
+    /// goes red if the bound is removed, which the label-level assertions
+    /// below cannot do.
+    #[test]
+    fn distinct_unaddressable_indices_are_refused_not_stored() {
+        let hostile: String = (MAX_COM_INDICES..MAX_COM_INDICES + 50_000)
+            .map(|i| i.to_string())
+            .collect::<Vec<_>>()
+            .join(",");
+        let set = com_indices(Some(hostile));
+        assert!(
+            set.is_empty(),
+            "kept {} unaddressable indices — the parse is still unbounded",
+            set.len()
+        );
+        // The addressable ones are still kept.
+        assert_eq!(com_indices(Some("0,2,4".to_string())).len(), 3);
     }
+
+    /// An index that cannot address any cell is dropped rather than stored.
+    ///
+    /// `u16::MAX` and beyond can never match, because the labelling loop
+    /// stops at `u16::try_from(i + 1)`. Keeping such entries would let a disc
+    /// inflate the set with values that can never be looked up — the
+    /// allocation half of the same defect.
+    #[test]
+    fn an_index_that_cannot_address_a_cell_is_not_retained() {
+        let feature = format!(
+            r#"<playlist name="Feature" sub="eng,eng" sub_com1_idx="1,{},{}" />"#,
+            MAX_COM_INDICES,
+            MAX_COM_INDICES + 1
+        );
+        let labels = labels_from_feature(&feature);
+        assert_eq!(labels.len(), 2);
+        assert_eq!(labels[0].purpose, LabelPurpose::Normal);
+        assert_eq!(
+            labels[1].purpose,
+            LabelPurpose::Commentary,
+            "the addressable index must still be honoured"
+        );
     }
 
     /// Headroom: the BD STN_table admits at most 32 PG streams per playlist,