fix(planner): bug fixes to planner's optimizer, added a separte binary to print candidate sketch configurations

asap-planner-rs/Cargo.toml

@@ -15,6 +15,10 @@ path = "src/main.rs"
 name = "asap-optimizer-cli"
 path = "src/bin/optimizer_cli.rs"
 
+[[bin]]
+name = "candidate-gen-dump"
+path = "src/bin/candidate_gen_dump.rs"
+
 [dependencies]
 asap_types.workspace = true
 promql_utilities.workspace = true

asap-planner-rs/Dockerfile

@@ -21,6 +21,7 @@ RUN mkdir -p asap-query-engine/src && echo "fn main() {}" > asap-query-engine/sr
     mkdir -p asap-query-engine/benches && echo "fn main() {}" > asap-query-engine/benches/simple_store_bench.rs && \
     mkdir -p asap-planner-rs/src/bin && echo "fn main() {}" > asap-planner-rs/src/main.rs && \
     echo "fn main() {}" > asap-planner-rs/src/bin/optimizer_cli.rs && \
+    echo "fn main() {}" > asap-planner-rs/src/bin/candidate_gen_dump.rs && \
     echo "pub fn placeholder() {}" >> asap-planner-rs/src/lib.rs
 
 # Build dependencies (this layer will be cached)

asap-planner-rs/src/bin/candidate_gen_dump.rs

@@ -0,0 +1,179 @@
+//! Dump all candidate configs produced by enumerate_candidates for each AQE,
+//! before greedy selection. Useful for auditing what the candidate space looks like.
+
+use std::collections::HashMap;
+use std::path::PathBuf;
+
+use asap_planner::{
+    optimizer::{enumerate_candidates, extract_aqes, CandidateConfig, RQE},
+    ControllerConfig,
+};
+use asap_types::enums::WindowType;
+use clap::Parser;
+use serde_json::Value;
+
+#[derive(Parser)]
+#[command(
+    name = "candidate-gen-dump",
+    about = "Print all candidates from enumerate_candidates (pre-selection) for a workload config"
+)]
+struct Args {
+    #[arg(long = "input_config")]
+    input_config: PathBuf,
+
+    #[arg(long = "data-ingestion-interval-ms")]
+    scrape_interval_ms: u64,
+}
+
+fn main() -> anyhow::Result<()> {
+    let args = Args::parse();
+
+    let yaml_str = std::fs::read_to_string(&args.input_config)?;
+    let config: ControllerConfig = serde_yaml::from_str(&yaml_str)?;
+    let schema = config.schema_from_hints();
+
+    let rqes: Vec<RQE> = config
+        .query_groups
+        .iter()
+        .flat_map(|qg| {
+            qg.queries.iter().map(|q| RQE {
+                query_string: q.clone(),
+                t_repeat_ms: qg.repetition_delay_ms,
+            })
+        })
+        .collect();
+
+    let aqes = extract_aqes(&rqes, &schema, args.scrape_interval_ms);
+    println!("=== {} AQE(s) ===", aqes.len());
+
+    for (i, aqe) in aqes.iter().enumerate() {
+        println!(
+            "\n--- AQE #{i}: metric={} stat={:?} range={}ms min_t={}ms gcd_t={}ms freq={:.4}Hz ---",
+            aqe.requirements.metric,
+            aqe.requirements.statistics,
+            aqe.requirements.data_range_ms,
+            aqe.min_t_repeat_ms,
+            aqe.t_repeat_gcd_ms,
+            aqe.query_frequency_hz,
+        );
+        println!("  queries: {:?}", aqe.query_strings);
+
+        let candidates = enumerate_candidates(aqe, args.scrape_interval_ms);
+        print_candidates_grouped(&candidates);
+    }
+
+    Ok(())
+}
+
+/// Group the flat candidate list by (agg_type, sub_type) and print hierarchically:
+///   SketchType (sub_type)
+///     windows (N):
+///       ...
+///     params (M)  [× N windows = NM total]:
+///       ...
+/// EXACT is printed last as a single line.
+fn print_candidates_grouped(candidates: &[CandidateConfig]) {
+    // Collect unique (agg_type_str, sub_type) keys in first-seen order.
+    let mut group_order: Vec<(String, String)> = Vec::new();
+    // (agg_type_str, sub_type) -> (unique windows, unique params)
+    type WindowKey = (WindowType, u64, u64, u64); // (type, size, slide, n)
+    #[allow(clippy::type_complexity)]
+    let mut groups: HashMap<(String, String), (Vec<WindowKey>, Vec<Vec<(String, Value)>>)> =
+        HashMap::new();
+
+    let mut has_exact = false;
+
+    for c in candidates {
+        let Some(cfg) = &c.config else {
+            has_exact = true;
+            continue;
+        };
+
+        let key = (
+            format!("{:?}", cfg.aggregation_type),
+            cfg.aggregation_sub_type.clone(),
+        );
+
+        let entry = groups.entry(key.clone()).or_insert_with(|| {
+            group_order.push(key);
+            (Vec::new(), Vec::new())
+        });
+
+        let wk: WindowKey = (
+            cfg.window_type,
+            cfg.window_size_ms,
+            cfg.slide_interval_ms,
+            c.n_windows,
+        );
+        if !entry.0.contains(&wk) {
+            entry.0.push(wk);
+        }
+
+        let mut params: Vec<(String, Value)> = cfg
+            .parameters
+            .iter()
+            .map(|(k, v)| (k.clone(), v.clone()))
+            .collect();
+        params.sort_by(|(a, _), (b, _)| a.cmp(b));
+        if !entry.1.contains(&params) {
+            entry.1.push(params);
+        }
+    }
+
+    let total_sketch: usize = group_order
+        .iter()
+        .map(|k| {
+            let (ws, ps) = &groups[k];
+            ws.len() * ps.len()
+        })
+        .sum();
+    let total = total_sketch + if has_exact { 1 } else { 0 };
+    println!(
+        "  {} candidate(s) across {} sketch group(s):",
+        total,
+        group_order.len()
+    );
+
+    for key in &group_order {
+        let (windows, params) = &groups[key];
+        let (agg_type_str, sub_type) = key;
+        let sub = if sub_type.is_empty() {
+            String::new()
+        } else {
+            format!(" ({})", sub_type)
+        };
+        println!(
+            "\n    {}{} — {} window(s) × {} param(s) = {} candidate(s):",
+            agg_type_str,
+            sub,
+            windows.len(),
+            params.len(),
+            windows.len() * params.len(),
+        );
+
+        println!("      windows:");
+        for (wtype, size, slide, n) in windows {
+            let window_str = match wtype {
+                WindowType::Tumbling => format!("Tumbling {}ms", size),
+                WindowType::Sliding => format!("Sliding  {}ms / slide {}ms", size, slide),
+            };
+            println!(
+                "        {} n={} → {:?}",
+                window_str,
+                n,
+                // Re-derive method label from n and window type for display.
+                if *n == 1 { "Direct" } else { "Merge/Subtract" }
+            );
+        }
+
+        println!("      params:");
+        for p in params {
+            let kv: Vec<_> = p.iter().map(|(k, v)| format!("{k}: {v}")).collect();
+            println!("        {{{}}}", kv.join(", "));
+        }
+    }
+
+    if has_exact {
+        println!("\n    [EXACT]");
+    }
+}

asap-planner-rs/src/optimizer/aqe_extractor.rs

@@ -131,7 +131,7 @@ pub fn extract_aqes(
 
 /// Euclidean GCD. `num-integer` is not in the workspace; this two-liner is
 /// sufficient and avoids a dependency.
-fn gcd(a: u64, b: u64) -> u64 {
+pub(super) fn gcd(a: u64, b: u64) -> u64 {
     if b == 0 {
         a
     } else {

asap-planner-rs/src/optimizer/candidate_gen.rs

@@ -49,7 +49,7 @@ pub fn enumerate_candidates(aqe: &AQE, scrape_interval_ms: u64) -> Vec<Candidate
 
         for params in param_grid(agg_type, aqe.requirements.topk_count_events) {
             for (window_type, w, slide_interval, n) in
-                window_candidates(range_a_ms, aqe.min_t_repeat_ms, scrape_interval_ms)
+                window_candidates(range_a_ms, aqe.t_repeat_gcd_ms, scrape_interval_ms)
             {
                 let Some(qm) = determine_query_method(n, &props) else {
                     continue;
@@ -88,30 +88,39 @@ fn exact_candidate() -> CandidateConfig {
 
 /// Window candidates: (WindowType, W_ms, slide_interval_ms, n_windows).
 ///
-/// Tumbling: all W that divide range_a, are multiples of scrape_interval, and ≤ min_t_repeat.
+/// Tumbling: W must divide GCD(range_a, t_repeat_gcd_ms) and be a multiple of scrape_interval.
+///           Dividing the GCD ensures (a) n complete windows cover range_a exactly, and
+///           (b) window completions are harmonically aligned with every dashboard refresh cycle.
 ///           Slide interval = W (a tumbling window "slides" by its own width).
 /// Sliding:  W = range_a / k for each W that is a multiple of scrape_interval and divides
-///           range_a (k = range_a / W). At query time k staggered readings are merged or
-///           subtracted to cover range_a. Freshness is bounded by S (the slide interval),
-///           not by W — so the guard is S ≤ min_t_repeat_ms, not W ≤ min_t_repeat_ms.
-///           S doubles from scrape_interval up to min(W, min_t_repeat_ms).
+///           range_a (k = range_a / W). At query time k staggered readings spaced W apart
+///           are merged or subtracted to cover range_a.
+///           S must satisfy three constraints:
+///             (a) S | W   — so W-spaced snapshots land on slide boundaries (multi-window correctness)
+///             (b) S | t_repeat_gcd — so slide boundaries align with every dashboard refresh cycle
+///             (c) S < W   — S=W is excluded because slide==window is tumbling (duplicate candidate)
+///           S is enumerated as multiples of scrape_interval < W; the divisibility check on
+///           gcd(W, t_repeat_gcd) rejects values that fail (a) or (b) without a separate bound.
 fn window_candidates(
     range_a_ms: u64,
-    min_t_repeat_ms: u64,
+    t_repeat_gcd_ms: u64,
     scrape_interval_ms: u64,
 ) -> Vec<(WindowType, u64, u64, u64)> {
     let range_a = range_a_ms;
     if range_a == 0 || scrape_interval_ms == 0 {
         return vec![];
     }
 
-    let max_w = range_a.min(min_t_repeat_ms);
     let mut out = Vec::new();
 
-    // Tumbling: W divides range_a, is a multiple of scrape_interval, and ≤ max_w.
+    // Tumbling: W divides GCD(range_a, t_repeat_gcd) and is a multiple of scrape_interval.
+    // W | t_repeat_gcd ensures window completions align harmonically with all dashboards.
+    // W | range_a (implied since t_repeat_gcd | range_a is checked at generation time, but
+    // we verify explicitly via the gcd) ensures n windows cover range_a exactly.
+    let tumbling_divisor = super::aqe_extractor::gcd(range_a, t_repeat_gcd_ms);
     let mut w = scrape_interval_ms;
-    while w <= max_w {
-        if range_a.is_multiple_of(w) {
+    while w <= tumbling_divisor {
+        if tumbling_divisor.is_multiple_of(w) {
             let n = range_a / w;
             out.push((WindowType::Tumbling, w, w, n));
         }
@@ -124,10 +133,15 @@ fn window_candidates(
     while w <= range_a {
         if range_a.is_multiple_of(w) {
             let k = range_a / w;
+            // Valid S: S | gcd(W, t_repeat_gcd). Iterate up to W (exclusive); the
+            // divisibility check rejects anything above gcd automatically.
+            let slide_divisor = super::aqe_extractor::gcd(w, t_repeat_gcd_ms);
             let mut s = scrape_interval_ms;
-            while s <= w.min(min_t_repeat_ms) {
-                out.push((WindowType::Sliding, w, s, k));
-                s *= 2;
+            while s < w {
+                if slide_divisor.is_multiple_of(s) {
+                    out.push((WindowType::Sliding, w, s, k));
+                }
+                s += scrape_interval_ms;
             }
         }
         w += scrape_interval_ms;
@@ -398,10 +412,9 @@ mod tests {
     }
 
     #[test]
-    fn sliding_freshness_guard_uses_slide_not_width() {
-        // range_a=600_000ms > min_t_repeat=30_000ms. The old guard (range_a <= min_t_repeat)
-        // incorrectly rejected all Sliding candidates here. New guard: S ≤ min_t_repeat.
-        // W=600_000, S=30_000 ≤ min_t_repeat → full-width Direct Sliding should be present.
+    fn sliding_full_width_direct_generated_when_range_exceeds_t_repeat() {
+        // range_a=600_000ms > t_repeat_gcd=30_000ms. W=range_a is valid for sliding since
+        // freshness is governed by S (not W). S=30_000 | gcd(600_000, 30_000)=30_000 → emitted.
         let aqe = make_aqe(Statistic::Sum, 600_000, 30_000);
         let candidates = enumerate_candidates(&aqe, 30_000);
         assert!(
@@ -415,6 +428,75 @@ mod tests {
         );
     }
 
+    #[test]
+    fn sliding_slide_must_divide_gcd_of_window_and_t_repeat() {
+        // range_a=20_000, t_repeat_gcd=5_000, scrape=1_000.
+        // W=10_000 (k=2): slide_divisor = gcd(10_000, 5_000) = 5_000.
+        // Valid S: divisors of 5_000 that are multiples of 1_000 and < 10_000 → {1_000, 5_000}.
+        // Invalid: S=2_000 (5_000 % 2_000 ≠ 0), S=4_000 (5_000 % 4_000 ≠ 0).
+        let aqe = make_aqe(Statistic::Sum, 20_000, 5_000);
+        let candidates = enumerate_candidates(&aqe, 1_000);
+
+        let sliding_w10: Vec<_> = candidates
+            .iter()
+            .filter(|c| {
+                c.config.as_ref().is_some_and(|cfg| {
+                    cfg.window_type == WindowType::Sliding && cfg.window_size_ms == 10_000
+                })
+            })
+            .collect();
+
+        let slides: Vec<u64> = sliding_w10
+            .iter()
+            .map(|c| c.config.as_ref().unwrap().slide_interval_ms)
+            .collect();
+
+        assert!(
+            slides.contains(&1_000),
+            "S=1_000 should be valid (divides 5_000)"
+        );
+        assert!(
+            slides.contains(&5_000),
+            "S=5_000 should be valid (divides 5_000)"
+        );
+        assert!(
+            !slides.contains(&2_000),
+            "S=2_000 should be rejected (5_000 % 2_000 ≠ 0)"
+        );
+        assert!(
+            !slides.contains(&4_000),
+            "S=4_000 should be rejected (5_000 % 4_000 ≠ 0)"
+        );
+    }
+
+    #[test]
+    fn sliding_slide_must_divide_window_size() {
+        // W=6_000, t_repeat_gcd=6_000: slide_divisor = gcd(6_000, 6_000) = 6_000.
+        // S=4_000: 6_000 % 4_000 = 2_000 ≠ 0 → rejected even though 4_000 < 6_000.
+        // S=2_000: 6_000 % 2_000 = 0 → valid.
+        let aqe = make_aqe(Statistic::Sum, 12_000, 6_000);
+        let candidates = enumerate_candidates(&aqe, 1_000);
+
+        let slides_w6: Vec<u64> = candidates
+            .iter()
+            .filter(|c| {
+                c.config.as_ref().is_some_and(|cfg| {
+                    cfg.window_type == WindowType::Sliding && cfg.window_size_ms == 6_000
+                })
+            })
+            .map(|c| c.config.as_ref().unwrap().slide_interval_ms)
+            .collect();
+
+        assert!(
+            slides_w6.contains(&2_000),
+            "S=2_000 should be valid (6_000 % 2_000 = 0)"
+        );
+        assert!(
+            !slides_w6.contains(&4_000),
+            "S=4_000 should be rejected (6_000 % 4_000 ≠ 0)"
+        );
+    }
+
     #[test]
     fn partial_sliding_subtractable_sketch_gets_subtract() {
         // Sum → CMS (subtractable): partial Sliding with k=2 should produce Subtract.