test(promql): add PR-derived gaps to the process E2E matrix

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

@@ -3928,6 +3928,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/asapquery_compatibility_process_e2e.rs

@@ -760,10 +760,46 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly()
         "true"
     );
 
+    // These complete expressions are NOT registered in this snapshot.
+    // This tests routing fallback, not absence of operator support: registered
+    // exact arithmetic is covered by shared_exact_dashboard_executes_selected_workload.
+    // Never partially warm an unregistered expression using a registered child.
+    let fallback_matrix = [
+        // ASAPQuery #700; backend #503.
+        "avg_over_time(asap_demo_gauge[5s])",
+        "count(asap_demo_gauge)",
+        "avg(asap_demo_gauge)",
+        // ASAPQuery #629/#700; backend #432.
+        "topk(5, asap_demo_gauge)",
+        // ASAPQuery #256/#572/#577/#644; Planner #343, backend #504.
+        "rate(asap_demo_counter_total[5s]) + rate(asap_demo_counter_total[5s])",
+        "rate(asap_demo_counter_total[5s]) / 2",
+        // ASAPQuery #466/#640; backend #473.
+        "sum_over_time(asap_demo_gauge[10s])",
+    ];
+    for query in fallback_matrix {
+        let response: Value = client
+            .get(format!("{backend}/api/v1/query"))
+            .query(&[
+                ("query", query.to_string()),
+                ("time", first_eval.to_string()),
+            ])
+            .send()
+            .await
+            .unwrap_or_else(|error| panic!("fallback request failed for {query}: {error}"))
+            .json()
+            .await
+            .unwrap_or_else(|error| panic!("fallback JSON failed for {query}: {error}"));
+        assert_eq!(
+            response["data"]["result"][0]["metric"]["fallback"], "true",
+            "unregistered matrix row must fall back atomically: {query}: {response}"
+        );
+    }
+
     let calls = fallback_calls.lock().await;
     assert_eq!(
         calls.len(),
-        2,
+        2 + fallback_matrix.len(),
         "planned queries unexpectedly fell back: {calls:?}"
     );
     assert_eq!(calls[0].0, "instant");