fix(revisions): keep Planner errors typed and reuse installed programs

Planner execution errors keep their type, so a memory limit during revision
admission or window building stays a resource error rather than a 400 input
rejection. Revisions use the installed raw programs instead of re-deriving
them per request, count buffered samples once however many windows share
them, and fail a non-empty window that yields no population. Heaps keep
their install-time update check while they use the kernel interpreter.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Author: zzylol

data_plane/src/drivers/ingest/prometheus_remote_write.rs

@@ -1736,7 +1736,7 @@ impl PrometheusRemoteWriteReceiver {
         now_ms: u64,
     ) -> Result<u64, crate::precompute_engine::revisions::RevisionError> {
         use crate::precompute_engine::revisions::{encode_state, InputSample, RevisionRecord};
-        use crate::precompute_engine::{raw_dag::RawDagProgram, window_manager::WindowManager};
+        use crate::precompute_engine::window_manager::WindowManager;
         use asap_types::sds::StoredOutputId;
         let (plan, store) = runtime.installed()?;
         let generation = plan
@@ -1825,15 +1825,21 @@ impl PrometheusRemoteWriteReceiver {
             }
             for config in plan.precompute_plan.materializations.iter().filter(|c| c.derived_input.is_none()) {
                 let filter = compile_spatial_filter(&config.spatial_filter_normalized)?;
-                let program = RawDagProgram::from_plan(&plan.precompute_plan, config)?;
+                let program = plan.installed_precompute_plan.raw_programs.get(&config.policy_fp_u64()).ok_or("revision raw program missing")?;
                 let mut matching = samples.iter().filter_map(|sample| {
                     let labels = sample.labels.iter().map(|(k,v)|(k.clone(),v.clone())).collect();
                     (sample.metric == config.metric && filter.matches(&labels)).then_some(())?;
                     Some((sample.series.as_str(), sample.timestamp_ms, sample.value?))
                 }).collect::<Vec<_>>();
                 matching.sort_by(|a, b| (a.0, a.1).cmp(&(b.0, b.1)));
-                // Reject the whole request when the Planner graph cannot admit it.
-                program.validate(matching, runtime.policy.max_checkpoint_bytes).map_err(|_| crate::precompute_engine::revisions::AdmissionRejected("sample is outside the bound summary kernel's input domain"))?;
+                // Reject the whole request when the Planner graph cannot admit its
+                // values; resource and setup failures are not input errors.
+                if let Err(error) = program.validate(matching, runtime.policy.max_checkpoint_bytes) {
+                    return Err(match error.downcast_ref::<asap_physical_operators::Error>() {
+                        Some(asap_physical_operators::Error::MemoryLimit) | None => error,
+                        Some(_) => crate::precompute_engine::revisions::AdmissionRejected("sample is outside the bound summary kernel's input domain").into(),
+                    });
+                }
             }
             Ok(())
         }, |inputs, revision, captured_at_ms, admitted, publish| {
@@ -1855,19 +1861,18 @@ impl PrometheusRemoteWriteReceiver {
                 let WorkerMessage::GroupSamples { sid, policy_fp, group_key, mut samples, .. } = message else { return Err("revision routing returned non-sample input".into()); };
                 let config = plan.precompute_plan.materializations.iter().find(|c| c.policy_fingerprint() == policy_fp).ok_or("revision raw configuration missing")?;
                 if committed.contains_key(&policy_fp.0) { continue; }
-                let program = RawDagProgram::from_plan(&plan.precompute_plan, config)?;
+                let program = plan.installed_precompute_plan.raw_programs.get(&config.policy_fp_u64()).ok_or("revision raw program missing")?;
                 let manager = WindowManager::with_layout(config.window_size, config.slide_interval, config.pane_origin_ms, &config.window_layout);
                 // Each stored window's admitted samples, built by the Planner graph below.
-                let mut windows: BTreeMap<(u64,u64), Vec<(String,i64,f64)>> = BTreeMap::new();
-                let mut buffered = 0usize;
+                // Overlapping windows share each sample's series key.
+                let mut windows: BTreeMap<(u64,u64), Vec<(Arc<str>,i64,f64)>> = BTreeMap::new();
                 samples.sort_by(|a,b| (&a.0,a.1).cmp(&(&b.0,b.1)));
                 for (series,time,value) in samples {
+                    let series: Arc<str> = Arc::from(series);
                     for start in manager.stored_bucket_starts(time) {
                         let (start,end) = manager.stored_bucket_bounds(start);
                         if start < 0 || end < 0 || end as u64 > captured_at_ms || (start as u64) < captured_at_ms.saturating_sub(retain.saturating_sub(max_window)) { continue; }
-                        buffered = buffered.saturating_add(series.len() + std::mem::size_of::<(String,i64,f64)>());
-                        if buffered > runtime.policy.max_checkpoint_bytes { return Err(Box::new(asap_physical_operators::Error::MemoryLimit)); }
-                        windows.entry((start as u64,end as u64)).or_default().push((series.clone(), time, value));
+                        windows.entry((start as u64,end as u64)).or_default().push((Arc::clone(&series), time, value));
                     }
                 }
                 // This captured revision contains every accepted local input.
@@ -1896,14 +1901,13 @@ impl PrometheusRemoteWriteReceiver {
                 for (window, samples) in windows {
                     let state = if samples.is_empty() {
                         // A known-empty counter pane of this captured revision.
-                        Some(program.empty_state()?)
+                        program.empty_state()?
                     } else {
                         let bounds = (i64::try_from(window.0)?, i64::try_from(window.1)?);
-                        program.build(samples.iter().map(|(s,t,v)|(s.as_str(),*t,*v)), bounds, runtime.policy.max_checkpoint_bytes)?
+                        program.build(samples.iter().map(|(s,t,v)|(s.as_ref(),*t,*v)), bounds, runtime.policy.max_checkpoint_bytes)?
+                            .ok_or("revision window admitted no population")?
                     };
-                    if let Some(state) = state {
-                        states.insert(window, Arc::from(state));
-                    }
+                    states.insert(window, Arc::from(state));
                 }
                 for ((start,end),state) in &states {
                     records.get_mut(&policy_fp.into()).unwrap().push(RevisionRecord { reference: reference.clone(), group: group.clone(), start_ms:*start, end_ms:*end, payload:encode_state(Arc::clone(state),program.family.clone())? });

data_plane/src/precompute_engine/raw_dag.rs

@@ -18,6 +18,10 @@ use planner_types::post_asap::{
 use planner_types::pre_asap::{ColumnRef, QueryExpr, Source};
 use std::collections::HashMap;
 
+/// Planner execution errors keep their type (e.g. `MemoryLimit`); setup and
+/// binding failures are messages.
+pub type BuildError = Box<dyn std::error::Error + Send + Sync>;
+
 /// A validated executable projection; semantics come from the installed node.
 /// The retained node ID makes failures attributable to the selected DAG.
 #[derive(Debug, Clone)]
@@ -197,6 +201,9 @@ impl RawDagProgram {
                     grouping: grouping.clone(),
                     reduction: reduction.clone(),
                 };
+                if program.heap() {
+                    program.validate_heap_update()?;
+                }
                 if let Some(old) = &selected {
                     if old.family != program.family
                         || old.input != program.input
@@ -233,7 +240,7 @@ impl RawDagProgram {
         samples: impl IntoIterator<Item = (&'a str, i64, f64)>,
         pane: (i64, i64),
         max_bytes: usize,
-    ) -> Result<Option<Box<dyn AggregateCore>>, String> {
+    ) -> Result<Option<Box<dyn AggregateCore>>, BuildError> {
         let samples = pane_batch(samples);
         if self.heap() {
             let mut updater = self.heap_updater()?;
@@ -247,11 +254,9 @@ impl RawDagProgram {
         match self.execute(samples, pane, max_bytes)?.as_slice() {
             [] => Ok(None),
             [row] => match row.as_slice() {
-                [_, _, Value::Summary { state, .. }] => {
-                    asap_summary_state::physical::from_physical(state.as_ref())
-                        .map(Some)
-                        .map_err(|e| e.to_string())
-                }
+                [_, _, Value::Summary { state, .. }] => Ok(Some(
+                    asap_summary_state::physical::from_physical(state.as_ref())?,
+                )),
                 _ => Err("raw precompute output is not a population state".into()),
             },
             _ => Err("one routed group produced several populations".into()),
@@ -263,7 +268,7 @@ impl RawDagProgram {
         &self,
         samples: impl IntoIterator<Item = (&'a str, i64, f64)>,
         max_bytes: usize,
-    ) -> Result<(), String> {
+    ) -> Result<(), BuildError> {
         let samples = pane_batch(samples);
         if self.heap() {
             let updater = self.heap_updater()?;
@@ -293,7 +298,7 @@ impl RawDagProgram {
         samples: impl IntoIterator<Item = (&'a str, i64, f64)>,
         pane: (i64, i64),
         max_bytes: usize,
-    ) -> Result<Vec<Vec<Value>>, String> {
+    ) -> Result<Vec<Vec<Value>>, BuildError> {
         use futures::StreamExt;
         let schema = precompute::raw_sample_schema();
         let rows = samples
@@ -322,7 +327,7 @@ impl RawDagProgram {
             },
         )
         .map_err(|e| e.to_string())?;
-        futures::executor::block_on(async {
+        Ok(futures::executor::block_on(async {
             let mut stream = graph.execute(program.roots(), context)?.pop().ok_or(
                 asap_physical_operators::Error::Invalid("missing output".into()),
             )?;
@@ -331,8 +336,31 @@ impl RawDagProgram {
                 rows.extend(batch?.rows().iter().cloned());
             }
             Ok::<_, asap_physical_operators::Error>(rows)
-        })
-        .map_err(|e| e.to_string())
+        })?)
+    }
+
+    /// Heaps still use the kernel interpreter, so install only updates it evaluates.
+    fn validate_heap_update(&self) -> Result<(), String> {
+        if !matches!(
+            &self.input.weight,
+            SummaryInputExpr::Column(ColumnRef::SampleValue) | SummaryInputExpr::Constant(_)
+        ) {
+            return Err("raw heap weight expression is unsupported".into());
+        }
+        fn item(expr: &SummaryInputExpr) -> bool {
+            match expr {
+                SummaryInputExpr::Column(ColumnRef::Named(_) | ColumnRef::SampleValue) => true,
+                SummaryInputExpr::Tuple(items) => items.iter().all(item),
+                SummaryInputExpr::EntityIdentity(
+                    planner_types::post_asap::EntityIdentity::PromqlLabelSet { excluding },
+                ) => excluding.is_empty(),
+                _ => false,
+            }
+        }
+        if !self.input.item.as_ref().is_some_and(item) {
+            return Err("raw heap item expression is unsupported".into());
+        }
+        self.heap_updater().map(|_| ())
     }
 
     fn heap_updater(&self) -> Result<Box<dyn AccumulatorUpdater>, String> {

data_plane/src/precompute_engine/revisions.rs

@@ -592,8 +592,12 @@ impl RevisionRuntime {
             .iter()
             .filter(|c| c.derived_input.is_none())
         {
-            let program = super::raw_dag::RawDagProgram::from_plan(&plan.precompute_plan, config)?;
-            let bytes = encode_state(Arc::from(program.empty_state()?), program.family)?;
+            let program = plan
+                .installed_precompute_plan
+                .raw_programs
+                .get(&config.policy_fp_u64())
+                .ok_or("revision raw program missing")?;
+            let bytes = encode_state(Arc::from(program.empty_state()?), program.family.clone())?;
             if bytes.len() > self.policy.max_checkpoint_bytes {
                 return Err(Box::new(asap_physical_operators::Error::MemoryLimit));
             }

data_plane/src/precompute_engine/worker.rs

@@ -1596,13 +1596,15 @@ fn build_pane(
     pane: (i64, i64),
 ) -> Result<Option<Box<dyn AggregateCore>>, String> {
     if let Some(program) = program {
-        return program.build(
-            samples
-                .iter()
-                .map(|(series, time, value)| (series.as_ref(), *time, *value)),
-            pane,
-            asap_physical_operators::runtime::Limits::default().max_bytes,
-        );
+        return program
+            .build(
+                samples
+                    .iter()
+                    .map(|(series, time, value)| (series.as_ref(), *time, *value)),
+                pane,
+                asap_physical_operators::runtime::Limits::default().max_bytes,
+            )
+            .map_err(|e| e.to_string());
     }
     #[cfg(test)]
     {
@@ -5032,7 +5034,8 @@ mod dag_execution_tests {
         }
     }
 
-    // Revision admission checks samples with the installed Planner graph.
+    // Revision admission checks samples with the installed Planner graph, and
+    // keeps a Planner memory limit distinguishable from invalid input.
     #[test]
     fn admission_validates_with_the_planner_graph() {
         let plan = plan("sum_over_time(asap_demo_gauge[5s])");
@@ -5045,9 +5048,21 @@ mod dag_execution_tests {
         assert!(program
             .validate([(series, 1000, 1.0), (series, 2000, 2.0)], 1 << 20)
             .is_ok());
-        assert!(program
+        let rejected = program
             .validate([(series, 1000, f64::INFINITY)], 1 << 20)
-            .is_err());
+            .unwrap_err();
+        assert!(!matches!(
+            rejected.downcast_ref::<asap_physical_operators::Error>(),
+            Some(asap_physical_operators::Error::MemoryLimit)
+        ));
+        // A resource limit stays typed, so admission does not report it as bad input.
+        let limited = program
+            .validate((0..64).map(|t| (series, t * 10, 1.0)), 1)
+            .unwrap_err();
+        assert!(matches!(
+            limited.downcast_ref::<asap_physical_operators::Error>(),
+            Some(asap_physical_operators::Error::MemoryLimit)
+        ));
     }
 
     // A raw output installs only with its Planner-compiled precompute graph.

data_plane/src/storage_engines/sketch_db/backfill/window_builder.rs

@@ -173,11 +173,13 @@ pub fn build_dag_accumulator(
         i64::try_from(window.0).map_err(|_| "backfill window start overflow")?,
         i64::try_from(window.1).map_err(|_| "backfill window end overflow")?,
     );
-    program.build(
-        samples
-            .iter()
-            .map(|sample| (sample.labels.as_str(), sample.timestamp_ms, sample.value)),
-        window,
-        asap_physical_operators::runtime::Limits::default().max_bytes,
-    )
+    program
+        .build(
+            samples
+                .iter()
+                .map(|sample| (sample.labels.as_str(), sample.timestamp_ms, sample.value)),
+            window,
+            asap_physical_operators::runtime::Limits::default().max_bytes,
+        )
+        .map_err(|e| e.to_string())
 }