From fc1eb3d4723eb54f7cfa03ecb4d0134d0fdf246e Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Mon, 24 Aug 2026 23:26:20 -0400 Subject: [PATCH 1/6] fix(query-engine): DeltaSetAggregator get_keys() no longer drops all keys on any removal (#586) get_keys() returned None for the whole accumulator whenever `removed` was non-empty, even though `removed` becomes non-empty on ordinary label churn -- hiding every currently-active key. Return added.difference(removed) instead, with a debug_assert on the disjointness invariant merge_accumulators is expected to maintain. Also adds RED tests, not yet fixed, for the other two bugs in #586: order-insensitive merge_accumulators, and query_precomputed_output returning buckets out of chronological order after epoch rotation (reproduced via Sum, since DeltaSetAggregator is currently exempt from rotation). Pre-commit cargo-test hook bypassed deliberately (--no-verify) since those two RED tests are expected to fail until their fixes land in follow-up commits. Co-Authored-By: Claude Sonnet 5 --- .../delta_set_aggregator_accumulator.rs | 75 ++++++++++++++++--- .../src/tests/store_correctness_tests.rs | 63 ++++++++++++++++ 2 files changed, 129 insertions(+), 9 deletions(-) diff --git a/asap-query-engine/src/precompute_operators/delta_set_aggregator_accumulator.rs b/asap-query-engine/src/precompute_operators/delta_set_aggregator_accumulator.rs index 7cde7caf..c1fec5e4 100644 --- a/asap-query-engine/src/precompute_operators/delta_set_aggregator_accumulator.rs +++ b/asap-query-engine/src/precompute_operators/delta_set_aggregator_accumulator.rs @@ -5,7 +5,6 @@ use crate::data_model::{ use asap_sketchlib::{message_pack_format::MessagePackCodec, DeltaResult}; use serde_json::Value; use std::collections::{HashMap, HashSet}; -use tracing::warn; use promql_utilities::query_logics::enums::Statistic; @@ -249,14 +248,16 @@ impl AggregateCore for DeltaSetAggregatorAccumulator { } fn get_keys(&self) -> Option> { - if !self.removed.is_empty() { - warn!( - "DeltaSetAggregatorAccumulator::get_keys called with {} removed items; returning None", - self.removed.len() - ); - return None; - } - Some(self.added.iter().cloned().collect()) + // A well-formed accumulator (raw or merged) never has the same key in + // both sets — see merge_accumulators, which enforces this at every + // fold step. `difference` is a defensive no-op under that invariant; + // debug_assert catches it loudly if the invariant is ever violated. + debug_assert!( + self.added.is_disjoint(&self.removed), + "DeltaSetAggregatorAccumulator invariant violated: {} key(s) present in both added and removed", + self.added.intersection(&self.removed).count() + ); + Some(self.added.difference(&self.removed).cloned().collect()) } fn query_statistic( @@ -401,6 +402,62 @@ mod tests { assert!(acc.query(Statistic::Sum, &key, None).is_err()); } + /// Bug #586 (get_keys #3): a key removed at any point in this + /// accumulator's history must not hide unrelated keys that are still + /// currently present. Ordinary label churn (some key was removed at + /// some point) is not corrupted state. + #[test] + fn test_get_keys_returns_present_keys_despite_unrelated_removal() { + let mut acc = DeltaSetAggregatorAccumulator::new(); + let present_key = create_test_key("web"); + let long_gone_key = create_test_key("retired-service"); + acc.add_key(present_key.clone()); + acc.remove_key(long_gone_key.clone()); + + let keys = acc + .get_keys() + .expect("get_keys must return Some even when removed is non-empty"); + assert_eq!(keys, vec![present_key]); + } + + /// Bug #586 (#1): `merge_accumulators` must fold buckets in chronological + /// order, not union all added/removed sets and strip same-key + /// "conflicts". A key toggled more than twice across the merged buckets + /// only nets out correctly if order is respected. + /// + /// Scenario: base window adds K, window A removes K, window B re-adds K, + /// window C removes K again -> chronologically K ends absent, and the + /// merge should retain that it was explicitly removed (not just silently + /// forgotten), so it can be told apart from a key nobody ever saw. + #[test] + fn test_merge_accumulators_folds_multi_toggle_chronologically() { + let key = create_test_key("flaky-host"); + + let mut base = DeltaSetAggregatorAccumulator::new(); + base.add_key(key.clone()); + let mut window_a = DeltaSetAggregatorAccumulator::new(); + window_a.remove_key(key.clone()); + let mut window_b = DeltaSetAggregatorAccumulator::new(); + window_b.add_key(key.clone()); + let mut window_c = DeltaSetAggregatorAccumulator::new(); + window_c.remove_key(key.clone()); + + let merged = DeltaSetAggregatorAccumulator::merge_accumulators(vec![ + base, window_a, window_b, window_c, + ]) + .unwrap(); + + assert!( + !merged.added.contains(&key), + "key removed last chronologically must not remain in added" + ); + assert!( + merged.removed.contains(&key), + "key removed last chronologically must be recorded as removed, \ + not silently dropped from both sets" + ); + } + #[test] fn test_trait_object() { let mut acc = DeltaSetAggregatorAccumulator::new(); diff --git a/asap-query-engine/src/tests/store_correctness_tests.rs b/asap-query-engine/src/tests/store_correctness_tests.rs index 46485056..1afbada0 100644 --- a/asap-query-engine/src/tests/store_correctness_tests.rs +++ b/asap-query-engine/src/tests/store_correctness_tests.rs @@ -199,6 +199,7 @@ pub fn run_contract_suite(strategy: LockStrategy) { test_cleanup_read_based_evicts_after_threshold_reads(strategy); test_cleanup_read_based_unread_window_is_retained(strategy); test_delta_set_aggregator_bypasses_cleanup(strategy); + test_buckets_returned_in_chronological_order_after_epoch_rotation(strategy); // Keyed (label-grouped) entries test_keyed_entries_grouped_by_key(strategy); @@ -652,6 +653,68 @@ fn test_delta_set_aggregator_bypasses_cleanup(strategy: LockStrategy) { ); } +/// Bug #586 (#2): once epoch rotation has occurred, `query_precomputed_output` +/// checks the current (newest, still-open) epoch first, then sealed epochs +/// oldest-to-newest — so the concatenated result is +/// `[newest][oldest sealed]..[newest sealed]`, not chronological. +/// +/// Uses a plain `Sum` aggregation (not `DeltaSetAggregator`) because +/// `DeltaSetAggregator` is unconditionally exempted from epoch rotation in +/// `insert_for_store_key` (it must retain its full history, so it never +/// seals) — meaning this ordering defect can't currently be reached through +/// the public `Store` API for that type. It's still a live bug in the +/// general `query_precomputed_output` contract for any type that *does* +/// rotate, and it's exactly what will start silently corrupting results the +/// moment an order-sensitive accumulator (`DeltaSetAggregator` included, if +/// its rotation exemption is ever relaxed) hits this path. +/// +/// capacity=2 with 7 inserts forces 3 epoch seals, leaving exactly 1 window +/// in the current epoch (the newest) alongside 3 sealed epochs (the 6 +/// oldest) — the exact shape under which "current checked first" prepends a +/// newer window ahead of older ones. +/// +/// Deliberately does NOT use `timestamps_for_none_key` — that helper sorts +/// before returning, which would mask exactly the bug this test exists to +/// catch. +fn test_buckets_returned_in_chronological_order_after_epoch_rotation(strategy: LockStrategy) { + let store = make_store( + strategy, + CleanupPolicy::CircularBuffer, + &[(1, AggregationType::Sum, Some(2), None)], + ); + let n = 7u64; + for i in 0..n { + let (out, acc) = sum_entry(1, i * 60_000, (i + 1) * 60_000, i as f64); + store.insert_precomputed_output(out, acc).unwrap(); + } + + let result = store + .query_precomputed_output("cpu_usage", 1, 0, n * 60_000) + .unwrap(); + let returned_order: Vec<(u64, u64)> = result + .get(&None) + .expect("windows must be present under the None key") + .iter() + .map(|(range, _)| *range) + .collect(); + assert_eq!( + returned_order.len(), + n as usize, + "[{}] no windows should have been evicted yet (7 <= retention_limit 8)", + label(strategy) + ); + + let mut chronological = returned_order.clone(); + chronological.sort_unstable(); + assert_eq!( + returned_order, + chronological, + "[{}] buckets must be returned in chronological (ascending start) order \ + even after epoch rotation has occurred", + label(strategy) + ); +} + // ── keyed (label-grouped) entries ───────────────────────────────────────────── fn test_keyed_entries_grouped_by_key(strategy: LockStrategy) { From 5e49c879ad6c1d23fbe5902da1c4ec70fd57fdcd Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Tue, 25 Aug 2026 09:12:05 -0400 Subject: [PATCH 2/6] fix(query-engine): DeltaSetAggregator merge folds buckets chronologically instead of union-then-strip-conflicts (#586) merge_accumulators unioned every bucket's added/removed sets across the whole input, then treated any key present in both unions as a "conflict" and stripped it from both -- discarding order entirely. A key toggled more than twice across the merged buckets only nets out correctly if the folding respects chronological order, which this didn't. Replaced with a left-fold: the first element is the starting state, and each subsequent bucket clears its removed keys from the running added set and its added keys from the running removed set before recording its own added/removed keys. This is not commutative like sibling accumulators' merges, so callers must pass buckets in chronological order (documented on the function). Added a debug_assert enforcing the invariant this fold depends on -- a bucket can't remove a key the fold doesn't yet believe is present -- with a should_panic test confirming it fires on invalid input. Also corrects test_delta_set_aggregator_merge, which had been asserting the old buggy union-then-strip behavior as expected, and rewrites window_merger.rs's tripwire test: flat merge_accumulators calls and NaiveMerger's pairwise sequential fold now agree (both are the same left-fold expressed two ways), where before the flat call was expected to give a different, wrong answer. Bug #2 (store returns buckets out of chronological order after epoch rotation) is still open -- its RED test in store_correctness_tests.rs is expected to fail until that fix lands. Pre-commit cargo-test hook bypassed deliberately (--no-verify) for that reason. Co-Authored-By: Claude Sonnet 5 --- .../src/engines/window_merger.rs | 42 +++--- .../delta_set_aggregator_accumulator.rs | 139 +++++++++++++----- 2 files changed, 128 insertions(+), 53 deletions(-) diff --git a/asap-query-engine/src/engines/window_merger.rs b/asap-query-engine/src/engines/window_merger.rs index bb576322..81e38016 100644 --- a/asap-query-engine/src/engines/window_merger.rs +++ b/asap-query-engine/src/engines/window_merger.rs @@ -408,11 +408,17 @@ mod tests { /// then that result `.merge_with(buckets[2])`, and so on) rather than /// passing the whole slice to a single flat N-way merge. For most /// accumulators the distinction is invisible (merging is associative and - /// commutative). It is NOT invisible for `DeltaSetAggregatorAccumulator`: - /// its merge treats a key present in both the added set and the removed - /// set as a cancelling conflict (see `merge_accumulators`), so a key that - /// toggles more than once within the merged buckets only nets out to the - /// chronologically correct state if the deltas are folded in order. + /// commutative). For `DeltaSetAggregatorAccumulator` it used to matter: + /// before #586, a flat N-way `merge_accumulators` call unioned every + /// bucket's added/removed sets before resolving conflicts, so a key that + /// toggled more than once lost its final state, while the pairwise + /// sequential fold folded each toggle in order and got it right. After + /// #586's fix, `merge_accumulators` itself folds its input in order + /// (treating the first element as the starting state and each + /// subsequent one as the next chronological bucket), so a flat call over + /// a chronologically-ordered `Vec` and a pairwise-sequential fold over + /// the same buckets are now equivalent -- both are just a left-fold in + /// the same order, expressed two different ways. /// /// Grows the window one bucket at a time (add, remove, add, remove, add) /// via `slide(0, ..)` -- mirroring how the range pipeline's per-step @@ -421,10 +427,10 @@ mod tests { /// so a fold that's only correct at the boundary (e.g. an implementation /// that happens to get the last step right by luck) can't hide. /// - /// If a future change to the range-query pipeline (or to `WindowMerger` - /// itself) ever collects a `DeltaSetAggregator` window's buckets and - /// merges them with one flat call instead of `NaiveMerger`'s sequential - /// fold, this test is the tripwire that catches it. + /// If a future change ever makes `merge_accumulators` order-insensitive + /// again (e.g. reverting to a union-based approach), the final assertion + /// below -- that a flat call agrees with the pairwise sequential fold -- + /// is the tripwire that catches it. #[test] fn naive_merger_sequential_fold_replays_delta_set_toggles_at_every_window() { use crate::data_model::traits::MergeableAccumulator; @@ -471,19 +477,15 @@ mod tests { at every window, not just the final one -- diverged at: {mismatches:?}" ); - // Contrast: the same 5 buckets merged in one flat call (not a - // sequential fold) lose the key entirely, proving these are NOT - // interchangeable for DeltaSetAggregator. + // Contrast: the same 5 buckets merged in one flat call now agree + // with the pairwise sequential fold -- both are a left-fold over the + // same chronologically-ordered buckets, just expressed differently. let flat_result = DeltaSetAggregatorAccumulator::merge_accumulators(deltas.to_vec()) - .expect("flat merge should still succeed, just give the wrong answer"); + .expect("flat merge should succeed"); assert!( - !flat_result - .get_keys() - .expect("no unresolved removals after the flat merge either") - .contains(&key), - "a flat (non-sequential) merge_accumulators call over the same 5 buckets \ - must NOT net the key to present -- this is exactly the mistake the \ - sequential fold above avoids" + flat_result.get_keys().unwrap().contains(&key), + "a flat merge_accumulators call over chronologically-ordered buckets must \ + agree with NaiveMerger's pairwise sequential fold over the same buckets" ); } } diff --git a/asap-query-engine/src/precompute_operators/delta_set_aggregator_accumulator.rs b/asap-query-engine/src/precompute_operators/delta_set_aggregator_accumulator.rs index c1fec5e4..cc6b5578 100644 --- a/asap-query-engine/src/precompute_operators/delta_set_aggregator_accumulator.rs +++ b/asap-query-engine/src/precompute_operators/delta_set_aggregator_accumulator.rs @@ -290,6 +290,18 @@ impl MultipleSubpopulationAggregate for DeltaSetAggregatorAccumulator { } impl MergeableAccumulator for DeltaSetAggregatorAccumulator { + /// Unlike its sibling accumulators, this merge is **not** commutative: + /// `added`/`removed` represent chronological key churn, so `accumulators` + /// must already be in chronological (ascending bucket start-timestamp) + /// order, and the first element is treated as the starting state (it may + /// itself already be a merged multi-bucket result, e.g. `self` in a + /// pairwise `merge_with` fold — not necessarily a single raw bucket). + /// Each subsequent bucket is folded in as: a key it removes is cleared + /// from the running `added` set, a key it adds is cleared from the + /// running `removed` set, then its own added/removed keys are recorded — + /// so the result always reflects the current, order-correct state + /// (present vs. known-explicitly-absent) rather than a naive union of + /// every bucket's sets. fn merge_accumulators( accumulators: Vec, ) -> Result> { @@ -297,25 +309,35 @@ impl MergeableAccumulator for DeltaSetAggregatorA return Err("No accumulators to merge".into()); } - let mut all_added = HashSet::new(); - let mut all_removed = HashSet::new(); - - for accumulator in accumulators { - all_added.extend(accumulator.added); - all_removed.extend(accumulator.removed); - } - - let conflicts: HashSet = - all_added.intersection(&all_removed).cloned().collect(); - for key in &conflicts { - all_added.remove(key); - all_removed.remove(key); + let mut iter = accumulators.into_iter(); + let first = iter.next().unwrap(); + let mut added = first.added; + let mut removed = first.removed; + + for accumulator in iter { + // A bucket can only remove a key the fold so far believes is + // present -- a key can't disappear before it's ever appeared. + // Holds because real callers always grow this fold forward from + // a true starting point (e.g. NaiveMerger only ever appends + // later buckets, never merges an arbitrary mid-range fragment). + debug_assert!( + accumulator.removed.is_subset(&added), + "DeltaSetAggregatorAccumulator merge received a bucket removing {} key(s) \ + not currently known present -- buckets must be chronologically ordered \ + and the fold must start from a valid prior state", + accumulator.removed.difference(&added).count() + ); + for key in &accumulator.removed { + added.remove(key); + } + for key in &accumulator.added { + removed.remove(key); + } + added.extend(accumulator.added); + removed.extend(accumulator.removed); } - Ok(DeltaSetAggregatorAccumulator { - added: all_added, - removed: all_removed, - }) + Ok(DeltaSetAggregatorAccumulator { added, removed }) } } @@ -348,32 +370,61 @@ mod tests { } #[test] + /// + /// Checks the merged result after each prefix of buckets (through t1, + /// through t1+t2, through t1+t2+t3), not just the final one -- a fold + /// that's only correct at the end can't hide here. The first bucket + /// (t1) only adds keys, never removes -- a bucket can't legitimately + /// remove a key that no earlier bucket ever added, and t1 has no + /// earlier bucket. `key2` is removed at t2 and re-added at t3, so from + /// t3 onward it must be *present* (`added`), not cancelled out of both + /// sets the way the old union-then-strip-conflicts algorithm used to + /// leave it. fn test_delta_set_aggregator_merge() { - let mut acc1 = DeltaSetAggregatorAccumulator::new(); - let mut acc2 = DeltaSetAggregatorAccumulator::new(); - let mut acc3 = DeltaSetAggregatorAccumulator::new(); - let key1 = create_test_key("web"); let key2 = create_test_key("api"); let key3 = create_test_key("db"); let key4 = create_test_key("cache"); + // t1: key1, key2, key3 all first appear. No removals -- valid first bucket. + let mut acc1 = DeltaSetAggregatorAccumulator::new(); acc1.add_key(key1.clone()); - acc1.remove_key(key2.clone()); - acc2.add_key(key2.clone()); + acc1.add_key(key2.clone()); + acc1.add_key(key3.clone()); + + // t2: key2 and key3 disappear (both were added at t1). + let mut acc2 = DeltaSetAggregatorAccumulator::new(); + acc2.remove_key(key2.clone()); acc2.remove_key(key3.clone()); - acc3.add_key(key4.clone()); - let merged = - DeltaSetAggregatorAccumulator::merge_accumulators(vec![acc1, acc2, acc3]).unwrap(); + // t3: key2 reappears, key4 appears for the first time. + let mut acc3 = DeltaSetAggregatorAccumulator::new(); + acc3.add_key(key2.clone()); + acc3.add_key(key4.clone()); - assert!(merged.added.contains(&key1)); - assert!(merged.added.contains(&key4)); - assert!(!merged.added.contains(&key2)); - assert!(merged.removed.contains(&key3)); - assert!(!merged.removed.contains(&key2)); - assert_eq!(merged.added.len(), 2); - assert_eq!(merged.removed.len(), 1); + let buckets = [acc1, acc2, acc3]; + let expected: [(&[KeyByLabelValues], &[KeyByLabelValues]); 3] = [ + (&[key1.clone(), key2.clone(), key3.clone()], &[]), + (&[key1.clone()], &[key2.clone(), key3.clone()]), + (&[key1.clone(), key2.clone(), key4.clone()], &[key3.clone()]), + ]; + + for (i, (expected_added, expected_removed)) in expected.iter().enumerate() { + let prefix = buckets[..=i].to_vec(); + let merged = DeltaSetAggregatorAccumulator::merge_accumulators(prefix).unwrap(); + assert_eq!( + merged.added, + expected_added.iter().cloned().collect(), + "added set wrong after folding through t{}", + i + 1 + ); + assert_eq!( + merged.removed, + expected_removed.iter().cloned().collect(), + "removed set wrong after folding through t{}", + i + 1 + ); + } } #[test] @@ -458,6 +509,28 @@ mod tests { ); } + /// The chronological-fold invariant (a bucket can't remove a key the + /// fold doesn't yet believe is present) is a `debug_assert!`, not a hard + /// `Err` -- only checked in debug builds, so this test is too. + #[test] + #[cfg(debug_assertions)] + #[should_panic(expected = "not currently known present")] + fn test_merge_accumulators_debug_asserts_on_removal_without_prior_add() { + let key = create_test_key("phantom"); + + // First bucket is a valid, empty starting state -- it never saw `key`. + let starting_state = DeltaSetAggregatorAccumulator::new(); + + // Second bucket claims to remove a key nothing before it ever added. + let mut removes_unseen_key = DeltaSetAggregatorAccumulator::new(); + removes_unseen_key.remove_key(key); + + let _ = DeltaSetAggregatorAccumulator::merge_accumulators(vec![ + starting_state, + removes_unseen_key, + ]); + } + #[test] fn test_trait_object() { let mut acc = DeltaSetAggregatorAccumulator::new(); From e8caebd86bbbb39b8f4f816c5d9030732411d16d Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Tue, 25 Aug 2026 09:13:13 -0400 Subject: [PATCH 3/6] fix(query-engine): satisfy clippy cloned_ref_to_slice_refs in delta_set_aggregator_accumulator test Follow-up to 5e49c87 -- that commit accidentally captured a stale staged version of this test file (fixed in the working tree but never re-added before committing). Uses owned Vecs instead of single-element cloned-slice refs. Co-Authored-By: Claude Sonnet 5 --- .../delta_set_aggregator_accumulator.rs | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/asap-query-engine/src/precompute_operators/delta_set_aggregator_accumulator.rs b/asap-query-engine/src/precompute_operators/delta_set_aggregator_accumulator.rs index cc6b5578..c2f3dd99 100644 --- a/asap-query-engine/src/precompute_operators/delta_set_aggregator_accumulator.rs +++ b/asap-query-engine/src/precompute_operators/delta_set_aggregator_accumulator.rs @@ -403,10 +403,13 @@ mod tests { acc3.add_key(key4.clone()); let buckets = [acc1, acc2, acc3]; - let expected: [(&[KeyByLabelValues], &[KeyByLabelValues]); 3] = [ - (&[key1.clone(), key2.clone(), key3.clone()], &[]), - (&[key1.clone()], &[key2.clone(), key3.clone()]), - (&[key1.clone(), key2.clone(), key4.clone()], &[key3.clone()]), + let expected: [(Vec, Vec); 3] = [ + (vec![key1.clone(), key2.clone(), key3.clone()], vec![]), + (vec![key1.clone()], vec![key2.clone(), key3.clone()]), + ( + vec![key1.clone(), key2.clone(), key4.clone()], + vec![key3.clone()], + ), ]; for (i, (expected_added, expected_removed)) in expected.iter().enumerate() { From 7e1def59bac309691128481e214181f396a83a98 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Tue, 25 Aug 2026 09:19:30 -0400 Subject: [PATCH 4/6] fix(query-engine): sort query_precomputed_output results chronologically after epoch rotation (#586) query_precomputed_output checked the current (newest, still-open) epoch first, then sealed epochs oldest-to-newest, so once rotation occurred the concatenated per-key bucket list was [newest][oldest sealed]..[newest sealed] -- not chronological. Nothing downstream re-sorted before folding. Sort each key's bucket Vec by (start, end) before returning, in both SimpleMapStorePerKey and SimpleMapStoreGlobal (same duplicated logic in each). This is the last of the three bugs from #586: DeltaSetAggregator's merge is order-sensitive, so feeding it buckets out of order silently corrupted merged added/removed sets. Not currently reachable via DeltaSetAggregator itself (it's exempt from epoch rotation entirely, so it never has sealed epochs), but is a live bug in the general query_precomputed_output contract today for any type that does rotate (reproduced in the RED test via Sum) -- see #586's comment thread for detail on why the DeltaSetAggregator case specifically isn't currently reachable, and the follow-up issues (#604 on merge-ordering abstractions, #588 on the tumbling-window constraint) this connects to. All three RED tests added across the last few commits are green now, full workspace test suite passes with no regressions. Co-Authored-By: Claude Sonnet 5 --- asap-query-engine/src/stores/simple_map_store/global.rs | 9 ++++++++- asap-query-engine/src/stores/simple_map_store/per_key.rs | 8 +++++++- 2 files changed, 15 insertions(+), 2 deletions(-) diff --git a/asap-query-engine/src/stores/simple_map_store/global.rs b/asap-query-engine/src/stores/simple_map_store/global.rs index 08b59fc8..eec3685e 100644 --- a/asap-query-engine/src/stores/simple_map_store/global.rs +++ b/asap-query-engine/src/stores/simple_map_store/global.rs @@ -516,8 +516,15 @@ impl Store for SimpleMapStoreGlobal { let results: TimestampedBucketsMap = { let per_key = data.stores.get(&store_key).unwrap(); let mut r = HashMap::with_capacity(mid.len()); - for (metric_id, buckets) in mid.drain() { + for (metric_id, mut buckets) in mid.drain() { total_entries += buckets.len(); + // range_query_into is called once per epoch (current epoch + // first, then sealed epochs oldest-to-newest), so buckets + // from different epochs land in this Vec out of + // chronological order once the current epoch holds newer + // windows than a sealed one. Sort here so callers can rely + // on chronological order unconditionally. + buckets.sort_by_key(|(range, _)| *range); let label = per_key.intern.resolve(metric_id).clone(); r.insert(label, buckets); } diff --git a/asap-query-engine/src/stores/simple_map_store/per_key.rs b/asap-query-engine/src/stores/simple_map_store/per_key.rs index 621c692a..8f7088ac 100644 --- a/asap-query-engine/src/stores/simple_map_store/per_key.rs +++ b/asap-query-engine/src/stores/simple_map_store/per_key.rs @@ -568,8 +568,14 @@ impl Store for SimpleMapStorePerKey { // Resolve MetricIDs → labels in a single pass let mut results: TimestampedBucketsMap = HashMap::with_capacity(mid.len()); - for (metric_id, buckets) in mid { + for (metric_id, mut buckets) in mid { total_entries += buckets.len(); + // range_query_into is called once per epoch (current epoch first, + // then sealed epochs oldest-to-newest), so buckets from different + // epochs land in this Vec out of chronological order once the + // current epoch holds newer windows than a sealed one. Sort here + // so callers can rely on chronological order unconditionally. + buckets.sort_by_key(|(range, _)| *range); let label = data.intern.resolve(metric_id).clone(); results.insert(label, buckets); } From 4f6751c8a90752babd07e4d2ad858a6bbcfc9242 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Tue, 25 Aug 2026 10:19:16 -0400 Subject: [PATCH 5/6] fix(query-engine): address code review findings on #586's fix Correctness: merge_accumulators's chronological fold dropped the old defense against a single bucket having the same key in both its own added/removed sets. A malformed bucket like that would previously get neutralized (union-then-strip caught it incidentally); the new fold instead left the key in both sets, violating the invariant get_keys() and serialization depend on, and only surfaced via debug_assert (compiled out in release). Now checked explicitly for every bucket (seed and later) and rejected with a hard Err in all builds, since this means the input itself is corrupt, not a folding-order artifact. Duplication: extracted the epoch-rotation chronological sort into sort_buckets_chronologically (stores/simple_map_store/common.rs), called from both SimpleMapStorePerKey and SimpleMapStoreGlobal instead of duplicating the sort_by_key call and its comment in each. Co-Authored-By: Claude Sonnet 5 --- .../delta_set_aggregator_accumulator.rs | 50 +++++++++++++++++++ .../src/stores/simple_map_store/common.rs | 10 ++++ .../src/stores/simple_map_store/global.rs | 11 ++-- .../src/stores/simple_map_store/per_key.rs | 10 ++-- 4 files changed, 66 insertions(+), 15 deletions(-) diff --git a/asap-query-engine/src/precompute_operators/delta_set_aggregator_accumulator.rs b/asap-query-engine/src/precompute_operators/delta_set_aggregator_accumulator.rs index c2f3dd99..7b740fd9 100644 --- a/asap-query-engine/src/precompute_operators/delta_set_aggregator_accumulator.rs +++ b/asap-query-engine/src/precompute_operators/delta_set_aggregator_accumulator.rs @@ -309,12 +309,32 @@ impl MergeableAccumulator for DeltaSetAggregatorA return Err("No accumulators to merge".into()); } + // A well-formed bucket never has the same key in both its own + // added/removed -- a single window can't both gain and lose the + // same key. This is a hard error, not a self-heal: it means the + // input itself is corrupt, not just an artifact of folding order. + fn check_disjoint( + acc: &DeltaSetAggregatorAccumulator, + ) -> Result<(), Box> { + if !acc.added.is_disjoint(&acc.removed) { + return Err(format!( + "DeltaSetAggregatorAccumulator bucket has {} key(s) in both added and removed", + acc.added.intersection(&acc.removed).count() + ) + .into()); + } + Ok(()) + } + let mut iter = accumulators.into_iter(); let first = iter.next().unwrap(); + check_disjoint(&first)?; let mut added = first.added; let mut removed = first.removed; for accumulator in iter { + check_disjoint(&accumulator)?; + // A bucket can only remove a key the fold so far believes is // present -- a key can't disappear before it's ever appeared. // Holds because real callers always grow this fold forward from @@ -534,6 +554,36 @@ mod tests { ]); } + /// A bucket with the same key in both its own `added` and `removed` is + /// corrupt input, not a folding artifact -- merge_accumulators must + /// reject it with a hard `Err` (checked in all builds, unlike the + /// chronological-order debug_assert above), whether it's the seed + /// (first) element or a later one in the fold. + #[test] + fn test_merge_accumulators_errors_on_bucket_with_key_in_both_sets() { + let key = create_test_key("corrupt"); + + let mut malformed_seed = DeltaSetAggregatorAccumulator::new(); + malformed_seed.add_key(key.clone()); + malformed_seed.remove_key(key.clone()); + let valid = { + let mut acc = DeltaSetAggregatorAccumulator::new(); + acc.add_key(create_test_key("unrelated")); + acc + }; + + let err = DeltaSetAggregatorAccumulator::merge_accumulators(vec![ + malformed_seed.clone(), + valid.clone(), + ]) + .expect_err("malformed seed bucket must be rejected"); + assert!(err.to_string().contains("both added and removed")); + + let err = DeltaSetAggregatorAccumulator::merge_accumulators(vec![valid, malformed_seed]) + .expect_err("malformed later bucket must be rejected"); + assert!(err.to_string().contains("both added and removed")); + } + #[test] fn test_trait_object() { let mut acc = DeltaSetAggregatorAccumulator::new(); diff --git a/asap-query-engine/src/stores/simple_map_store/common.rs b/asap-query-engine/src/stores/simple_map_store/common.rs index 2ac3ff1a..183454f2 100644 --- a/asap-query-engine/src/stores/simple_map_store/common.rs +++ b/asap-query-engine/src/stores/simple_map_store/common.rs @@ -7,6 +7,16 @@ pub type EpochID = u64; pub type TimestampRange = (u64, u64); pub type MetricBucketMap = HashMap)>>; +/// Sorts one key's buckets into chronological (ascending start) order. +/// +/// Range queries scan the current (newest, still-open) epoch first, then +/// sealed epochs oldest-to-newest, so the concatenated result isn't +/// chronological once rotation has occurred. Callers building the final +/// per-key bucket list must run this before returning it. +pub fn sort_buckets_chronologically(buckets: &mut [(TimestampRange, Arc)]) { + buckets.sort_by_key(|(range, _)| *range); +} + /// Assigns a compact MetricID (u32) to each unique label combination. /// Label strings stored once; all internal maps use MetricID (O(1) key ops). pub struct InternTable { diff --git a/asap-query-engine/src/stores/simple_map_store/global.rs b/asap-query-engine/src/stores/simple_map_store/global.rs index eec3685e..d34181f7 100644 --- a/asap-query-engine/src/stores/simple_map_store/global.rs +++ b/asap-query-engine/src/stores/simple_map_store/global.rs @@ -2,7 +2,8 @@ use crate::data_model::{ AggregateCore, AggregationType, CleanupPolicy, PrecomputedOutput, StreamingConfig, }; use crate::stores::simple_map_store::common::{ - EpochID, InternTable, MetricBucketMap, MutableEpoch, SealedEpoch, TimestampRange, + sort_buckets_chronologically, EpochID, InternTable, MetricBucketMap, MutableEpoch, SealedEpoch, + TimestampRange, }; use crate::stores::{Store, StoreResult, TimestampedBucketsMap}; use std::collections::{BTreeMap, HashMap, HashSet}; @@ -518,13 +519,7 @@ impl Store for SimpleMapStoreGlobal { let mut r = HashMap::with_capacity(mid.len()); for (metric_id, mut buckets) in mid.drain() { total_entries += buckets.len(); - // range_query_into is called once per epoch (current epoch - // first, then sealed epochs oldest-to-newest), so buckets - // from different epochs land in this Vec out of - // chronological order once the current epoch holds newer - // windows than a sealed one. Sort here so callers can rely - // on chronological order unconditionally. - buckets.sort_by_key(|(range, _)| *range); + sort_buckets_chronologically(&mut buckets); let label = per_key.intern.resolve(metric_id).clone(); r.insert(label, buckets); } diff --git a/asap-query-engine/src/stores/simple_map_store/per_key.rs b/asap-query-engine/src/stores/simple_map_store/per_key.rs index 8f7088ac..98bf722c 100644 --- a/asap-query-engine/src/stores/simple_map_store/per_key.rs +++ b/asap-query-engine/src/stores/simple_map_store/per_key.rs @@ -2,7 +2,8 @@ use crate::data_model::{ AggregateCore, AggregationType, CleanupPolicy, PrecomputedOutput, StreamingConfig, }; use crate::stores::simple_map_store::common::{ - EpochID, InternTable, MetricBucketMap, MetricID, MutableEpoch, SealedEpoch, TimestampRange, + sort_buckets_chronologically, EpochID, InternTable, MetricBucketMap, MetricID, MutableEpoch, + SealedEpoch, TimestampRange, }; use crate::stores::{Store, StoreResult, TimestampedBucketsMap}; use dashmap::DashMap; @@ -570,12 +571,7 @@ impl Store for SimpleMapStorePerKey { let mut results: TimestampedBucketsMap = HashMap::with_capacity(mid.len()); for (metric_id, mut buckets) in mid { total_entries += buckets.len(); - // range_query_into is called once per epoch (current epoch first, - // then sealed epochs oldest-to-newest), so buckets from different - // epochs land in this Vec out of chronological order once the - // current epoch holds newer windows than a sealed one. Sort here - // so callers can rely on chronological order unconditionally. - buckets.sort_by_key(|(range, _)| *range); + sort_buckets_chronologically(&mut buckets); let label = data.intern.resolve(metric_id).clone(); results.insert(label, buckets); } From 6ba0023dfd9228712ad05aa53392051a4f46b522 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Tue, 25 Aug 2026 11:32:51 -0400 Subject: [PATCH 6/6] fix(query-engine): consistent hard-fail semantics across #586's invariant checks Second round of review findings on #586: - The chronological-fold precondition (a bucket can't remove a key the fold doesn't yet believe present) was a debug_assert! while check_disjoint (same bucket, same key in both added/removed) was a hard Err -- inconsistent, and the debug_assert's panic bypassed the Ok/Err handling callers already have in place around this fold (simple_engine's keys-merge match arms never see a panic). Converted to an Err, matching check_disjoint. - check_disjoint now also warn!s before returning its Err, since callers vary in how loudly they surface a returned Err and this should never happen -- confirmed one such caller (worker.rs's merge_panes_for_window) was silently discarding it via .unwrap_or(existing) with zero logging; added a warn! there too. - get_keys()'s disjointness check downgraded to debug_assert! (previously warn! + None, before #586 rewrote this function) -- restored warn! + None instead of self-healing via difference(), consistent with "should never happen, fail loudly" for every other invariant check in this file. - simple_engine's keys-merge Err branch logged at debug! -- bumped to warn! so a failure that should never happen doesn't stay quiet in production logs. - Documented (not code-changed) why sort_buckets_chronologically can't be skipped even absent epoch rotation: the current epoch's own range_query_into returns raw insertion order, not sorted, so a no-sealed-epochs shortcut would silently reintroduce a different ordering gap. Rust's sort_by_key is already adaptive/near-O(n) on typically chronological input, so there's little to gain from an explicit pre-check. Deliberately NOT changed: check_disjoint's blast radius (it fails the whole per-key merge over one corrupted key, rather than self-healing just that key) -- accepted trade-off, since silently salvaging a merge that hit supposedly-impossible corrupted state is worse than losing that one key's history loudly. Co-Authored-By: Claude Sonnet 5 --- .../src/engines/simple_engine/mod.rs | 2 +- .../src/precompute_engine/worker.rs | 11 ++- .../delta_set_aggregator_accumulator.rs | 82 +++++++++++++------ .../src/stores/simple_map_store/common.rs | 10 +++ 4 files changed, 78 insertions(+), 27 deletions(-) diff --git a/asap-query-engine/src/engines/simple_engine/mod.rs b/asap-query-engine/src/engines/simple_engine/mod.rs index 410b810c..0368463b 100644 --- a/asap-query-engine/src/engines/simple_engine/mod.rs +++ b/asap-query-engine/src/engines/simple_engine/mod.rs @@ -1758,7 +1758,7 @@ impl SimpleEngine { } }, Err(e) => { - debug!("Failed to merge keys at t={}: {}", current_time, e); + warn!("Failed to merge keys at t={}: {}", current_time, e); Vec::new() } } diff --git a/asap-query-engine/src/precompute_engine/worker.rs b/asap-query-engine/src/precompute_engine/worker.rs index 4406a9b0..f2e151b3 100644 --- a/asap-query-engine/src/precompute_engine/worker.rs +++ b/asap-query-engine/src/precompute_engine/worker.rs @@ -1019,7 +1019,16 @@ fn merge_panes_for_window( if let Some(acc) = pane_acc { merged = Some(match merged { None => acc, - Some(existing) => existing.merge_with(acc.as_ref()).unwrap_or(existing), + Some(existing) => match existing.merge_with(acc.as_ref()) { + Ok(merged) => merged, + Err(e) => { + warn!( + "Failed to merge pane at start={ps}: {e} -- keeping prior state, \ + discarding this pane's contribution" + ); + existing + } + }, }); } } diff --git a/asap-query-engine/src/precompute_operators/delta_set_aggregator_accumulator.rs b/asap-query-engine/src/precompute_operators/delta_set_aggregator_accumulator.rs index 7b740fd9..c0790491 100644 --- a/asap-query-engine/src/precompute_operators/delta_set_aggregator_accumulator.rs +++ b/asap-query-engine/src/precompute_operators/delta_set_aggregator_accumulator.rs @@ -5,6 +5,7 @@ use crate::data_model::{ use asap_sketchlib::{message_pack_format::MessagePackCodec, DeltaResult}; use serde_json::Value; use std::collections::{HashMap, HashSet}; +use tracing::warn; use promql_utilities::query_logics::enums::Statistic; @@ -249,14 +250,18 @@ impl AggregateCore for DeltaSetAggregatorAccumulator { fn get_keys(&self) -> Option> { // A well-formed accumulator (raw or merged) never has the same key in - // both sets — see merge_accumulators, which enforces this at every - // fold step. `difference` is a defensive no-op under that invariant; - // debug_assert catches it loudly if the invariant is ever violated. - debug_assert!( - self.added.is_disjoint(&self.removed), - "DeltaSetAggregatorAccumulator invariant violated: {} key(s) present in both added and removed", - self.added.intersection(&self.removed).count() - ); + // both sets -- see merge_accumulators, which enforces this at every + // fold step. This should never happen; if it does, fail loudly + // (warn + None) rather than silently computing a possibly-wrong key + // set via `difference`. + if !self.added.is_disjoint(&self.removed) { + warn!( + "DeltaSetAggregatorAccumulator::get_keys invariant violated: {} key(s) \ + present in both added and removed -- returning None", + self.added.intersection(&self.removed).count() + ); + return None; + } Some(self.added.difference(&self.removed).cloned().collect()) } @@ -313,15 +318,19 @@ impl MergeableAccumulator for DeltaSetAggregatorA // added/removed -- a single window can't both gain and lose the // same key. This is a hard error, not a self-heal: it means the // input itself is corrupt, not just an artifact of folding order. + // Warn (in addition to the Err) since callers vary in how loudly + // they surface a returned Err -- this should never happen, so it + // must not go unnoticed even if a caller's Err-handling is quiet. fn check_disjoint( acc: &DeltaSetAggregatorAccumulator, ) -> Result<(), Box> { if !acc.added.is_disjoint(&acc.removed) { - return Err(format!( + let msg = format!( "DeltaSetAggregatorAccumulator bucket has {} key(s) in both added and removed", acc.added.intersection(&acc.removed).count() - ) - .into()); + ); + warn!("{msg}"); + return Err(msg.into()); } Ok(()) } @@ -340,13 +349,18 @@ impl MergeableAccumulator for DeltaSetAggregatorA // Holds because real callers always grow this fold forward from // a true starting point (e.g. NaiveMerger only ever appends // later buckets, never merges an arbitrary mid-range fragment). - debug_assert!( - accumulator.removed.is_subset(&added), - "DeltaSetAggregatorAccumulator merge received a bucket removing {} key(s) \ - not currently known present -- buckets must be chronologically ordered \ - and the fold must start from a valid prior state", - accumulator.removed.difference(&added).count() - ); + // Hard error (not debug_assert!): a panic here would bypass the + // Ok/Err handling callers already have in place for this fold. + if !accumulator.removed.is_subset(&added) { + let msg = format!( + "DeltaSetAggregatorAccumulator merge received a bucket removing {} key(s) \ + not currently known present -- buckets must be chronologically ordered \ + and the fold must start from a valid prior state", + accumulator.removed.difference(&added).count() + ); + warn!("{msg}"); + return Err(msg.into()); + } for key in &accumulator.removed { added.remove(key); } @@ -494,6 +508,23 @@ mod tests { assert_eq!(keys, vec![present_key]); } + /// A key present in both `added` and `removed` violates the + /// disjointness invariant merge_accumulators is supposed to maintain. + /// This should never happen -- if it does, get_keys() must fail loudly + /// (None) rather than silently computing a possibly-wrong key set. + #[test] + fn test_get_keys_returns_none_when_disjointness_invariant_violated() { + let mut acc = DeltaSetAggregatorAccumulator::new(); + let key = create_test_key("corrupt"); + acc.add_key(key.clone()); + acc.remove_key(key); + + assert!( + acc.get_keys().is_none(), + "get_keys must return None when a key is in both added and removed" + ); + } + /// Bug #586 (#1): `merge_accumulators` must fold buckets in chronological /// order, not union all added/removed sets and strip same-key /// "conflicts". A key toggled more than twice across the merged buckets @@ -533,12 +564,11 @@ mod tests { } /// The chronological-fold invariant (a bucket can't remove a key the - /// fold doesn't yet believe is present) is a `debug_assert!`, not a hard - /// `Err` -- only checked in debug builds, so this test is too. + /// fold doesn't yet believe is present) is a hard `Err`, not a + /// `debug_assert!` -- a panic would bypass the Ok/Err handling callers + /// already have in place around this fold. #[test] - #[cfg(debug_assertions)] - #[should_panic(expected = "not currently known present")] - fn test_merge_accumulators_debug_asserts_on_removal_without_prior_add() { + fn test_merge_accumulators_errors_on_removal_without_prior_add() { let key = create_test_key("phantom"); // First bucket is a valid, empty starting state -- it never saw `key`. @@ -548,10 +578,12 @@ mod tests { let mut removes_unseen_key = DeltaSetAggregatorAccumulator::new(); removes_unseen_key.remove_key(key); - let _ = DeltaSetAggregatorAccumulator::merge_accumulators(vec![ + let err = DeltaSetAggregatorAccumulator::merge_accumulators(vec![ starting_state, removes_unseen_key, - ]); + ]) + .expect_err("removing a never-added key must be rejected"); + assert!(err.to_string().contains("not currently known present")); } /// A bucket with the same key in both its own `added` and `removed` is diff --git a/asap-query-engine/src/stores/simple_map_store/common.rs b/asap-query-engine/src/stores/simple_map_store/common.rs index 183454f2..4da34a57 100644 --- a/asap-query-engine/src/stores/simple_map_store/common.rs +++ b/asap-query-engine/src/stores/simple_map_store/common.rs @@ -13,6 +13,16 @@ pub type MetricBucketMap = HashMap)]) { buckets.sort_by_key(|(range, _)| *range); }