feat(bench): per-service C2 decomposition + v8 B2 pricing (#444)

crates/ourios-bench/src/c2.rs

@@ -40,10 +40,41 @@
 //! - **Pass**: `ratio ≥ 0.5` on a ≥ 1 M-line corpus; corpora
 //!   below 1 M lines abstain (`pass = None`).
 
+use std::collections::BTreeMap;
+
+use ourios_core::otlp::any_value::Value as AnyValueKind;
 use ourios_core::record::MinedRecord;
 use ourios_miner::cluster::NO_TEMPLATE;
+use ourios_parquet::promoted::SERVICE_NAME_KEY;
+
+use crate::{C2Result, ConvergenceSample, PerServiceC2};
+
+/// Distinct-`service.name` cap for the per-service decomposition. Real
+/// OTLP corpora carry tens of services; the cap is a cardinality guard
+/// (mirroring §3.2's ethos) so a pathological corpus with millions of
+/// distinct service names can't balloon the `by_service` map. Overflow
+/// folds into a single `<other>` bucket.
+const MAX_SERVICES: usize = 1024;
+
+/// The `<other>` overflow bucket name (see [`MAX_SERVICES`]).
+const OTHER_SERVICES: &str = "<other>";
+
+/// The `service.name` used when a record carries no such resource
+/// attribute (never expected on OTLP corpora; possible on the
+/// plain-text form, where the whole decomposition is one bucket).
+const UNKNOWN_SERVICE: &str = "<unknown>";
 
-use crate::{C2Result, ConvergenceSample};
+/// Per-service tally for the C2 decomposition. Template creation is a
+/// global monotonic event attributed to the creating line's service,
+/// so this is O(services) memory — no per-service id set (the module's
+/// whole memory-safety argument would otherwise break on exactly the
+/// non-converging corpora C2 exists to flag).
+#[derive(Default)]
+struct PerService {
+    lines: u64,
+    created: u64,
+    created_at_1m: Option<u64>,
+}
 
 /// Curve-size cap: the cadence is chosen so a corpus of any
 /// size yields at most this many samples (§3.4.3).
@@ -68,6 +99,10 @@ pub(crate) struct C2Accumulator {
     max_template_id: u64,
     curve: Vec<ConvergenceSample>,
     processed: u64,
+    /// Per-`service.name` decomposition (diagnostic; see [`PerServiceC2`]).
+    by_service: BTreeMap<String, PerService>,
+    /// [`MAX_SERVICES`] hit — extra services folded into `<other>`.
+    services_truncated: bool,
 }
 
 impl C2Accumulator {
@@ -83,24 +118,30 @@ impl C2Accumulator {
             max_template_id: 0,
             curve: Vec::new(),
             processed: 0,
+            by_service: BTreeMap::new(),
+            services_truncated: false,
         }
     }
 
-    /// Observe one emitted record. Only `template_id` matters
-    /// to C2; the rest of the record is ignored.
+    /// Observe one emitted record: fold its `template_id` into the
+    /// whole-corpus curve and attribute any template creation to the
+    /// record's `service.name` for the per-service decomposition.
     pub(crate) fn record(&mut self, emitted: &MinedRecord) {
-        self.observe(emitted.template_id);
+        let created = self.observe(emitted.template_id);
+        self.attribute(service_name(emitted), created);
     }
 
     /// Core of [`Self::record`], split out so the colocated
     /// tests can drive the sampling + convergence math at
     /// scale (millions of synthetic ids) without constructing
-    /// `MinedRecord`s or running the miner.
-    fn observe(&mut self, template_id: u64) {
+    /// `MinedRecord`s or running the miner. Returns whether this id
+    /// created a new template (a monotonic-max advance).
+    fn observe(&mut self, template_id: u64) -> bool {
         // A non-`NO_TEMPLATE` id larger than any seen before is
         // a freshly-created template (monotonic allocation);
         // a smaller-or-equal id is a reuse already counted.
-        if template_id != NO_TEMPLATE && template_id > self.max_template_id {
+        let created = template_id != NO_TEMPLATE && template_id > self.max_template_id;
+        if created {
             self.max_template_id = template_id;
             self.template_count += 1;
         }
@@ -118,6 +159,52 @@ impl C2Accumulator {
                 template_count: self.template_count,
             });
         }
+        created
+    }
+
+    /// Attribute a line (and any template creation) to its service.
+    /// The 1 M-line snapshot is taken at exactly the millionth line of
+    /// *that* service — within one line of the whole-corpus rule's
+    /// nearest-sample, sufficient for a diagnostic.
+    fn attribute(&mut self, service: &str, created: bool) {
+        // Cardinality guard: a known service (or a new one below the
+        // cap) keeps its name; once the cap is hit, unseen services fold
+        // into one `<other>` bucket rather than grow unboundedly. The
+        // one presence lookup here is reused below, so the hot path is
+        // two map ops (this + `get_mut`), never three.
+        let present = self.by_service.contains_key(service);
+        let key = if present || self.by_service.len() < MAX_SERVICES {
+            service
+        } else {
+            self.services_truncated = true;
+            OTHER_SERVICES
+        };
+        // Insert (allocating the owned key) only on a first sighting;
+        // `entry(key.to_string())` would copy on every line, a per-record
+        // allocation across a multi-million-line corpus. A known service
+        // is already answered by `present`; only the `<other>` sentinel
+        // needs its own check, since the cap path can arrive with the
+        // bucket already created.
+        let needs_insert = if key == service {
+            !present
+        } else {
+            !self.by_service.contains_key(key)
+        };
+        if needs_insert {
+            self.by_service
+                .insert(key.to_string(), PerService::default());
+        }
+        let entry = self
+            .by_service
+            .get_mut(key)
+            .expect("bucket present after the insert above");
+        entry.lines += 1;
+        if created {
+            entry.created += 1;
+        }
+        if entry.lines == ONE_MILLION {
+            entry.created_at_1m = Some(entry.created);
+        }
     }
 
     /// Compute the §3.4.3 [`C2Result`] from the accumulated
@@ -150,6 +237,37 @@ impl C2Accumulator {
             (None, None, None)
         };
 
+        // Per-service decomposition, largest service first. Each
+        // service's gate follows §3.4.3 on its own line count; template
+        // creation is attributed to the minting service, so
+        // `templates_created` sums to `template_count_at_end`.
+        let mut by_service: Vec<PerServiceC2> = self
+            .by_service
+            .into_iter()
+            .map(|(service_name, s)| {
+                let (at_1m, ratio, pass) = if s.lines >= ONE_MILLION && s.created > 0 {
+                    let c = s.created_at_1m.unwrap_or(s.created);
+                    let ratio = (c as f64) / (s.created as f64);
+                    (Some(c), Some(ratio), Some(ratio >= 0.5))
+                } else {
+                    (None, None, None)
+                };
+                PerServiceC2 {
+                    service_name,
+                    lines: s.lines,
+                    templates_created: s.created,
+                    templates_created_at_1m_lines: at_1m,
+                    convergence_ratio: ratio,
+                    pass,
+                }
+            })
+            .collect();
+        by_service.sort_by(|a, b| {
+            b.lines
+                .cmp(&a.lines)
+                .then(a.service_name.cmp(&b.service_name))
+        });
+
         C2Result {
             sample_cadence: self.cadence,
             total_lines: self.total_lines,
@@ -159,10 +277,27 @@ impl C2Accumulator {
             convergence_curve: self.curve,
             pass,
             corpus_at_least_1m,
+            by_service,
+            services_truncated: self.services_truncated,
         }
     }
 }
 
+/// The record's `service.name` resource attribute, or a sentinel when
+/// absent. Borrowed — the caller copies into the map key only on a
+/// first sighting.
+fn service_name(emitted: &MinedRecord) -> &str {
+    emitted
+        .resource_attributes
+        .iter()
+        .find(|kv| kv.key == SERVICE_NAME_KEY)
+        .and_then(|kv| match kv.value.as_ref()?.value.as_ref()? {
+            AnyValueKind::StringValue(s) => Some(s.as_str()),
+            _ => None,
+        })
+        .unwrap_or(UNKNOWN_SERVICE)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -269,4 +404,124 @@ mod tests {
             "a non-converged corpus must fail the C2 gate",
         );
     }
+
+    /// A `MinedRecord` carrying just the two fields the per-service
+    /// decomposition reads: `template_id` and a `service.name` resource
+    /// attribute.
+    fn rec(template_id: u64, service: &str) -> MinedRecord {
+        use ourios_core::otlp::{AnyValue, KeyValue};
+        use ourios_core::record::BodyKind;
+        use ourios_core::tenant::TenantId;
+        MinedRecord {
+            tenant_id: TenantId::new("bench-tenant"),
+            template_id,
+            template_version: 0,
+            severity_number: 9,
+            severity_text: None,
+            scope_name: None,
+            scope_version: None,
+            scope_attributes: Vec::new(),
+            resource_schema_url: None,
+            scope_schema_url: None,
+            time_unix_nano: 0,
+            observed_time_unix_nano: None,
+            attributes: Vec::new(),
+            dropped_attributes_count: 0,
+            resource_attributes: vec![KeyValue {
+                key: SERVICE_NAME_KEY.to_string(),
+                value: Some(AnyValue {
+                    value: Some(AnyValueKind::StringValue(service.to_string())),
+                }),
+                ..Default::default()
+            }],
+            trace_id: None,
+            span_id: None,
+            flags: 0,
+            event_name: None,
+            body_kind: BodyKind::String,
+            params: Vec::new(),
+            separators: vec![String::new(), String::new()],
+            body: None,
+            confidence: 1.0,
+            lossy_flag: false,
+        }
+    }
+
+    /// The per-service decomposition attributes each template creation
+    /// to the service of the *creating* line, so per-service creations
+    /// partition the whole-corpus end count exactly. Two services share
+    /// the id space (ids interleave), which the max-id attribution must
+    /// handle without a per-service id set.
+    #[test]
+    fn per_service_creations_partition_the_whole() {
+        // "svc-a" mints ids 1,2; "svc-b" mints ids 3,4,5. Interleaved,
+        // with reuse — but first-appearances stay monotonic (1,2,3,4,5),
+        // the miner's id-allocation invariant the max-id attribution
+        // relies on (RFC 0001 §6.1: ids are handed out in creation
+        // order). A script that minted id 2 *after* id 3 would be
+        // physically impossible from the miner and would (correctly)
+        // not register as a creation.
+        let script = [
+            (1, "svc-a"), // a creates 1
+            (2, "svc-a"), // a creates 2
+            (3, "svc-b"), // b creates 3
+            (1, "svc-a"), // a reuse
+            (4, "svc-b"), // b creates 4
+            (5, "svc-b"), // b creates 5
+            (2, "svc-a"), // a reuse
+            (3, "svc-b"), // b reuse
+        ];
+        let mut acc = C2Accumulator::new(script.len() as u64);
+        for (id, svc) in script {
+            acc.record(&rec(id, svc));
+        }
+        let r = acc.finalize();
+        assert_eq!(r.template_count_at_end, 5, "5 distinct templates overall");
+        assert_eq!(r.by_service.len(), 2);
+        let a = r
+            .by_service
+            .iter()
+            .find(|s| s.service_name == "svc-a")
+            .unwrap();
+        let b = r
+            .by_service
+            .iter()
+            .find(|s| s.service_name == "svc-b")
+            .unwrap();
+        assert_eq!(a.templates_created, 2, "svc-a minted ids 1,2");
+        assert_eq!(b.templates_created, 3, "svc-b minted ids 3,4,5");
+        assert_eq!(
+            a.templates_created + b.templates_created,
+            r.template_count_at_end,
+            "per-service creations partition the whole",
+        );
+        // Both services are < 1 M lines → each abstains.
+        assert_eq!(a.pass, None);
+        assert_eq!(b.pass, None);
+        // Sorted largest-first: svc-a and svc-b both have 4 lines, tie
+        // broken by name → svc-a first.
+        assert_eq!(r.by_service[0].service_name, "svc-a");
+    }
+
+    /// A record with no `service.name` attribute lands in the
+    /// `<unknown>` bucket rather than being dropped.
+    #[test]
+    fn missing_service_name_falls_back_to_unknown() {
+        let mut acc = C2Accumulator::new(2);
+        acc.record(&rec(1, "svc")); // has service.name
+        let mut bare = rec(2, "svc");
+        bare.resource_attributes.clear();
+        acc.record(&bare);
+        let r = acc.finalize();
+        assert!(
+            r.by_service
+                .iter()
+                .any(|s| s.service_name == UNKNOWN_SERVICE)
+        );
+        assert_eq!(
+            r.by_service.iter().map(|s| s.lines).sum::<u64>(),
+            2,
+            "every line is attributed to some bucket",
+        );
+    }
 }

crates/ourios-bench/src/lib.rs

@@ -595,6 +595,38 @@ pub struct C2Result {
     pub convergence_curve: Vec<ConvergenceSample>,
     pub pass: Option<bool>,
     pub corpus_at_least_1m: bool,
+    /// Per-`service.name` convergence, largest service first.
+    /// **Diagnostic only** — the gate above is defined on the whole
+    /// corpus; this decomposition attributes it. On a multi-service
+    /// corpus (every OTel-Demo capture) a whole-corpus C2 conflates a
+    /// noisy broker with clean application services, so this surfaces
+    /// where non-convergence actually lives (v8 §9.12 / #444). A
+    /// plain-text corpus (no `service.name`) collapses to a single
+    /// `<unknown>` bucket rather than being empty; empty only when C2
+    /// did not run.
+    #[serde(default)]
+    pub by_service: Vec<PerServiceC2>,
+    /// The distinct-`service.name` cap (`MAX_SERVICES`) was hit and
+    /// further services were folded into an `<other>` bucket — a
+    /// cardinality guard, never expected on real corpora.
+    #[serde(default)]
+    pub services_truncated: bool,
+}
+
+/// One service's slice of the [`C2Result`] decomposition. Template
+/// creation is a globally-monotonic event attributed to the service of
+/// the creating line, so `templates_created` sums across services to
+/// the whole-corpus `template_count_at_end`. `pass`/`ratio` follow the
+/// §3.4.3 gate rule applied to this service's own line count (abstain
+/// below 1 M).
+#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
+pub struct PerServiceC2 {
+    pub service_name: String,
+    pub lines: u64,
+    pub templates_created: u64,
+    pub templates_created_at_1m_lines: Option<u64>,
+    pub convergence_ratio: Option<f64>,
+    pub pass: Option<bool>,
 }
 
 #[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]