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 <noreply@anthropic.com>

Author: milindsrivastava1997

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()
                                 }
                             }

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
+                    }
+                },
             });
         }
     }

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<Vec<KeyByLabelValues>> {
         // 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<DeltaSetAggregatorAccumulator> 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<dyn std::error::Error + Send + Sync>> {
             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<DeltaSetAggregatorAccumulator> 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

asap-query-engine/src/stores/simple_map_store/common.rs

@@ -13,6 +13,16 @@ pub type MetricBucketMap = HashMap<MetricID, Vec<(TimestampRange, Arc<dyn Aggreg
 /// 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.
+///
+/// Runs unconditionally rather than only after detected rotation: the
+/// current epoch's own `range_query_into` (`MutableEpoch`) returns buckets
+/// in raw insertion order, not sorted, so an out-of-order insert can violate
+/// chronological order even with a single epoch that's never rotated.
+/// Skipping this based on "no sealed epochs" would silently reintroduce
+/// that gap. `sort_by_key`'s adaptive (Timsort-derived) algorithm is
+/// already close to O(n) on the common case of already- or
+/// mostly-chronological input, so there's little to gain from an explicit
+/// pre-check.
 pub fn sort_buckets_chronologically(buckets: &mut [(TimestampRange, Arc<dyn AggregateCore>)]) {
     buckets.sort_by_key(|(range, _)| *range);
 }