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/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_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 7cde7caf..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 @@ -249,14 +249,20 @@ impl AggregateCore for DeltaSetAggregatorAccumulator { } fn get_keys(&self) -> Option> { - if !self.removed.is_empty() { + // 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. 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 called with {} removed items; returning None", - self.removed.len() + "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.iter().cloned().collect()) + Some(self.added.difference(&self.removed).cloned().collect()) } fn query_statistic( @@ -289,6 +295,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> { @@ -296,25 +314,64 @@ 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); + // 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. + // 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) { + let msg = format!( + "DeltaSetAggregatorAccumulator bucket has {} key(s) in both added and removed", + acc.added.intersection(&acc.removed).count() + ); + warn!("{msg}"); + return Err(msg.into()); + } + Ok(()) } - 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(); + 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 + // a true starting point (e.g. NaiveMerger only ever appends + // later buckets, never merges an arbitrary mid-range fragment). + // 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); + } + 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 }) } } @@ -347,32 +404,64 @@ 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: [(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() { + 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] @@ -401,6 +490,132 @@ 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]); + } + + /// 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 + /// 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" + ); + } + + /// The chronological-fold invariant (a bucket can't remove a key the + /// 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] + 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`. + 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 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 + /// 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..4da34a57 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,26 @@ 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. +/// +/// 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)]) { + 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 08b59fc8..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}; @@ -516,8 +517,9 @@ 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(); + 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 621c692a..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; @@ -568,8 +569,9 @@ 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(); + sort_buckets_chronologically(&mut buckets); let label = data.intern.resolve(metric_id).clone(); results.insert(label, buckets); } 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) {