From 7a610ce0f2128d4c5195f35657b514f3a45155cd Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 24 May 2026 10:54:35 -0600 Subject: [PATCH] perf: reduce backend precompute engine CPU MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The OTLP warm-tier ingest path ran `reconcile_from_streaming_config` on *every* ingest batch, and that function deep-cloned every `SketchInstanceMetadata` in the catalog via `snapshot_instances()` (each carries a `String` + `BTreeSet` + `AggKind` strings). Live `perf` on node2's `asap-backend` showed this as the dominant backend CPU cost: `BTreeMap::clone::clone_subtree` + `String::clone` + `SketchInstanceMetadata::clone` + drops, plus the malloc/free churn they drive (~23% in malloc/cfree alone), all under `reconcile_from_streaming_config`. Two focused changes, both correctness-preserving: 1. Eliminate the per-reconcile catalog clone. Add `SketchStore::for_each_instance` (lock-held visitor) and rewrite reconcile to derive each sid's signature under the read lock into a reused scratch buffer, collecting only the `u64` orphan sids, then retiring them after the lock drops. No `SketchInstanceMetadata` clone, no per-sid signature `Vec` allocation. 2. Skip reconcile entirely when the config is unchanged. The streaming config is an `Arc>` whose `Arc` identity only changes on a (rare) control-plane swap. New `reconcile_if_config_changed` gates on the `Arc` data pointer via an `AtomicUsize` on `SketchStore`, collapsing the steady-state per-batch reconcile to one relaxed atomic load. Measured (criterion `reconcile_per_batch`, full un-gated scan): 100 sids: 35.2 us -> 12.6 us (2.8x) 1000 sids: 388 us -> 134 us (2.9x) 10000 sids: 3.99 ms -> 1.45 ms (2.75x) The pointer gate additionally takes the steady-state per-batch cost to ~one atomic load (reconcile runs only on config swaps). Reconcile semantics are unchanged: it only ever transitions Active->Retired on signature mismatch against the config (no time-based expiry — that stays in the eviction service), so an unchanged config produces identical results regardless of wall clock. All 9 reconcile lib tests pass (incl. 2 new gate tests + the HTTP config-swap path test); 228 sketch_db lib tests + edge-runtime adapter tests green. Co-Authored-By: Claude Opus 4.7 (1M context) --- data_plane/benches/sketch_db.rs | 67 +++++++ data_plane/src/drivers/ingest/otel.rs | 12 +- .../storage_engines/sketch_db/index/mod.rs | 135 ++++++++++---- .../sketch_db/lifecycle/mod.rs | 4 +- .../sketch_db/lifecycle/reconcile.rs | 172 +++++++++++++++--- 5 files changed, 319 insertions(+), 71 deletions(-) diff --git a/data_plane/benches/sketch_db.rs b/data_plane/benches/sketch_db.rs index b6d6c1874..e5064143b 100644 --- a/data_plane/benches/sketch_db.rs +++ b/data_plane/benches/sketch_db.rs @@ -387,11 +387,78 @@ fn bench_query_precomputes_by_agg(c: &mut Criterion) { g.finish(); } +/// Build a `StreamingConfig` whose single agg-config's content +/// signature matches every sid registered by `build_precompute_store` +/// (metric / `Sum` / no grouping / empty params+filter). With this +/// config the reconciler retires nothing — the steady-state ingest +/// case, where the per-batch reconcile is pure scan overhead. +fn matching_streaming_config(metric: &str) -> data_plane::storage_engines::types::StreamingConfig { + use asap_types::aggregation_config::AggregationConfig; + use asap_types::enums::{AggregationType as AT, WindowType}; + use promql_utilities::data_model::key_by_label_names::KeyByLabelNames; + use std::collections::HashMap; + + let cfg = AggregationConfig::new( + AT::Sum, + String::new(), + HashMap::new(), + KeyByLabelNames::empty(), + KeyByLabelNames::empty(), + KeyByLabelNames::empty(), + String::new(), + 60, + 60, + WindowType::Tumbling, + String::new(), + metric.to_string(), + None, + None, + None, + ); + let mut map = HashMap::new(); + map.insert(1u64, cfg); + data_plane::storage_engines::types::StreamingConfig::new(map) +} + +/// `reconcile_from_streaming_config` ran on EVERY ingest batch and, in +/// the pre-optimization code, deep-cloned every `SketchInstanceMetadata` +/// in the catalog (`snapshot_instances()`) — the dominant ingest-path +/// CPU cost in live `perf` profiling (BTreeMap/String clone + malloc +/// churn). This bench measures one un-gated reconcile against a +/// populated store at a few catalog sizes, so the before/after clone +/// elimination is directly visible. +fn bench_reconcile_per_batch(c: &mut Criterion) { + use std::time::Duration; + + let mut g = c.benchmark_group("reconcile_per_batch"); + g.sample_size(50); + + let metric = "bench_metric"; + for num_sids in [100usize, 1_000, 10_000] { + let store = build_precompute_store(num_sids, 1, metric); + let config = matching_streaming_config(metric); + g.throughput(Throughput::Elements(num_sids as u64)); + g.bench_function(BenchmarkId::new("full_scan", num_sids), |b| { + b.iter(|| { + let summary = + data_plane::storage_engines::sketch_db::lifecycle::reconcile_from_streaming_config( + black_box(&store), + black_box(&config), + Duration::from_secs(60), + ); + black_box(summary); + }); + }); + } + g.finish(); +} + criterion_group!( benches, bench_append_sample, bench_append_precompute, bench_query_range, bench_query_precomputes_by_agg, + bench_reconcile_per_batch, ); criterion_main!(benches); diff --git a/data_plane/src/drivers/ingest/otel.rs b/data_plane/src/drivers/ingest/otel.rs index 2f54d59b2..b70af015e 100644 --- a/data_plane/src/drivers/ingest/otel.rs +++ b/data_plane/src/drivers/ingest/otel.rs @@ -633,7 +633,12 @@ async fn route_otlp_to_precompute( // newly-retired sids transition immediately. The §6.3 ingest // barrier is enforced at the sid level inside // `SketchStore::ingest_precompute_for_agg_config`. - let _ = crate::storage_engines::sketch_db::lifecycle::reconcile_from_streaming_config( + // + // Gated on the config `Arc` identity: the streaming config only + // changes on a (rare) control-plane swap, so in steady state this + // collapses to a single relaxed atomic load and skips the full + // catalog scan — the dominant ingest-path CPU cost in profiling. + let _ = crate::storage_engines::sketch_db::lifecycle::reconcile_if_config_changed( ingest_state.sketch_index.as_ref(), &snap, crate::storage_engines::sketch_db::DEFAULT_RETIREMENT_RETENTION, @@ -860,8 +865,9 @@ async fn route_modified_otlp_sketches_to_precompute( let agg_configs = snap.get_all_aggregation_configs(); // Schema retirement #5 — agg_id-keyed registry retired; sid-level // reconcile is the only path going forward. See the raw-OTLP - // routine above for the full rationale. - let _ = crate::storage_engines::sketch_db::lifecycle::reconcile_from_streaming_config( + // routine above for the full rationale (incl. the config-Arc gate + // that skips the catalog scan when the config is unchanged). + let _ = crate::storage_engines::sketch_db::lifecycle::reconcile_if_config_changed( ingest_state.sketch_index.as_ref(), &snap, crate::storage_engines::sketch_db::DEFAULT_RETIREMENT_RETENTION, diff --git a/data_plane/src/storage_engines/sketch_db/index/mod.rs b/data_plane/src/storage_engines/sketch_db/index/mod.rs index bf35daf1b..da439930d 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -226,6 +226,16 @@ pub struct SketchStore { /// sids without a fingerprint reach those sids through /// `instances_matching(metric, gbk)`. policy_to_sids: RwLock>>, + /// Pointer (as `usize`) of the `Arc` this store + /// last reconciled against. `reconcile_from_streaming_config` runs + /// on every ingest batch, but the config is a lock-free + /// `Arc>` that only changes its `Arc` + /// identity on a control-plane swap (rare). Gating the full + /// catalog scan on a cheap pointer compare against this field lets + /// the steady-state ingest path skip reconcile entirely. + /// `0` (the `Default`) means "never reconciled" so the first batch + /// always runs. A real `Arc` data pointer is never null. + last_reconciled_config_ptr: std::sync::atomic::AtomicUsize, } /// Three possible outcomes of looking up a sid in the SketchStore. @@ -491,8 +501,7 @@ impl SketchStore { by_label_id .into_iter() .map(|(label_id, samples)| { - let label_values_map = - guard.intern.resolve(label_id).cloned().unwrap_or_default(); + let label_values_map = guard.intern.resolve(label_id).cloned().unwrap_or_default(); (label_values_map, samples) }) .collect() @@ -583,11 +592,17 @@ impl SketchStore { end_unix_ms: u64, ) -> std::collections::HashMap< Option, - Vec<((u64, u64), Arc)>, + Vec<( + (u64, u64), + Arc, + )>, > { let mut out: std::collections::HashMap< Option, - Vec<((u64, u64), Arc)>, + Vec<( + (u64, u64), + Arc, + )>, > = std::collections::HashMap::new(); // Pick the sids whose metadata describes this (metric, @@ -649,11 +664,8 @@ impl SketchStore { .range_query_into(start_unix_ms, end_unix_ms, &mut buf); for (win, label_id, payload) in &buf { if let Some(p) = payload.as_exact_agg() { - let label_values_map = guard - .intern - .resolve(*label_id) - .cloned() - .unwrap_or_default(); + let label_values_map = + guard.intern.resolve(*label_id).cloned().unwrap_or_default(); let key = if label_values_map.is_empty() { None } else { @@ -661,10 +673,9 @@ impl SketchStore { labels: label_values_map.values().cloned().collect(), }) }; - out.entry(key).or_default().push(( - *win, - Arc::from(p.clone_boxed_core()), - )); + out.entry(key) + .or_default() + .push((*win, Arc::from(p.clone_boxed_core()))); } } buf.clear(); @@ -673,11 +684,8 @@ impl SketchStore { sealed.range_query_into(start_unix_ms, end_unix_ms, &mut buf); for (win, label_id, payload) in &buf { if let Some(p) = payload.as_exact_agg() { - let label_values_map = guard - .intern - .resolve(*label_id) - .cloned() - .unwrap_or_default(); + let label_values_map = + guard.intern.resolve(*label_id).cloned().unwrap_or_default(); let key = if label_values_map.is_empty() { None } else { @@ -685,10 +693,9 @@ impl SketchStore { labels: label_values_map.values().cloned().collect(), }) }; - out.entry(key).or_default().push(( - *win, - Arc::from(p.clone_boxed_core()), - )); + out.entry(key) + .or_default() + .push((*win, Arc::from(p.clone_boxed_core()))); } } buf.clear(); @@ -763,6 +770,48 @@ impl SketchStore { .unwrap_or(false) } + /// Record that the store has reconciled against the + /// `Arc` identified by `config_ptr` (the value of + /// `Arc::as_ptr(..) as usize`), returning `true` if this is a *new* + /// config pointer (i.e. the caller should run a full reconcile) or + /// `false` if the store already reconciled against this exact + /// config and the scan can be skipped. + /// + /// Used by [`crate::storage_engines::sketch_db::lifecycle::reconcile_if_config_changed`] + /// to make the per-ingest-batch reconcile a single relaxed atomic + /// load in the common (config-unchanged) case. + pub fn mark_reconciled_config(&self, config_ptr: usize) -> bool { + use std::sync::atomic::Ordering; + if self.last_reconciled_config_ptr.load(Ordering::Relaxed) == config_ptr { + return false; + } + self.last_reconciled_config_ptr + .store(config_ptr, Ordering::Relaxed); + true + } + + /// Visit every registered instance under a single read lock, + /// invoking `f(sid, &meta)` for each. Lets read-side scans that + /// only need to *inspect* metadata (signature derivation, + /// status filtering) avoid the O(N) deep clone that + /// [`Self::snapshot_instances`] performs — each + /// `SketchInstanceMetadata` carries a `String` + `BTreeSet` + /// + `AggKind` (more strings), so the clone is allocation-heavy at + /// production catalog sizes. + /// + /// The closure runs while the read lock is held, so it must not + /// call back into the store (which would deadlock) and should stay + /// allocation-light. Callers that need to mutate or call user code + /// should collect the cheap data they need (e.g. `Vec` of + /// sids) here, then act after this returns. + pub fn for_each_instance(&self, mut f: F) { + if let Ok(map) = self.instances.read() { + for (sid, meta) in map.iter() { + f(*sid, meta); + } + } + } + /// Iterate (clones) all instance metadata matching `status`. /// Used by the eviction service to enumerate `Expired` sids /// without holding a long read lock. @@ -771,7 +820,10 @@ impl SketchStore { Ok(m) => m, Err(_) => return Vec::new(), }; - map.values().filter(|s| s.status() == status).cloned().collect() + map.values() + .filter(|s| s.status() == status) + .cloned() + .collect() } /// Force `sid` into `Retired` status, scheduling expiry @@ -955,7 +1007,12 @@ impl SketchStore { } let window = (output.start_timestamp, output.end_timestamp); - self.append_precompute(sid, label_values_map, window, accumulator.clone_boxed_core()); + self.append_precompute( + sid, + label_values_map, + window, + accumulator.clone_boxed_core(), + ); Some(sid) } @@ -1057,8 +1114,9 @@ impl SketchStore { ); let manifest = Arc::new(Manifest::open_or_init(&cfg.disk_path)?); - let parts_root = - crate::storage_engines::sketch_db::index::persistence::flusher::parts_root(&cfg.disk_path); + let parts_root = crate::storage_engines::sketch_db::index::persistence::flusher::parts_root( + &cfg.disk_path, + ); let part_cache = PartCache::new(parts_root.clone(), cfg.part_cache_bytes); let flusher = FlusherHandle::start(cfg, Arc::clone(&manifest), Arc::clone(self))?; @@ -1225,7 +1283,10 @@ mod tests { meta_with_policy(sid, asap_types::PolicyFingerprint::UNSET) } - fn meta_with_policy(sid: u64, policy_fp: asap_types::PolicyFingerprint) -> SketchInstanceMetadata { + fn meta_with_policy( + sid: u64, + policy_fp: asap_types::PolicyFingerprint, + ) -> SketchInstanceMetadata { let cfg = SketchConfig::DDSketch { relative_accuracy: 0.01, }; @@ -1544,12 +1605,7 @@ mod tests { Box::new(SumAccumulator::with_sum(2.0)), ); - let result = idx.query_precomputes_by_agg( - "cpu_seconds", - AggregationType::Sum, - 0, - 10_000, - ); + let result = idx.query_precomputes_by_agg("cpu_seconds", AggregationType::Sum, 0, 10_000); assert_eq!(result.len(), 1, "one label-values key"); let buckets = result.values().next().expect("populated"); assert_eq!(buckets.len(), 2, "two windows for that key"); @@ -1563,12 +1619,7 @@ mod tests { idx.register(meta(7)); idx.append_sample(7, BTreeMap::new(), (1000, 2000), sample(1)); - let result = idx.query_precomputes_by_agg( - "m", - AggregationType::Sum, - 0, - 10_000, - ); + let result = idx.query_precomputes_by_agg("m", AggregationType::Sum, 0, 10_000); assert!(result.is_empty(), "sketch sids must not surface"); } @@ -1626,7 +1677,11 @@ mod tests { "sketch_type_name should reflect sid metadata's sketch_kind: {}", entry.sketch_type_name ); - assert_eq!(entry.sketch_bytes.len(), 1, "single-byte sample bytes carry"); + assert_eq!( + entry.sketch_bytes.len(), + 1, + "single-byte sample bytes carry" + ); } #[test] diff --git a/data_plane/src/storage_engines/sketch_db/lifecycle/mod.rs b/data_plane/src/storage_engines/sketch_db/lifecycle/mod.rs index 6bc69d26f..292ebefc9 100644 --- a/data_plane/src/storage_engines/sketch_db/lifecycle/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/lifecycle/mod.rs @@ -27,5 +27,7 @@ pub mod status; pub use eviction::{ warn_if_retention_inverted, SchemaEvictionConfig, SchemaEvictionHandle, SchemaEvictionService, }; -pub use reconcile::{reconcile_from_streaming_config, SidReconcileSummary}; +pub use reconcile::{ + reconcile_from_streaming_config, reconcile_if_config_changed, SidReconcileSummary, +}; pub use status::{AggStatus, DEFAULT_RETIREMENT_RETENTION}; diff --git a/data_plane/src/storage_engines/sketch_db/lifecycle/reconcile.rs b/data_plane/src/storage_engines/sketch_db/lifecycle/reconcile.rs index 7de209bf3..2e363fb20 100644 --- a/data_plane/src/storage_engines/sketch_db/lifecycle/reconcile.rs +++ b/data_plane/src/storage_engines/sketch_db/lifecycle/reconcile.rs @@ -32,13 +32,14 @@ //! until M3 unifies the two. use std::collections::{BTreeSet, HashSet}; +use std::sync::Arc; use std::time::Duration; use asap_types::aggregation_config::AggregationConfig; use asap_types::streaming_config::StreamingConfig; use crate::storage_engines::sketch_db::data::{canonical_parameters, AggKind}; -use crate::storage_engines::sketch_db::index::{SketchInstanceMetadata, SketchStore}; +use crate::storage_engines::sketch_db::index::SketchStore; use crate::storage_engines::sketch_db::lifecycle::AggStatus; /// Sids the reconciler force-retired this call. @@ -57,6 +58,34 @@ pub struct SidReconcileSummary { /// [`SketchStore::force_retire`] so the schedule of eventual /// `Retired → Expired` matches the pre-retirement `SchemaRegistry` /// behavior. +/// Ingest-path entry point: reconcile only when `config` is a config +/// the store has not yet reconciled against. The streaming config is a +/// lock-free `Arc>` whose `Arc` identity only +/// changes on a (rare) control-plane swap, so in steady state every +/// ingest batch hands us the *same* `Arc`. Gating on the `Arc` data +/// pointer collapses the per-batch reconcile to a single relaxed atomic +/// load in that common case, skipping the full catalog scan + any +/// signature derivation. +/// +/// Pass the same `Arc` the ingest batch snapshotted so +/// the pointer is stable; the snapshot is held alive for the call's +/// duration, so the pointer cannot be reused by a concurrently-dropped +/// config (no ABA hazard within a batch). +/// +/// Returns `None` when the scan was skipped (config unchanged), or +/// `Some(summary)` with the retired sids when a reconcile actually ran. +pub fn reconcile_if_config_changed( + store: &SketchStore, + config: &Arc, + retention: Duration, +) -> Option { + let config_ptr = Arc::as_ptr(config) as usize; + if !store.mark_reconciled_config(config_ptr) { + return None; + } + Some(reconcile_from_streaming_config(store, config, retention)) +} + pub fn reconcile_from_streaming_config( store: &SketchStore, config: &StreamingConfig, @@ -64,16 +93,42 @@ pub fn reconcile_from_streaming_config( ) -> SidReconcileSummary { let live_signatures = build_live_signature_set(config); - let mut retired = Vec::new(); - for meta in store.snapshot_instances() { + // First pass: find orphaned Active sids without cloning the + // catalog. `reconcile_from_streaming_config` runs on every ingest + // batch, and the old `snapshot_instances()` here deep-cloned every + // `SketchInstanceMetadata` (String + BTreeSet + AggKind) + // on each call — the dominant ingest-path CPU cost in profiling + // (BTreeMap/String clone + malloc churn). We only need to read each + // instance's signature under the read lock; collect just the cheap + // `u64` sids to retire, then act after the lock drops. + // + // Reuse one scratch buffer across instances so signature + // derivation doesn't allocate a fresh `Vec` per sid. + let mut orphans: Vec = Vec::new(); + let mut scratch: Vec = Vec::new(); + store.for_each_instance(|sid, meta| { if !matches!(meta.status(), AggStatus::Active) { - continue; + return; } - let sig = signature_from_meta(&meta); - if !live_signatures.contains(&sig) { - if store.force_retire(meta.sid, retention).is_some() { - retired.push(meta.sid); - } + scratch.clear(); + signature_into( + &meta.metric_name, + &meta.agg_kind, + &meta.group_by_keys, + &mut scratch, + ); + if !live_signatures.contains(scratch.as_slice()) { + orphans.push(sid); + } + }); + + // Second pass: retire the orphans (takes the write lock per sid). + // Common case is zero orphans — steady-state ingest matches every + // live signature — so this loop is usually empty. + let mut retired = Vec::new(); + for sid in orphans { + if store.force_retire(sid, retention).is_some() { + retired.push(sid); } } SidReconcileSummary { retired } @@ -83,21 +138,33 @@ pub fn reconcile_from_streaming_config( /// HashSet key. Two sids of the same signature produce identical /// bytes; cross-signature collisions require a 64-bit hash birthday /// — we don't compress here, we compare the full bytes. -fn signature_bytes(metric_name: &str, agg_kind: &AggKind, group_by_keys: &BTreeSet) -> Vec { +fn signature_bytes( + metric_name: &str, + agg_kind: &AggKind, + group_by_keys: &BTreeSet, +) -> Vec { let mut buf: Vec = Vec::new(); + signature_into(metric_name, agg_kind, group_by_keys, &mut buf); + buf +} + +/// Append the canonical signature encoding into `buf` (which the +/// caller may have pre-cleared and reuse across instances to avoid +/// per-sid allocation). +fn signature_into( + metric_name: &str, + agg_kind: &AggKind, + group_by_keys: &BTreeSet, + buf: &mut Vec, +) { buf.extend_from_slice(metric_name.as_bytes()); buf.push(0); - encode_agg_kind(agg_kind, &mut buf); + encode_agg_kind(agg_kind, buf); buf.push(0); for k in group_by_keys { buf.extend_from_slice(k.as_bytes()); buf.push(b','); } - buf -} - -fn signature_from_meta(meta: &SketchInstanceMetadata) -> Vec { - signature_bytes(&meta.metric_name, &meta.agg_kind, &meta.group_by_keys) } fn signature_from_agg_config(cfg: &AggregationConfig) -> Vec { @@ -191,7 +258,11 @@ mod tests { use crate::storage_engines::sketch_db::data::AggKind; use crate::storage_engines::sketch_db::index::{SketchInstanceMetadata, SketchStore}; - fn agg_config(metric: &str, agg_type: AggregationType, group_by: Vec<&str>) -> AggregationConfig { + fn agg_config( + metric: &str, + agg_type: AggregationType, + group_by: Vec<&str>, + ) -> AggregationConfig { AggregationConfig::new( agg_type, String::new(), @@ -217,8 +288,7 @@ mod tests { agg_type: AggregationType, group_by: Vec<&str>, ) -> SketchInstanceMetadata { - let group_by_keys: BTreeSet = - group_by.into_iter().map(|s| s.to_string()).collect(); + let group_by_keys: BTreeSet = group_by.into_iter().map(|s| s.to_string()).collect(); SketchInstanceMetadata { sid, metric_name: metric.to_string(), @@ -271,7 +341,11 @@ mod tests { let store = SketchStore::new(); // sid 1 is Sum-by-host; only Sum-by-region is in the new config. store.register(meta(1, "cpu", AggregationType::Sum, vec!["host"])); - let cfg = streaming(vec![agg_config("cpu", AggregationType::Sum, vec!["region"])]); + let cfg = streaming(vec![agg_config( + "cpu", + AggregationType::Sum, + vec!["region"], + )]); let summary = reconcile_from_streaming_config(&store, &cfg, Duration::from_secs(60)); assert_eq!(summary.retired, vec![1]); } @@ -281,13 +355,11 @@ mod tests { let store = SketchStore::new(); store.register(meta(1, "cpu", AggregationType::Sum, vec!["host"])); // First reconcile with empty config retires it. - let _ = reconcile_from_streaming_config(&store, &streaming(vec![]), Duration::from_secs(60)); + let _ = + reconcile_from_streaming_config(&store, &streaming(vec![]), Duration::from_secs(60)); // Second reconcile reports nothing new. - let summary = reconcile_from_streaming_config( - &store, - &streaming(vec![]), - Duration::from_secs(60), - ); + let summary = + reconcile_from_streaming_config(&store, &streaming(vec![]), Duration::from_secs(60)); assert!(summary.retired.is_empty()); } @@ -306,8 +378,54 @@ mod tests { fn different_agg_type_does_not_match() { let store = SketchStore::new(); store.register(meta(1, "cpu", AggregationType::Sum, vec!["host"])); - let cfg = streaming(vec![agg_config("cpu", AggregationType::Increase, vec!["host"])]); + let cfg = streaming(vec![agg_config( + "cpu", + AggregationType::Increase, + vec!["host"], + )]); let summary = reconcile_from_streaming_config(&store, &cfg, Duration::from_secs(60)); assert_eq!(summary.retired, vec![1]); } + + #[test] + fn gate_skips_reconcile_on_unchanged_config_arc() { + let store = SketchStore::new(); + store.register(meta(1, "cpu", AggregationType::Sum, vec!["host"])); + // Empty config would orphan sid 1. + let cfg = Arc::new(streaming(vec![])); + + // First call against this Arc runs a real reconcile and + // retires the orphan. + let first = reconcile_if_config_changed(&store, &cfg, Duration::from_secs(60)); + assert_eq!(first.expect("first call reconciles").retired, vec![1]); + + // Second call with the SAME Arc is skipped entirely — no + // catalog scan, returns None. (Even though a fresh + // unconditional reconcile would report nothing, the point is + // the scan didn't run.) + let second = reconcile_if_config_changed(&store, &cfg, Duration::from_secs(60)); + assert!(second.is_none(), "same config Arc must skip the scan"); + } + + #[test] + fn gate_runs_reconcile_when_config_arc_changes() { + let store = SketchStore::new(); + store.register(meta(1, "cpu", AggregationType::Sum, vec!["host"])); + + // A config that keeps sid 1 alive — first reconcile retires + // nothing and arms the gate. + let keep = Arc::new(streaming(vec![agg_config( + "cpu", + AggregationType::Sum, + vec!["host"], + )])); + let r1 = reconcile_if_config_changed(&store, &keep, Duration::from_secs(60)); + assert!(r1.expect("first reconcile runs").retired.is_empty()); + + // A NEW Arc (different config) must defeat the gate and run a + // fresh reconcile that now retires the orphaned sid. + let drop_all = Arc::new(streaming(vec![])); + let r2 = reconcile_if_config_changed(&store, &drop_all, Duration::from_secs(60)); + assert_eq!(r2.expect("changed config Arc reconciles").retired, vec![1]); + } }