Verify shared producer updates using the actual Collector runtime

data_plane/src/precompute_engine/operators/sum_accumulator.rs

@@ -49,12 +49,15 @@ impl SumAccumulator {
     }
 
     pub fn deserialize_from_bytes(buffer: &[u8]) -> Result<Self, Box<dyn std::error::Error>> {
-        if buffer.len() < 4 {
-            return Err("Buffer too short for f32".into());
+        match buffer.len() {
+            // Legacy Python scalar sums carry no sample-count evidence.
+            4 => Ok(Self::with_sum(f32::from_le_bytes(buffer.try_into()?) as f64)),
+            // Counted sums use the same fixed layout as the Collector Sum payload.
+            16 => Self::from_sum_bytes(buffer),
+            len => {
+                Err(format!("Invalid persisted Sum payload length: {len} (want 4 or 16)").into())
+            }
         }
-        // Python uses struct.pack("<f", self.sum) which is 4-byte little-endian float
-        let sum = f32::from_le_bytes([buffer[0], buffer[1], buffer[2], buffer[3]]) as f64;
-        Ok(Self::with_sum(sum))
     }
 
     /// Decode the fixed Sum payload produced by the first-class Sum
@@ -96,8 +99,15 @@ impl SerializableToSink for SumAccumulator {
     }
 
     fn serialize_to_bytes(&self) -> Vec<u8> {
-        // Match Python's struct.pack("<f", self.sum) - 4-byte little-endian float
-        (self.sum as f32).to_le_bytes().to_vec()
+        match self.observation_count {
+            Some(count) => {
+                let mut bytes = Vec::with_capacity(16);
+                bytes.extend_from_slice(&self.sum.to_le_bytes());
+                bytes.extend_from_slice(&count.to_le_bytes());
+                bytes
+            }
+            None => (self.sum as f32).to_le_bytes().to_vec(),
+        }
     }
 }
 
@@ -272,6 +282,40 @@ mod tests {
         assert_eq!(merged.aux_stats().count, None);
     }
 
+    // Persistence retains known counts, including zero and the full u64 range.
+    #[test]
+    fn counted_sum_binary_round_trip() {
+        for count in [0, 3, u64::MAX] {
+            let acc = SumAccumulator {
+                sum: 1.0000000000001,
+                observation_count: Some(count),
+            };
+            let bytes = acc.serialize_to_bytes();
+            assert_eq!(bytes.len(), 16);
+            let restored = SumAccumulator::deserialize_from_bytes(&bytes).unwrap();
+            assert_eq!(restored.sum, acc.sum);
+            assert_eq!(restored.observation_count, Some(count));
+        }
+    }
+
+    // Existing scalar-only files remain readable without inventing counts.
+    #[test]
+    fn legacy_binary_sum_has_unknown_count() {
+        let bytes = 42.5f32.to_le_bytes();
+        let restored = SumAccumulator::deserialize_from_bytes(&bytes).unwrap();
+        assert_eq!(restored.sum, 42.5);
+        assert_eq!(restored.observation_count, None);
+        assert_eq!(restored.serialize_to_bytes(), bytes);
+    }
+
+    // Truncated counted payloads must not silently decode as scalar sums.
+    #[test]
+    fn persisted_sum_rejects_invalid_lengths() {
+        for len in [0, 3, 5, 8, 15, 17] {
+            assert!(SumAccumulator::deserialize_from_bytes(&vec![0; len]).is_err());
+        }
+    }
+
     #[test]
     fn test_sum_accumulator_creation() {
         let acc = SumAccumulator::new();

data_plane/src/storage_engines/sketch_db/index/mod.rs

@@ -3924,6 +3924,70 @@ mod tests {
         drop(p);
     }
 
+    // Raw sample counts must survive production durable storage.
+    #[test]
+    fn raw_count_survives_disk_eviction() {
+        use crate::storage_engines::types::AggregationType;
+        let tmp = tempfile::TempDir::new().unwrap();
+        let idx = Arc::new(SketchStore::new());
+        // Register an ExactAgg(Sum) sid keyed by `zone`.
+        let mut m = meta(8001);
+        m.metric_name = "http_requests_total".into();
+        m.group_by_keys = ["zone".to_string()].into_iter().collect();
+        m.agg_kind = AggKind::ExactAgg {
+            agg_type: AggregationType::Sum,
+            parameters_canonical: String::new(),
+            spatial_filter_canonical: String::new(),
+        };
+        idx.register(m);
+        let p = idx
+            .start_persistence(durable_cfg(tmp.path().to_path_buf()))
+            .unwrap();
+
+        let lv_zone = |v: &str| {
+            let mut x = BTreeMap::new();
+            x.insert("zone".to_string(), v.to_string());
+            x
+        };
+        for i in 0..10u64 {
+            let s = i * 30_000;
+            idx.append_precompute(
+                8001,
+                lv_zone("z0"),
+                (s, s + 30_000),
+                Box::new({
+                    let mut acc = crate::precompute_engine::operators::SumAccumulator::new();
+                    acc.update((i + 1) as f64);
+                    acc.update(10.0);
+                    acc
+                }),
+            );
+        }
+        assert!(
+            wait_until(
+                || idx.approx_memory_bytes() == 0 && idx.list_sealed_epochs_len() == 0,
+                std::time::Duration::from_secs(5),
+            ),
+            "exact-agg windows never fully evicted"
+        );
+        // Query the EVICTED portion [0, 150_000) — must come back from disk.
+        let series = idx.query_exact_agg_range(8001, 0, 150_000);
+        assert!(
+            !series.is_empty(),
+            "exact-agg query returned no result after flush and eviction"
+        );
+        let (_label, samples) = &series[0];
+        assert!(
+            samples.contains_key(&30_000),
+            "evicted exact-agg window missing from disk"
+        );
+        let stats = samples[&30_000].aux_stats();
+        assert_eq!(stats.count, Some(2));
+        assert_eq!(stats.sum, Some(11.0));
+        assert_eq!(stats.sum.unwrap() / stats.count.unwrap() as f64, 5.5);
+        drop(p);
+    }
+
     /// BUG #3: the memory diagnostic + the flusher's memory-pressure
     /// trigger must account for `current_epoch`, not just sealed epochs.
     /// On origin/main `approx_memory_bytes()` sums ONLY sealed epochs, so

data_plane/tests/backend_process_e2e.rs

@@ -16,7 +16,8 @@ use asap_otel_proto::tonic::metrics::v1::{
     metric::Data, DdSketch, DdSketchDataPoint, DdSketchEncoding, Metric, ResourceMetrics,
     ScopeMetrics,
 };
-use asap_sketchlib::proto::sketchlib::DdSketchState;
+use asap_precompute_rs::Precompute;
+use asap_sketchlib::proto::sketchlib::{sketch_envelope, SketchEnvelope as ProtoEnvelope};
 use control_plane::opamp::{
     opamp_proto, CollectorPlanStatus, CollectorPlanStatusKind, COLLECTOR_PLAN_CAPABILITY,
     COLLECTOR_PLAN_MESSAGE, PLAN_STATUS_MESSAGE,
@@ -75,10 +76,49 @@ fn ddsketch_export(
     plan: &serde_json::Value,
     sequence: u64,
 ) -> Vec<u8> {
-    let mut sketch = asap_sketchlib::DdSketch::new(alpha);
+    let decoded = asap_precompute_rs::CollectorPlan::from_json(
+        &serde_json::to_vec(plan).unwrap(),
+        "whole-e2e-collector",
+    )
+    .unwrap();
+    let mut configs = decoded.to_precompute_config_set().unwrap().configs;
+    assert_eq!(
+        configs.len(),
+        1,
+        "two query roots must create only one producer"
+    );
+    let config = configs.remove(0);
+    assert_eq!(config.sketch_params["relative_accuracy"], alpha);
+    let runtime = asap_precompute_rs::precompute::PrecomputeImpl::new(
+        Some(config),
+        Some(Box::new(move || {
+            Box::new(asap_precompute_rs::sketches::DDSketchWrapper::new(alpha))
+        })),
+        Some(Box::new(asap_precompute_rs::sketches::DDSketchObserver)),
+    );
     for value in values {
-        sketch.update(*value);
+        runtime
+            .observe(&asap_precompute_rs::Observation::new(
+                timestamp_ns / 1_000_000 - 500,
+                metric,
+                vec![],
+                vec![asap_precompute_rs::KeyValue::new("service", "whole-e2e")],
+                asap_precompute_rs::ObservationValue {
+                    kind: asap_precompute_rs::ObservationValueKind::Float,
+                    float: *value,
+                    ..Default::default()
+                },
+            ))
+            .unwrap();
     }
+    let envelopes = runtime.tick(timestamp_ns / 1_000_000);
+    assert_eq!(runtime.stats().input_observations, values.len() as u64);
+    assert_eq!(envelopes.len(), 1);
+    assert_eq!(envelopes[0].count, values.len() as u64);
+    let wire = ProtoEnvelope::decode(envelopes[0].payload.as_slice()).unwrap();
+    let Some(sketch_envelope::SketchState::Ddsketch(state)) = wire.sketch_state else {
+        panic!("expected actual Collector DDSketch state")
+    };
     let materialization = plan["materializations"][0]["materialization"]
         .as_u64()
         .unwrap();
@@ -126,12 +166,7 @@ fn ddsketch_export(
         attributes,
         start_time_unix_nano: timestamp_ns.saturating_sub(1_000_000_000),
         time_unix_nano: timestamp_ns,
-        sketch: DdSketchState {
-            alpha: sketch.wire_alpha(),
-            store_counts: sketch.store_counts,
-            store_offset: sketch.store_offset,
-        }
-        .encode_to_vec(),
+        sketch: state.encode_to_vec(),
         encoding: DdSketchEncoding::DdsketchEncodingProto as i32,
         exemplars: Vec::new(),
         flags: 0,
@@ -519,26 +554,6 @@ async fn production_control_plane_to_data_plane_otlp_to_promql() {
     let body = ingestion.text().await.unwrap();
     assert!(status.is_success(), "OTLP rejected: {status}: {body}");
 
-    // The sealed full frame already carries its exact window. A subsequent
-    // checkpoint exercises the same producer's next window independently.
-    let watermark_ns = sample_ns + window_ms * 1_000_000;
-    client
-        .post(format!("http://{otlp_http}/v1/metrics"))
-        .header("content-type", "application/x-protobuf")
-        .body(ddsketch_export(
-            "whole_process_e2e_latency_ms",
-            watermark_ns,
-            &[],
-            planned_alpha,
-            &collector_plan,
-            2,
-        ))
-        .send()
-        .await
-        .expect("POST watermark OTLP to production data plane")
-        .error_for_status()
-        .expect("data plane accepted watermark");
-
     let query = "quantile_over_time(0.99, whole_process_e2e_latency_ms[1s])";
     let mut last_response = serde_json::Value::Null;
     for _ in 0..50 {