fix(chartgen): name the backend when a drain slice's corpora disagree

examples/common/chartgen.rs

@@ -2825,44 +2825,74 @@ where
 /// points keep their own captions), what the window was, that no producer
 /// ran inside it, the corpus sizes, and any redeliveries the consumers saw.
 /// Empty when no in-slice row is a drain.
+///
+/// The corpus is stated **per backend as soon as the backends disagree**. A
+/// drain rate is a rate whatever the corpus size, but the corpus is what sets
+/// the window length, and a slice where one backend ran a deliberately
+/// smaller one is a documented deviation the reader has to be able to see. An
+/// unattributed set ("6M / 30k") names both sizes and tells nobody which bar
+/// is which, which is worse than either naming one or naming them all — so
+/// the flat sentence is kept only for the case it is actually true of, every
+/// backend on the same corpus.
 fn drain_notes<P>(runs: &[BackendRun], in_slice: P) -> Vec<String>
 where
     P: Fn(&ScenarioResult) -> bool,
 {
     let mut drained: Vec<&str> = Vec::new();
-    let mut corpora: BTreeSet<u64> = BTreeSet::new();
+    let mut per_backend: Vec<(&str, BTreeSet<u64>)> = Vec::new();
     let mut flows: BTreeSet<&str> = BTreeSet::new();
     let mut duplicates: Vec<String> = Vec::new();
     for run in runs {
         let mut dups = 0u64;
-        let mut any = false;
+        let mut corpora: BTreeSet<u64> = BTreeSet::new();
         for r in run.results.iter().filter(|r| in_slice(r)) {
             let Some(d) = &r.drain else {
                 continue;
             };
-            any = true;
             corpora.insert(d.corpus);
             flows.insert(canonical_flow(&r.flow));
             dups = dups.saturating_add(d.duplicates);
         }
-        if any {
+        if !corpora.is_empty() {
             drained.push(run.backend.as_str());
+            per_backend.push((run.backend.as_str(), corpora));
             if dups > 0 {
                 duplicates.push(format!("{} saw {dups} redeliveries", run.backend));
             }
         }
     }
     let mut notes = Vec::new();
     if !drained.is_empty() {
-        let corpora: Vec<String> = corpora.iter().map(|c| fmt_count(*c as f64)).collect();
+        // Every backend on one corpus reads as one number; the moment two
+        // disagree the sentence stops naming a size and a second line
+        // attributes them.
+        let uniform = per_backend
+            .iter()
+            .all(|(_, c)| *c == per_backend[0].1)
+            .then(|| fmt_counts(&per_backend[0].1));
+        let corpus_clause = match &uniform {
+            Some(sizes) => format!("a corpus of {sizes} messages"),
+            None => "its corpus".to_string(),
+        };
         notes.push(format!(
             "drain ({}): each {} point is the processing rate of the consumers from the moment \
-             every worker was assigned until 90 % of a corpus of {} messages, published before \
+             every worker was assigned until 90 % of {}, published before \
              they started, had been consumed; no producer ran in the window",
             drained.join(", "),
             flows.iter().copied().collect::<Vec<_>>().join(" / "),
-            corpora.join(" / "),
+            corpus_clause,
         ));
+        if uniform.is_none() {
+            notes.push(format!(
+                "corpus differs by backend: {} — a smaller corpus is a shorter window, not a \
+                 different measurement",
+                per_backend
+                    .iter()
+                    .map(|(b, c)| format!("{b} {}", fmt_counts(c)))
+                    .collect::<Vec<_>>()
+                    .join(", "),
+            ));
+        }
         if !duplicates.is_empty() {
             notes.push(format!(
                 "redeliveries counted once: {}",
@@ -2873,6 +2903,16 @@ where
     notes
 }
 
+/// Corpus sizes as they read in a caption: one backend's slice can span
+/// payload legs whose byte cap sized them differently, so this is a set.
+fn fmt_counts(corpora: &BTreeSet<u64>) -> String {
+    corpora
+        .iter()
+        .map(|c| fmt_count(*c as f64))
+        .collect::<Vec<_>>()
+        .join(" / ")
+}
+
 /// One caption line per backend whose in-slice rows arrived under an alias:
 /// the bar or line is real, and the reader must be told which name it was
 /// measured under.

tests/chartgen.rs

@@ -4350,9 +4350,14 @@ fn a_drain_document_publishes_drain_rows_and_withholds_rungs() {
         !svg.contains("100k"),
         "the sustained rung must not supply a point in a drain document"
     );
+    // The caption wraps at NOTE_WRAP, so the rule is asserted against the
+    // text nodes joined rather than the raw SVG: where the line happens to
+    // break is not part of the claim.
+    let caption = joined_text(&svg);
     assert!(
-        svg.contains("drain (inmemory, kafka)") && svg.contains("no producer ran in the window"),
-        "the caption must state the drain rule: {svg}"
+        caption.contains("drain (inmemory, kafka)")
+            && caption.contains("no producer ran in the window"),
+        "the caption must state the drain rule: {caption}"
     );
     assert!(
         svg.contains("consume_parallel point"),
@@ -4425,6 +4430,76 @@ fn the_drain_caption_covers_the_consume_bars_and_leaves_the_fifo_bar_its_own() {
     );
 }
 
+/// Every text node of a chart in reading order. A caption sentence wraps
+/// across nodes at [`chartgen::NOTE_WRAP`], so a phrase is asserted against
+/// this rather than the raw SVG, which would make the assertion a claim
+/// about where the line happened to break.
+fn joined_text(svg: &str) -> String {
+    texts(svg)
+        .into_iter()
+        .map(|(_, _, _, t)| t)
+        .collect::<Vec<_>>()
+        .join(" ")
+}
+
+/// Consumer-axis drain rows for the second backend, all at one rate so the
+/// run has a single corpus: `rate` picks whether it matches the in-process
+/// run's cell on that axis (10k msg/s over the fixture's 2 s window) or
+/// deviates from it.
+fn consumer_axis_drains(rate: f64) -> Vec<String> {
+    [1u32, 2, 4]
+        .into_iter()
+        .map(|consumers| drain_scenario("consume_parallel", 64, consumers, rate, 2.0, 0))
+        .collect()
+}
+
+#[test]
+fn one_shared_corpus_is_stated_once_and_never_attributed() {
+    // Both backends drained the same corpus — what every published chart
+    // renders today, since one matrix pins one corpus. The caption names it
+    // once; per-backend attribution here would be noise.
+    let run = kafka_ladder_run(&consumer_axis_drains(10_000.0));
+    let doc = parse(&document(&format!("{},{}", inmemory_run(true), run)));
+    let svg = chartgen::render_to_string(&doc, Family::ThroughputVsConsumers, Mode::Light)
+        .expect("chart should render");
+    let caption = joined_text(&svg);
+    assert!(
+        caption.contains("a corpus of 22k messages"),
+        "a shared corpus is stated in the drain sentence: {caption}"
+    );
+    assert!(
+        !caption.contains("corpus differs by backend"),
+        "backends that agree are not attributed: {caption}"
+    );
+}
+
+#[test]
+fn a_backend_that_drained_a_smaller_corpus_is_named_in_the_caption() {
+    // The deviation this exists for: a backend too slow to drain the pinned
+    // corpus in a sane wall clock runs a smaller one. A bare set
+    // ("22k / 2.2k") names both sizes and attributes neither, so the reader
+    // cannot tell which series had the shorter window — which is the whole
+    // difference between a documented deviation and a footnote nobody can
+    // apply.
+    let run = kafka_ladder_run(&consumer_axis_drains(1_000.0));
+    let doc = parse(&document(&format!("{},{}", inmemory_run(true), run)));
+    let svg = chartgen::render_to_string(&doc, Family::ThroughputVsConsumers, Mode::Light)
+        .expect("chart should render");
+    let caption = joined_text(&svg);
+    assert!(
+        caption.contains("corpus differs by backend: inmemory 22k, kafka 2.2k"),
+        "each backend's corpus is named: {caption}"
+    );
+    assert!(
+        !caption.contains("a corpus of"),
+        "the flat sentence must not claim one corpus the backends did not share: {caption}"
+    );
+    assert!(
+        caption.contains("no producer ran in the window"),
+        "the drain rule itself still stands: {caption}"
+    );
+}
+
 #[test]
 fn a_drain_account_the_numbers_refute_is_rejected() {
     let good = drain_scenario("consume_parallel", 64, 1, 60_000.0, 10.0, 0);