From 4c2629dea0bcd7b22dae86441077a7961dd5b26e Mon Sep 17 00:00:00 2001 From: Zeying Zhu Date: Tue, 16 Jun 2026 10:02:38 -0600 Subject: [PATCH 1/5] fix(precompute): port wall-clock grace fallback from backend to close idle windows MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Sync the precompute-engine window-closing fix from ASAPQuery-backend. Problem: with strict event-time semantics, a tumbling window [T, T+size) only closes when the watermark reaches T+size. For a one-shot batch ingest where every record carries (nearly) the same timestamp, the watermark freezes and flush_all's +1ms advance is a no-op, so the trailing window never closes — emit_batch is never called and the store stays empty even though data was ingested. Fix (ported from the backend's flush_all "sweep blocker #2" fallback): track each pane's wall-clock birth time and, in flush_all, force the effective watermark past pane_start + window_size_ms for any pane that has been alive longer than window_size_ms + wall_clock_grace_period_ms of wall-clock time, regardless of event-time. Set wall_clock_grace_period_ms <= 0 to opt out and keep strict event-time-only semantics. Changes: - config: add wall_clock_grace_period_ms (serde default 5000ms) - worker: GroupState.pane_wall_clock_starts_ms + prune; injectable now_ms_fn (default SystemTime::now, test override via set_now_ms_fn); record pane birth in process_group_samples; wall-clock fallback in flush_all - engine/main/CLI: thread the setting through (PrecomputeSettings + --wall-clock-grace-period-ms flag) - tests: wall_clock_fallback_closes_idle_window and the grace=0 opt-out; update existing config literals Co-Authored-By: Claude Opus 4.8 (1M context) --- .../src/bin/bench_precompute_sketch.rs | 1 + .../src/bin/e2e_quickstart_resource_test.rs | 1 + .../src/bin/precompute_engine.rs | 6 + .../src/bin/test_e2e_precompute.rs | 3 + asap-query-engine/src/engine_config.rs | 7 + asap-query-engine/src/main.rs | 1 + .../src/precompute_engine/config.rs | 19 ++ .../src/precompute_engine/engine.rs | 1 + .../src/precompute_engine/worker.rs | 258 +++++++++++++++++- .../tests/e2e_precompute_equivalence.rs | 3 + 10 files changed, 298 insertions(+), 2 deletions(-) diff --git a/asap-query-engine/src/bin/bench_precompute_sketch.rs b/asap-query-engine/src/bin/bench_precompute_sketch.rs index acc1bb32..f2646956 100644 --- a/asap-query-engine/src/bin/bench_precompute_sketch.rs +++ b/asap-query-engine/src/bin/bench_precompute_sketch.rs @@ -175,6 +175,7 @@ async fn start_engine( pass_raw_samples: false, raw_mode_aggregation_id: 0, late_data_policy: LateDataPolicy::Drop, + wall_clock_grace_period_ms: 5_000, }; let sources: Vec> = vec![Box::new(HttpIngestSource::new(HttpIngestConfig { port }))]; diff --git a/asap-query-engine/src/bin/e2e_quickstart_resource_test.rs b/asap-query-engine/src/bin/e2e_quickstart_resource_test.rs index d926c9ab..cc34b502 100644 --- a/asap-query-engine/src/bin/e2e_quickstart_resource_test.rs +++ b/asap-query-engine/src/bin/e2e_quickstart_resource_test.rs @@ -234,6 +234,7 @@ async fn main() -> Result<(), Box> { pass_raw_samples: false, raw_mode_aggregation_id: 0, late_data_policy: LateDataPolicy::Drop, + wall_clock_grace_period_ms: 5_000, }; let output_sink = Arc::new(StoreOutputSink::new(store.clone())); let sources: Vec> = diff --git a/asap-query-engine/src/bin/precompute_engine.rs b/asap-query-engine/src/bin/precompute_engine.rs index 1a658989..8c479980 100644 --- a/asap-query-engine/src/bin/precompute_engine.rs +++ b/asap-query-engine/src/bin/precompute_engine.rs @@ -67,6 +67,11 @@ struct Args { #[arg(long, value_enum, default_value_t = LateDataPolicy::Drop)] late_data_policy: LateDataPolicy, + /// Wall-clock grace period (ms) for the flush fallback that force-closes + /// idle windows when event-time stagnates. Set to <= 0 to disable. + #[arg(long, default_value_t = 5000)] + wall_clock_grace_period_ms: i64, + // --- CSV file ingest (alternative to HTTP) --- /// Path to a local CSV file to ingest instead of listening for HTTP writes #[arg(long)] @@ -170,6 +175,7 @@ async fn main() -> Result<(), Box> { pass_raw_samples: args.pass_raw_samples, raw_mode_aggregation_id: args.raw_mode_aggregation_id, late_data_policy: args.late_data_policy, + wall_clock_grace_period_ms: args.wall_clock_grace_period_ms, }; // Create the output sink (writes directly to the store) diff --git a/asap-query-engine/src/bin/test_e2e_precompute.rs b/asap-query-engine/src/bin/test_e2e_precompute.rs index 2fc781fc..085aa9c3 100644 --- a/asap-query-engine/src/bin/test_e2e_precompute.rs +++ b/asap-query-engine/src/bin/test_e2e_precompute.rs @@ -151,6 +151,7 @@ async fn main() -> Result<(), Box> { pass_raw_samples: false, raw_mode_aggregation_id: 0, late_data_policy: LateDataPolicy::Drop, + wall_clock_grace_period_ms: 5000, }; let output_sink = Arc::new(StoreOutputSink::new(store.clone())); let sources: Vec> = @@ -299,6 +300,7 @@ async fn main() -> Result<(), Box> { pass_raw_samples: true, raw_mode_aggregation_id: raw_agg_id, late_data_policy: LateDataPolicy::Drop, + wall_clock_grace_period_ms: 5000, }; let raw_sink = Arc::new(RawPassthroughSink::new(store.clone())); let raw_sources: Vec> = @@ -655,6 +657,7 @@ async fn run_single_bench( pass_raw_samples: false, raw_mode_aggregation_id: 0, late_data_policy: LateDataPolicy::Drop, + wall_clock_grace_period_ms: 5000, }; let sources: Vec> = vec![Box::new(HttpIngestSource::new(HttpIngestConfig { port }))]; diff --git a/asap-query-engine/src/engine_config.rs b/asap-query-engine/src/engine_config.rs index a6a1141b..1bd78fbd 100644 --- a/asap-query-engine/src/engine_config.rs +++ b/asap-query-engine/src/engine_config.rs @@ -310,6 +310,12 @@ pub struct PrecomputeSettings { pub flush_interval_ms: u64, pub channel_buffer_size: usize, pub dump_precomputes: bool, + /// Wall-clock grace period (ms) for the flush fallback that force-closes + /// idle windows when event-time stagnates (e.g. one-shot batch ingest + /// where every record shares a timestamp). Set to <= 0 to disable and + /// keep strict event-time-only semantics. See + /// `PrecomputeEngineConfig::wall_clock_grace_period_ms`. + pub wall_clock_grace_period_ms: i64, } impl Default for PrecomputeSettings { @@ -321,6 +327,7 @@ impl Default for PrecomputeSettings { flush_interval_ms: 1000, channel_buffer_size: 10000, dump_precomputes: false, + wall_clock_grace_period_ms: 5000, } } } diff --git a/asap-query-engine/src/main.rs b/asap-query-engine/src/main.rs index 243cd355..7581ab23 100644 --- a/asap-query-engine/src/main.rs +++ b/asap-query-engine/src/main.rs @@ -209,6 +209,7 @@ async fn main() -> Result<()> { pass_raw_samples: false, raw_mode_aggregation_id: 0, late_data_policy: LateDataPolicy::Drop, + wall_clock_grace_period_ms: config.precompute_engine.wall_clock_grace_period_ms, }; let output_sink = Arc::new(StoreOutputSink::new(store.clone())); let sources: Vec> = match &config.ingest { diff --git a/asap-query-engine/src/precompute_engine/config.rs b/asap-query-engine/src/precompute_engine/config.rs index 656f8edb..129621d4 100644 --- a/asap-query-engine/src/precompute_engine/config.rs +++ b/asap-query-engine/src/precompute_engine/config.rs @@ -30,6 +30,19 @@ pub struct PrecomputeEngineConfig { pub raw_mode_aggregation_id: u64, /// Policy for handling late samples that arrive after their window has closed. pub late_data_policy: LateDataPolicy, + /// Wall-clock grace period (milliseconds) for the watermark fallback in + /// `flush_all`. When event-time stagnates (e.g. a one-shot batch where + /// every record carries the same timestamp), `flush_all`'s `+1ms` + /// watermark advance is a no-op and idle windows never close. The + /// wall-clock fallback closes a pane whose creation has been older + /// than `window_size_ms + wall_clock_grace_period_ms` of *wall-clock* + /// time, regardless of where event-time is. The grace period tolerates + /// late-arriving events that would otherwise be evicted as "the window + /// already closed". Set to `<= 0` to opt out and keep strict + /// event-time-only semantics. Default: 5000 ms (matches + /// `allowed_lateness_ms` default). + #[serde(default = "default_wall_clock_grace_period_ms")] + pub wall_clock_grace_period_ms: i64, } impl Default for PrecomputeEngineConfig { @@ -43,10 +56,15 @@ impl Default for PrecomputeEngineConfig { pass_raw_samples: false, raw_mode_aggregation_id: 0, late_data_policy: LateDataPolicy::Drop, + wall_clock_grace_period_ms: default_wall_clock_grace_period_ms(), } } } +fn default_wall_clock_grace_period_ms() -> i64 { + 5_000 +} + #[cfg(test)] mod tests { use super::*; @@ -62,5 +80,6 @@ mod tests { assert!(!config.pass_raw_samples); assert_eq!(config.raw_mode_aggregation_id, 0); assert_eq!(config.late_data_policy, LateDataPolicy::Drop); + assert_eq!(config.wall_clock_grace_period_ms, 5_000); } } diff --git a/asap-query-engine/src/precompute_engine/engine.rs b/asap-query-engine/src/precompute_engine/engine.rs index 42fae769..ee2e7720 100644 --- a/asap-query-engine/src/precompute_engine/engine.rs +++ b/asap-query-engine/src/precompute_engine/engine.rs @@ -171,6 +171,7 @@ impl PrecomputeEngine { pass_raw_samples: self.config.pass_raw_samples, raw_mode_aggregation_id: self.config.raw_mode_aggregation_id, late_data_policy: self.config.late_data_policy, + wall_clock_grace_period_ms: self.config.wall_clock_grace_period_ms, }, self.diagnostics.worker_group_counts[id].clone(), self.diagnostics.worker_watermarks[id].clone(), diff --git a/asap-query-engine/src/precompute_engine/worker.rs b/asap-query-engine/src/precompute_engine/worker.rs index 02207994..bb088d02 100644 --- a/asap-query-engine/src/precompute_engine/worker.rs +++ b/asap-query-engine/src/precompute_engine/worker.rs @@ -28,6 +28,25 @@ struct GroupState { /// Per-group watermark: tracks the maximum timestamp seen across all /// series in this group on this worker. previous_watermark_ms: i64, + /// Wall-clock-time (ms since epoch) at which each currently-open pane + /// was first opened. Used by `flush_all`'s wall-clock fallback to + /// close panes that have been alive too long when event-time has + /// stagnated. Keyed by `pane_start_ms`, mirroring `active_panes`. + /// Entries are GC'd by `prune_pane_wall_clock_starts` after each + /// window-close cycle so the bookkeeping doesn't leak as panes turn over. + pane_wall_clock_starts_ms: BTreeMap, +} + +impl GroupState { + /// Drop wall-clock-start entries whose pane no longer exists in + /// `active_panes`. Called after window-close cycles in + /// `process_group_samples` and `flush_all` so the bookkeeping doesn't + /// leak as panes turn over. + fn prune_pane_wall_clock_starts(&mut self) { + let active = &self.active_panes; + self.pane_wall_clock_starts_ms + .retain(|ps, _| active.contains_key(ps)); + } } /// Runtime configuration for a Worker, grouping non-structural parameters. @@ -37,6 +56,10 @@ pub struct WorkerRuntimeConfig { pub pass_raw_samples: bool, pub raw_mode_aggregation_id: u64, pub late_data_policy: LateDataPolicy, + /// See `PrecomputeEngineConfig::wall_clock_grace_period_ms`. Set to a + /// non-positive value to disable the wall-clock fallback entirely + /// (event-time-only behaviour, matching pre-fix semantics). + pub wall_clock_grace_period_ms: i64, } /// Worker that processes samples for a shard of the group space. @@ -72,6 +95,13 @@ pub struct Worker { all_worker_watermarks: Vec>, /// Externally-readable group count for diagnostics. group_count: Arc, + /// Grace period (ms) for the wall-clock fallback in `flush_all`. + /// `<= 0` disables the fallback (event-time-only). + wall_clock_grace_period_ms: i64, + /// Injectable clock returning current wall-clock time in milliseconds + /// since the unix epoch. Production uses `SystemTime::now`; tests + /// override with a deterministic fake via `set_now_ms_fn`. + now_ms_fn: Box i64 + Send + Sync>, } impl Worker { @@ -92,6 +122,7 @@ impl Worker { pass_raw_samples, raw_mode_aggregation_id, late_data_policy, + wall_clock_grace_period_ms, } = runtime_config; Self { id, @@ -106,9 +137,20 @@ impl Worker { worker_watermark, all_worker_watermarks, group_count, + wall_clock_grace_period_ms, + now_ms_fn: Box::new(default_now_ms), } } + /// Test/diagnostic-only setter for the wall-clock source. Replaces the + /// default `SystemTime::now`-backed clock with a deterministic fake so + /// unit tests can drive the wall-clock fallback in `flush_all` without + /// `std::thread::sleep`. Production code never calls this. + #[cfg(test)] + pub fn set_now_ms_fn(&mut self, f: Box i64 + Send + Sync>) { + self.now_ms_fn = f; + } + /// Run the worker loop. Blocks until shutdown. pub async fn run(mut self) { info!("Worker {} started", self.id); @@ -290,6 +332,7 @@ impl Worker { config, active_panes: BTreeMap::new(), previous_watermark_ms: i64::MIN, + pane_wall_clock_starts_ms: BTreeMap::new(), }; self.group_states .entry(agg_id) @@ -314,6 +357,9 @@ impl Worker { let worker_id = self.id; let allowed_lateness_ms = self.allowed_lateness_ms; let late_data_policy = self.late_data_policy; + // Sample the wall clock before borrowing `state` (the closure lives + // on `self`); used to stamp each pane's birth time below. + let now_ms = (self.now_ms_fn)(); let state = match self.get_or_create_group_state(agg_id, group_key) { Some(state) => state, @@ -389,7 +435,14 @@ impl Worker { } } - // Normal path: route sample to its single pane accumulator + // Normal path: route sample to its single pane accumulator. + // Record the pane's wall-clock birth time the first time we + // touch it, so the wall-clock fallback in `flush_all` can age + // it out even if event-time freezes. + state + .pane_wall_clock_starts_ms + .entry(pane_start) + .or_insert(now_ms); let updater = state .active_panes .entry(pane_start) @@ -419,6 +472,7 @@ impl Worker { } state.previous_watermark_ms = current_wm; + state.prune_pane_wall_clock_starts(); // Emit to output sink if !emit_batch.is_empty() { @@ -491,6 +545,11 @@ impl Worker { // Step 3: Compute global watermark = min(all worker watermarks). let global_wm = self.compute_global_watermark(); + // Sample the wall clock and grace period once for this flush cycle, + // before borrowing `group_states` mutably below. + let now_ms = (self.now_ms_fn)(); + let grace_ms = self.wall_clock_grace_period_ms; + // Step 4: For each group, advance watermark and close due windows. let mut emit_batch: Vec<(PrecomputedOutput, Box)> = Vec::new(); @@ -506,7 +565,29 @@ impl Worker { } else { state.previous_watermark_ms }; - let effective_wm = propagated_wm.saturating_add(1); + let mut effective_wm = propagated_wm.saturating_add(1); + + // Wall-clock fallback for stuck event-time. If every sample + // carries the same timestamp (e.g. a one-shot batch where + // all records fall in the same second), `previous_watermark_ms` + // freezes and `closed_windows(prev, prev+1)` returns empty + // forever — the window never closes and the store stays empty + // even though data has been ingested. Force `effective_wm` + // past `pane_start + window_size_ms` for any pane older than + // `window_size + grace` of WALL-CLOCK time. Set + // `wall_clock_grace_period_ms <= 0` to opt out and keep strict + // event-time semantics. + if grace_ms > 0 { + let window_size_ms = state.window_manager.window_size_ms(); + for (&pane_start, &pane_birth_ms) in &state.pane_wall_clock_starts_ms { + if now_ms.saturating_sub(pane_birth_ms) >= window_size_ms + grace_ms { + let force_to = pane_start.saturating_add(window_size_ms); + if force_to > effective_wm { + effective_wm = force_to; + } + } + } + } let closed = state .window_manager @@ -531,9 +612,14 @@ impl Worker { } // Update group watermark to reflect the advancement. + // Monotonic advance — never retreat. Both the event-time + // boundary `propagated_wm + 1` and the wall-clock fallback + // only push `effective_wm` forward, so this is safe. if effective_wm > state.previous_watermark_ms { state.previous_watermark_ms = effective_wm; } + + state.prune_pane_wall_clock_starts(); } } @@ -569,6 +655,19 @@ impl Worker { } } +/// Default wall-clock-now source: milliseconds since the unix epoch. +/// Used by `Worker::new`. Tests override via `set_now_ms_fn`. +fn default_now_ms() -> i64 { + use std::time::{SystemTime, UNIX_EPOCH}; + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as i64) + // Pre-1970 wall clock (only happens if the host clock is grossly + // misconfigured) — fall back to 0 so the fallback simply doesn't + // trigger rather than panicking. + .unwrap_or(0) +} + /// Build a `KeyByLabelValues` from a semicolon-delimited group key string. /// e.g. "constant" → KeyByLabelValues { labels: ["constant"] } /// e.g. "us-east;svc-a" → KeyByLabelValues { labels: ["us-east", "svc-a"] } @@ -873,6 +972,7 @@ mod tests { pass_raw_samples: pass_raw, raw_mode_aggregation_id: raw_agg_id, late_data_policy: late_policy, + wall_clock_grace_period_ms: 0, }, Arc::new(AtomicUsize::new(0)), wm.clone(), @@ -1668,6 +1768,7 @@ mod tests { pass_raw_samples: false, raw_mode_aggregation_id: 0, late_data_policy: LateDataPolicy::Drop, + wall_clock_grace_period_ms: 0, }, Arc::new(AtomicUsize::new(0)), wm.clone(), @@ -1720,6 +1821,7 @@ mod tests { pass_raw_samples: false, raw_mode_aggregation_id: 0, late_data_policy: LateDataPolicy::ForwardToStore, + wall_clock_grace_period_ms: 0, }, Arc::new(AtomicUsize::new(0)), wm.clone(), @@ -1949,6 +2051,7 @@ aggregations: pass_raw_samples: false, raw_mode_aggregation_id: 0, late_data_policy: LateDataPolicy::Drop, + wall_clock_grace_period_ms: 0, }, Arc::new(AtomicUsize::new(0)), wm0, @@ -1976,6 +2079,7 @@ aggregations: pass_raw_samples: false, raw_mode_aggregation_id: 0, late_data_policy: LateDataPolicy::Drop, + wall_clock_grace_period_ms: 0, }, Arc::new(AtomicUsize::new(0)), wm0, @@ -2007,6 +2111,7 @@ aggregations: pass_raw_samples: false, raw_mode_aggregation_id: 0, late_data_policy: LateDataPolicy::Drop, + wall_clock_grace_period_ms: 0, }, Arc::new(AtomicUsize::new(0)), wm0, @@ -2047,6 +2152,7 @@ aggregations: pass_raw_samples: false, raw_mode_aggregation_id: 0, late_data_policy: LateDataPolicy::Drop, + wall_clock_grace_period_ms: 0, }, Arc::new(AtomicUsize::new(0)), wm.clone(), @@ -2100,6 +2206,7 @@ aggregations: pass_raw_samples: false, raw_mode_aggregation_id: 0, late_data_policy: LateDataPolicy::Drop, + wall_clock_grace_period_ms: 0, }, Arc::new(AtomicUsize::new(0)), wm.clone(), @@ -2170,4 +2277,151 @@ aggregations: sum_acc.sum ); } + + // ----------------------------------------------------------------------- + // Test: wall-clock fallback closes an idle window when event-time freezes + // + // Reproduces the one-shot-batch failure mode: every record falls in the + // same window and no later timestamp ever arrives to advance the + // watermark, so `flush_all`'s `+1ms` event-time advance is a no-op and the + // window never closes — the store stays empty. The wall-clock fallback + // force-closes the pane once it has been alive for `window_size + grace` + // of wall-clock time. Uses an injected fake clock so it runs in + // microseconds instead of sleeping for real seconds. + // ----------------------------------------------------------------------- + + /// Build a Worker identical to the inline test setups but with an explicit + /// `wall_clock_grace_period_ms`. Test-local helper. + fn make_worker_with_grace( + agg_configs: HashMap, + sink: Arc, + wall_clock_grace_period_ms: i64, + ) -> Worker { + let (_tx, rx) = tokio::sync::mpsc::channel(1); + let wm = Arc::new(AtomicI64::new(i64::MIN)); + Worker::new( + 0, + rx, + sink, + arc_configs(agg_configs), + WorkerRuntimeConfig { + max_buffer_per_series: 10_000, + allowed_lateness_ms: 0, + pass_raw_samples: false, + raw_mode_aggregation_id: 0, + late_data_policy: LateDataPolicy::Drop, + wall_clock_grace_period_ms, + }, + Arc::new(AtomicUsize::new(0)), + wm.clone(), + vec![wm], + ) + } + + #[test] + fn wall_clock_fallback_closes_idle_window() { + // 10s tumbling window. + let cfg = make_agg_config( + 7, + "cpu", + AggregationType::SingleSubpopulation, + "Sum", + 10, + 0, + vec![], + ); + let agg_configs = HashMap::from([(7, cfg)]); + let sink = Arc::new(CapturingOutputSink::new()); + // 5s grace period — production default. + let mut worker = make_worker_with_grace(agg_configs, sink.clone(), 5_000); + + // Pin wall-clock at t_wall=1_000_000ms during ingest. Every sample + // carries the SAME frozen event-time (t_event=0), so the watermark + // never advances past the window. + let wall_clock = Arc::new(AtomicI64::new(1_000_000)); + let wc_clone = wall_clock.clone(); + worker.set_now_ms_fn(Box::new(move || wc_clone.load(Ordering::Relaxed))); + + for i in 0..10 { + worker + .process_group_samples(7, "", group_samples("cpu", vec![(0, 1.0 + i as f64)])) + .expect("ingest must accept frozen-event-time samples"); + } + assert_eq!( + sink.len(), + 0, + "no output yet: event-time hasn't advanced past the window" + ); + + // Flush at the same wall-clock time (pane just born). Fallback must + // NOT trigger — pane is younger than window_size + grace = 15s. + worker.flush_all().unwrap(); + assert_eq!( + sink.len(), + 0, + "flush at t_wall=pane_birth must not close the window" + ); + + // Advance wall-clock by exactly window_size + grace = 15s. Now the + // pane is old enough that the fallback must close and emit its window, + // even though event-time is still pinned at 0. + wall_clock.store(1_000_000 + 10_000 + 5_000, Ordering::Relaxed); + worker.flush_all().unwrap(); + + let captured = sink.drain(); + assert_eq!( + captured.len(), + 1, + "wall-clock fallback must close the idle window once wall-clock age \ + exceeds window_size + grace" + ); + let (output, _acc) = &captured[0]; + assert_eq!(output.aggregation_id, 7); + assert_eq!(output.start_timestamp, 0); + assert_eq!(output.end_timestamp, 10_000); + + // Idempotent: a window closed by the fallback drains its pane and its + // wall-clock bookkeeping, so a subsequent flush must not re-emit. + worker.flush_all().unwrap(); + assert_eq!( + sink.len(), + 0, + "already-closed window must not re-emit" + ); + } + + #[test] + fn wall_clock_fallback_disabled_preserves_event_time_only_semantics() { + let cfg = make_agg_config( + 7, + "cpu", + AggregationType::SingleSubpopulation, + "Sum", + 10, + 0, + vec![], + ); + let agg_configs = HashMap::from([(7, cfg)]); + let sink = Arc::new(CapturingOutputSink::new()); + // grace=0 disables the fallback entirely. + let mut worker = make_worker_with_grace(agg_configs, sink.clone(), 0); + + let wall_clock = Arc::new(AtomicI64::new(1_000_000)); + let wc_clone = wall_clock.clone(); + worker.set_now_ms_fn(Box::new(move || wc_clone.load(Ordering::Relaxed))); + + worker + .process_group_samples(7, "", group_samples("cpu", vec![(0, 42.0)])) + .unwrap(); + + // Even after a wall-clock eternity, grace=0 keeps strict event-time + // semantics — the window never closes because event-time is frozen. + wall_clock.store(1_000_000 + 86_400_000, Ordering::Relaxed); // +24h + worker.flush_all().unwrap(); + assert_eq!( + sink.len(), + 0, + "grace=0 must disable the fallback — event-time-only semantics" + ); + } } diff --git a/asap-query-engine/tests/e2e_precompute_equivalence.rs b/asap-query-engine/tests/e2e_precompute_equivalence.rs index 888ef5ba..1c613077 100644 --- a/asap-query-engine/tests/e2e_precompute_equivalence.rs +++ b/asap-query-engine/tests/e2e_precompute_equivalence.rs @@ -153,6 +153,9 @@ fn engine_config() -> PrecomputeEngineConfig { pass_raw_samples: false, raw_mode_aggregation_id: 0, late_data_policy: LateDataPolicy::Drop, + // Strict event-time semantics for the equivalence comparison vs Arroyo + // (disable the wall-clock fallback so timing can't perturb output). + wall_clock_grace_period_ms: 0, } } From 31692e5bb36c70dc0a189bf1a6ad89d3b123a833 Mon Sep 17 00:00:00 2001 From: Zeying Zhu Date: Tue, 16 Jun 2026 10:25:56 -0600 Subject: [PATCH 2/5] fix(precompute): force-close open windows on worker shutdown MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up to the wall-clock grace fallback. The fallback only fires once a pane has aged window_size + grace of wall-clock time on a flush tick, so a one-shot batch that ingests and immediately shuts down still leaves its trailing window open — flush_all's +1ms advance never reaches the window end and the data never lands in the store. Add Worker::force_close_all(), invoked from the Shutdown handler after the final flush_all(): for every group with open panes, advance to a finite bound (max_pane + window_size_ms) and emit every remaining window. A finite bound is used deliberately — WindowManager::closed_windows enumerates window starts one slide at a time, so passing i64::MAX would loop ~i64::MAX/slide times and overflow. The pass is idempotent: drained panes are removed and their wall-clock bookkeeping pruned. - worker: force_close_all() + Shutdown handler wiring - tests: shutdown_force_close_emits_trailing_window; update test_update_agg_configs_enables_new_aggregation_at_runtime to assert the trailing window [10_000, 20_000) is now emitted (previously lost) Co-Authored-By: Claude Opus 4.8 (1M context) --- .../src/precompute_engine/worker.rs | 188 +++++++++++++++++- 1 file changed, 184 insertions(+), 4 deletions(-) diff --git a/asap-query-engine/src/precompute_engine/worker.rs b/asap-query-engine/src/precompute_engine/worker.rs index bb088d02..e38c08f5 100644 --- a/asap-query-engine/src/precompute_engine/worker.rs +++ b/asap-query-engine/src/precompute_engine/worker.rs @@ -213,6 +213,15 @@ impl Worker { if let Err(e) = self.flush_all() { warn!("Worker {} final flush error: {}", self.id, e); } + // Force-close any windows still open after the final flush. + // `flush_all` only advances the watermark by +1ms (plus the + // wall-clock fallback, whose grace may not have elapsed for a + // one-shot batch), so the trailing window can remain open and + // its data would never reach the store. No more samples will + // arrive after shutdown, so close every remaining pane. + if let Err(e) = self.force_close_all() { + warn!("Worker {} shutdown force-close error: {}", self.id, e); + } break; } WorkerMessage::UpdateAggConfigs(new_configs) => { @@ -635,6 +644,85 @@ impl Worker { Ok(()) } + /// Force-close every window still open on shutdown. + /// + /// Unlike `flush_all` — which only advances the watermark by `+1ms` (plus + /// the wall-clock fallback, gated on grace having elapsed) — this emits the + /// window for every remaining pane unconditionally, because no further + /// samples will arrive once the engine is shutting down. Without it, a + /// one-shot batch whose records all fall in a single window (so event-time + /// never advances past the window end) would leave that window open forever + /// and never write it to the store. + /// + /// To advance past the open windows we use a *finite* bound derived from + /// the largest open pane (`max_pane + window_size_ms`) rather than + /// `i64::MAX`: `WindowManager::closed_windows` enumerates window starts up + /// to `current_wm` one slide at a time, so passing `i64::MAX` would loop + /// ~`i64::MAX / slide` times and overflow. `max_pane + window_size_ms` is + /// the smallest watermark that closes the latest open window. + /// + /// Idempotent: closed panes are drained from `active_panes` and their + /// wall-clock bookkeeping is pruned, so a second call emits nothing. + fn force_close_all(&mut self) -> Result<(), Box> { + if self.pass_raw_samples { + return Ok(()); + } + + let mut emit_batch: Vec<(PrecomputedOutput, Box)> = Vec::new(); + + for (agg_id, inner) in &mut self.group_states { + for (group_key, state) in inner.iter_mut() { + if state.previous_watermark_ms == i64::MIN { + continue; // never received data — nothing to close + } + // The latest window start equals the largest open pane start; + // closing window `[start, start + size)` needs `wm >= start + size`. + let Some(&max_pane) = state.active_panes.keys().next_back() else { + continue; // no open panes + }; + let force_wm = max_pane.saturating_add(state.window_manager.window_size_ms()); + + let closed = state + .window_manager + .closed_windows(state.previous_watermark_ms, force_wm); + + for window_start in &closed { + let (_, window_end) = state.window_manager.window_bounds(*window_start); + let pane_starts = state.window_manager.panes_for_window(*window_start); + + if let Some(accumulator) = + merge_panes_for_window(&mut state.active_panes, &pane_starts) + { + let key = build_group_key_label_values(group_key); + let output = PrecomputedOutput::new( + *window_start as u64, + window_end as u64, + Some(key), + *agg_id, + ); + emit_batch.push((output, accumulator)); + } + } + + if force_wm > state.previous_watermark_ms { + state.previous_watermark_ms = force_wm; + } + state.prune_pane_wall_clock_starts(); + } + } + + if !emit_batch.is_empty() { + debug!( + "Worker {} shutdown force-close emitting {} outputs", + self.id, + emit_batch.len() + ); + self.output_sink.emit_batch(emit_batch)?; + } + + Ok(()) + } + /// Compute the global watermark as min(all worker watermarks), ignoring /// workers that haven't started yet (still at i64::MIN). fn compute_global_watermark(&self) -> i64 { @@ -2253,11 +2341,18 @@ aggregations: tx.send(WorkerMessage::Shutdown).await.unwrap(); handle.await.unwrap(); - let captured = sink.drain(); + let mut captured = sink.drain(); + // Two windows close: + // 1. [0, 10_000) — closed inline when the t=10_000 sample advanced + // the watermark; contains only the post-update t=5_000 sample. + // 2. [10_000, 20_000) — left open by the watermark but force-closed on + // shutdown; contains the t=10_000 sample. Before the shutdown + // force-close this trailing window was silently lost. + captured.sort_by_key(|(o, _)| o.start_timestamp); assert_eq!( captured.len(), - 1, - "one window should close after UpdateAggConfigs" + 2, + "window [0,10_000) closes inline; [10_000,20_000) force-closes on shutdown" ); let (output, acc) = &captured[0]; @@ -2270,12 +2365,27 @@ aggregations: .downcast_ref::() .expect("should be SumAccumulator"); // Pre-update sample (t=1000, val=1.0) was dropped — agg_id was unknown. - // Post-update sample (t=5000, val=2.0) is the only one aggregated. + // Post-update sample (t=5000, val=2.0) is the only one in this window. assert!( (sum_acc.sum - 2.0).abs() < 1e-10, "only post-update sample should be aggregated, got {}", sum_acc.sum ); + + // The trailing window force-closed on shutdown holds the t=10_000 + // sample (val=0.0). + let (trailing_output, trailing_acc) = &captured[1]; + assert_eq!(trailing_output.start_timestamp, 10_000); + assert_eq!(trailing_output.end_timestamp, 20_000); + let trailing_sum = trailing_acc + .as_any() + .downcast_ref::() + .expect("should be SumAccumulator"); + assert!( + trailing_sum.sum.abs() < 1e-10, + "trailing window should hold the t=10_000 sample (val=0.0), got {}", + trailing_sum.sum + ); } // ----------------------------------------------------------------------- @@ -2424,4 +2534,74 @@ aggregations: "grace=0 must disable the fallback — event-time-only semantics" ); } + + // ----------------------------------------------------------------------- + // Test: shutdown force-close emits the trailing window + // + // The immediate-shutdown batch case: every record falls in one window and + // no later timestamp ever advances the watermark, so flush_all (with the + // wall-clock fallback disabled, grace=0) leaves the window open. On + // shutdown, force_close_all must close and emit it so the data reaches the + // store instead of being lost. + // ----------------------------------------------------------------------- + + #[test] + fn shutdown_force_close_emits_trailing_window() { + // 10s tumbling window; grace=0 isolates the force-close from the + // wall-clock fallback. + let cfg = make_agg_config( + 7, + "cpu", + AggregationType::SingleSubpopulation, + "Sum", + 10, + 0, + vec![], + ); + let agg_configs = HashMap::from([(7, cfg)]); + let sink = Arc::new(CapturingOutputSink::new()); + let mut worker = make_worker_with_grace(agg_configs, sink.clone(), 0); + + // All samples land in window [0, 10_000); the watermark freezes below + // the window end because no later timestamp ever arrives. + for i in 0..5 { + worker + .process_group_samples( + 7, + "", + group_samples("cpu", vec![(1_000 + i as i64 * 100, 1.0)]), + ) + .unwrap(); + } + + // A final flush must NOT close the window (event-time frozen, fallback + // disabled) — this is the bug the force-close fixes. + worker.flush_all().unwrap(); + assert_eq!( + sink.len(), + 0, + "trailing window must remain open after the final flush" + ); + + // Shutdown force-close closes and emits the trailing window. + worker.force_close_all().unwrap(); + let captured = sink.drain(); + assert_eq!( + captured.len(), + 1, + "shutdown force-close must emit the trailing window" + ); + let (output, _acc) = &captured[0]; + assert_eq!(output.aggregation_id, 7); + assert_eq!(output.start_timestamp, 0); + assert_eq!(output.end_timestamp, 10_000); + + // Idempotent: panes are drained, so a second force-close emits nothing. + worker.force_close_all().unwrap(); + assert_eq!( + sink.len(), + 0, + "force-close must be idempotent once panes are drained" + ); + } } From 6807dbc35606476f234dd2d270e01f792b383e34 Mon Sep 17 00:00:00 2001 From: Zeying Zhu Date: Tue, 16 Jun 2026 10:42:06 -0600 Subject: [PATCH 3/5] test(precompute): e2e regression for one-shot NetFlow single-second ingest MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Drives the real JsonFileIngestSource -> PrecomputeEngine -> sink path with a synthesized NetFlow JSONL whose records all fall in the same second. Sets wall_clock_grace_period_ms=0 so only the shutdown force-close can close the trailing window — a non-empty sink proves the force-close rescues the batch. Verified both directions: with the Shutdown force-close the single-second window is emitted with the summed bytes; with it disabled the run produces 0 outputs (reproducing the reported empty-store bug). Co-Authored-By: Claude Opus 4.8 (1M context) --- .../tests/e2e_netflow_single_second.rs | 161 ++++++++++++++++++ 1 file changed, 161 insertions(+) create mode 100644 asap-query-engine/tests/e2e_netflow_single_second.rs diff --git a/asap-query-engine/tests/e2e_netflow_single_second.rs b/asap-query-engine/tests/e2e_netflow_single_second.rs new file mode 100644 index 00000000..624df4eb --- /dev/null +++ b/asap-query-engine/tests/e2e_netflow_single_second.rs @@ -0,0 +1,161 @@ +//! End-to-end regression test for the one-shot NetFlow ingest failure mode. +//! +//! Scenario (reported in the field): a one-shot JSON ingest of NetFlow records +//! whose timestamps all fall within the **same second**. Every record lands in +//! a single 1-second tumbling window and no later timestamp ever arrives to +//! advance the watermark, so the window never closes via event-time. Before the +//! shutdown force-close, the store stayed empty even after the worker shut down. +//! +//! This test drives the real `JsonFileIngestSource → PrecomputeEngine → sink` +//! path. Crucially it sets `wall_clock_grace_period_ms = 0`, disabling the +//! wall-clock fallback, so the *only* thing that can close the window is the +//! shutdown force-close. A non-empty sink therefore proves the force-close is +//! what rescues the one-shot batch. + +use std::collections::HashMap; +use std::io::Write; +use std::sync::Arc; + +use asap_types::aggregation_config::AggregationConfig; +use asap_types::enums::{AggregationType, WindowType}; + +use query_engine_rust::data_model::StreamingConfig; +use query_engine_rust::precompute_engine::config::{LateDataPolicy, PrecomputeEngineConfig}; +use query_engine_rust::precompute_engine::output_sink::CapturingOutputSink; +use query_engine_rust::precompute_engine::{ + IngestSource, JsonFileIngestConfig, JsonFileIngestSource, PrecomputeEngine, TimestampUnit, +}; +use query_engine_rust::precompute_operators::sum_accumulator::SumAccumulator; + +fn netflow_agg_config(metric: &str, window_secs: u64) -> AggregationConfig { + AggregationConfig::new( + 1, + AggregationType::SingleSubpopulation, + "Sum".to_string(), + HashMap::new(), + promql_utilities::data_model::key_by_label_names::KeyByLabelNames::new(vec![]), + promql_utilities::data_model::key_by_label_names::KeyByLabelNames::new(vec![]), + promql_utilities::data_model::key_by_label_names::KeyByLabelNames::new(vec![]), + String::new(), + window_secs, + 0, + WindowType::Tumbling, + metric.to_string(), + metric.to_string(), + None, + None, + None, + None, + ) +} + +fn engine_config_grace_disabled() -> PrecomputeEngineConfig { + PrecomputeEngineConfig { + num_workers: 2, + allowed_lateness_ms: 0, + max_buffer_per_series: 10_000, + flush_interval_ms: 100, + channel_buffer_size: 10_000, + pass_raw_samples: false, + raw_mode_aggregation_id: 0, + late_data_policy: LateDataPolicy::Drop, + // Disable the wall-clock fallback: with grace=0 the trailing window can + // ONLY be closed by the shutdown force-close. This isolates the fix. + wall_clock_grace_period_ms: 0, + } +} + +#[tokio::test] +async fn netflow_single_second_batch_is_not_lost_on_shutdown() { + // --- Synthesize a NetFlow JSONL where every record is in the same second. + // json_ingest parses "YYYY-MM-DD HH:MM:SS" at second resolution, so all + // rows collapse onto one timestamp → one window, no watermark advance. + let rows = [ + ("2024-06-01 12:00:00", "10.0.0.1", "10.0.0.2", 100.0), + ("2024-06-01 12:00:00", "10.0.0.3", "10.0.0.4", 250.0), + ("2024-06-01 12:00:00", "10.0.0.5", "10.0.0.6", 75.0), + ("2024-06-01 12:00:00", "10.0.0.7", "10.0.0.8", 500.0), + ("2024-06-01 12:00:00", "10.0.0.9", "10.0.0.10", 25.0), + ]; + let expected_total_bytes: f64 = rows.iter().map(|r| r.3).sum(); + + let path = std::env::temp_dir().join(format!( + "netflow_single_second_{}.jsonl", + std::process::id() + )); + { + let mut f = std::fs::File::create(&path).expect("create temp netflow file"); + for (ts, src, dst, bytes) in rows.iter() { + writeln!( + f, + r#"{{"timestamp":"{ts}","src_ip":"{src}","dst_ip":"{dst}","bytes":{bytes}}}"# + ) + .unwrap(); + } + } + + // --- Wire up the real engine: 1s tumbling Sum over `bytes`. + let metric = "netflow_bytes"; + let mut agg_map = HashMap::new(); + agg_map.insert(1u64, netflow_agg_config(metric, 1)); + let streaming_config = Arc::new(StreamingConfig::new(agg_map)); + + let json_cfg = JsonFileIngestConfig { + path: path.to_string_lossy().to_string(), + metric_name: metric.to_string(), + value_col: "bytes".to_string(), + label_cols: vec![], // aggregate all flows in the second together + timestamp_col: "timestamp".to_string(), + timestamp_unit: TimestampUnit::Seconds, + batch_size: 1024, + }; + + let sink = Arc::new(CapturingOutputSink::new()); + let sources: Vec> = + vec![Box::new(JsonFileIngestSource::new(json_cfg))]; + let engine = PrecomputeEngine::new( + engine_config_grace_disabled(), + streaming_config, + sink.clone(), + sources, + ); + + // The JSON source reads the file then calls broadcast_shutdown(); engine.run() + // awaits the source and then the workers, so when it returns the shutdown + // force-close has already emitted into the sink. + engine.run().await.expect("engine run failed"); + + let _ = std::fs::remove_file(&path); + + // --- Verify: the one-second window reached the store. + let captured = sink.drain(); + assert_eq!( + captured.len(), + 1, + "the single-second NetFlow window must be emitted on shutdown (got {} outputs) — \ + before the shutdown force-close this was 0 and the store stayed empty", + captured.len() + ); + + let (output, acc) = &captured[0]; + // A 1-second tumbling window aligned to the epoch second. + assert_eq!( + output.end_timestamp - output.start_timestamp, + 1_000, + "expected a 1s window, got [{}, {})", + output.start_timestamp, + output.end_timestamp + ); + assert_eq!(output.start_timestamp % 1_000, 0, "window must be second-aligned"); + + let sum_acc = acc + .as_any() + .downcast_ref::() + .expect("NetFlow Sum aggregation should emit a SumAccumulator"); + assert!( + (sum_acc.sum - expected_total_bytes).abs() < 1e-9, + "window must hold the summed bytes of all flows in the second: expected {}, got {}", + expected_total_bytes, + sum_acc.sum + ); +} From d52591334dab78ec00e899bb8ce817302a46913e Mon Sep 17 00:00:00 2001 From: Zeying Zhu Date: Tue, 16 Jun 2026 13:26:57 -0600 Subject: [PATCH 4/5] style: cargo fmt for precompute worker tests and netflow e2e Co-Authored-By: Claude Opus 4.8 (1M context) --- asap-query-engine/src/precompute_engine/worker.rs | 6 +----- asap-query-engine/tests/e2e_netflow_single_second.rs | 9 ++++++--- 2 files changed, 7 insertions(+), 8 deletions(-) diff --git a/asap-query-engine/src/precompute_engine/worker.rs b/asap-query-engine/src/precompute_engine/worker.rs index 3b183481..f67d2f0f 100644 --- a/asap-query-engine/src/precompute_engine/worker.rs +++ b/asap-query-engine/src/precompute_engine/worker.rs @@ -2495,11 +2495,7 @@ aggregations: // Idempotent: a window closed by the fallback drains its pane and its // wall-clock bookkeeping, so a subsequent flush must not re-emit. worker.flush_all().unwrap(); - assert_eq!( - sink.len(), - 0, - "already-closed window must not re-emit" - ); + assert_eq!(sink.len(), 0, "already-closed window must not re-emit"); } #[test] diff --git a/asap-query-engine/tests/e2e_netflow_single_second.rs b/asap-query-engine/tests/e2e_netflow_single_second.rs index 624df4eb..8c99c200 100644 --- a/asap-query-engine/tests/e2e_netflow_single_second.rs +++ b/asap-query-engine/tests/e2e_netflow_single_second.rs @@ -111,8 +111,7 @@ async fn netflow_single_second_batch_is_not_lost_on_shutdown() { }; let sink = Arc::new(CapturingOutputSink::new()); - let sources: Vec> = - vec![Box::new(JsonFileIngestSource::new(json_cfg))]; + let sources: Vec> = vec![Box::new(JsonFileIngestSource::new(json_cfg))]; let engine = PrecomputeEngine::new( engine_config_grace_disabled(), streaming_config, @@ -146,7 +145,11 @@ async fn netflow_single_second_batch_is_not_lost_on_shutdown() { output.start_timestamp, output.end_timestamp ); - assert_eq!(output.start_timestamp % 1_000, 0, "window must be second-aligned"); + assert_eq!( + output.start_timestamp % 1_000, + 0, + "window must be second-aligned" + ); let sum_acc = acc .as_any() From 23e2bdeb7812d96adc071addf8052c07811e91d6 Mon Sep 17 00:00:00 2001 From: Zeying Zhu Date: Thu, 18 Jun 2026 06:59:23 -0600 Subject: [PATCH 5/5] fix(precompute): use finite bound in UpdateAggConfigs removed-agg force-close Addresses review feedback on #400. The removed-agg-id cleanup in the UpdateAggConfigs handler advanced the watermark to i64::MAX, the exact hazard force_close_all's docstring warns about: closed_windows enumerates window starts one slide at a time, so with a real epoch-ms watermark it loops ~i64::MAX/slide times and overflows `start + window_size_ms` (panics in debug). Use the same finite bound as force_close_all (max_pane + window_size_ms). Add test_update_agg_configs_removed_id_force_closes_at_epoch_ms_timestamps, which opens a window at a realistic epoch-ms timestamp and removes the agg_id; it hangs/overflows under the old i64::MAX code (verified: times out) and passes with the finite bound. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../src/precompute_engine/worker.rs | 117 +++++++++++++++++- 1 file changed, 112 insertions(+), 5 deletions(-) diff --git a/asap-query-engine/src/precompute_engine/worker.rs b/asap-query-engine/src/precompute_engine/worker.rs index f67d2f0f..e00df70e 100644 --- a/asap-query-engine/src/precompute_engine/worker.rs +++ b/asap-query-engine/src/precompute_engine/worker.rs @@ -249,12 +249,23 @@ impl Worker { if state.previous_watermark_ms == i64::MIN { continue; // No samples received — nothing to emit. } - // Force-close all open windows by advancing the watermark - // to i64::MAX. No new samples will arrive for this group. + // Force-close all open windows; no new samples will arrive + // for this removed agg_id. Advance to a *finite* bound + // (`max_pane + window_size_ms`), NOT `i64::MAX`: + // `closed_windows` enumerates window starts one slide at a + // time, so `i64::MAX` would loop ~`i64::MAX / slide` times + // and overflow (see `force_close_all`). let mut active_panes = state.active_panes; - let closed = state - .window_manager - .closed_windows(state.previous_watermark_ms, i64::MAX); + let closed = match active_panes.keys().next_back() { + Some(&max_pane) => { + let force_wm = max_pane + .saturating_add(state.window_manager.window_size_ms()); + state + .window_manager + .closed_windows(state.previous_watermark_ms, force_wm) + } + None => Vec::new(), // no open panes + }; for window_start in &closed { let (_, window_end) = @@ -2390,6 +2401,102 @@ aggregations: ); } + // ----------------------------------------------------------------------- + // Test: removing an agg_id force-closes its open windows with a finite + // bound — at realistic (epoch-ms) timestamps. + // + // The removed-agg cleanup used to advance the watermark to `i64::MAX`. + // `closed_windows` enumerates window starts one slide at a time, so with a + // real epoch-ms watermark that loop runs ~`i64::MAX / slide` iterations and + // overflows `start + window_size_ms` (panics in debug) — the bug the + // reviewer flagged. With the finite `max_pane + window_size_ms` bound this + // completes instantly and emits exactly the open window. + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn test_update_agg_configs_removed_id_force_closes_at_epoch_ms_timestamps() { + // Realistic event time: ~2023-11-14T22:13:20Z in ms. With the old + // i64::MAX code this is what made closed_windows blow up. + let base_ms: i64 = 1_700_000_000_000; + let window_secs = 10u64; + + let config = make_agg_config( + 1, + "cpu", + AggregationType::SingleSubpopulation, + "Sum", + window_secs, + 0, + vec![], + ); + + let sink = Arc::new(CapturingOutputSink::new()); + let (tx, rx) = tokio::sync::mpsc::channel(32); + let wm = Arc::new(AtomicI64::new(i64::MIN)); + let mut agg_configs = HashMap::new(); + agg_configs.insert(1u64, Arc::new(config)); + let worker = Worker::new( + 0, + rx, + sink.clone(), + agg_configs, + WorkerRuntimeConfig { + max_buffer_per_series: 10_000, + allowed_lateness_ms: 0, + pass_raw_samples: false, + raw_mode_aggregation_id: 0, + late_data_policy: LateDataPolicy::Drop, + wall_clock_grace_period_ms: 0, + }, + Arc::new(AtomicUsize::new(0)), + wm.clone(), + vec![wm], + ); + let handle = tokio::spawn(async move { worker.run().await }); + + // Open a window at a large epoch-ms timestamp; nothing advances the + // watermark past it, so it's still open when the agg_id is removed. + tx.send(WorkerMessage::GroupSamples { + agg_id: 1, + group_key: String::new(), + samples: vec![("cpu".to_string(), base_ms, 7.0)], + ingest_received_at: std::time::Instant::now(), + }) + .await + .unwrap(); + + // Remove agg_id=1 from the config → triggers the removed-agg force-close. + tx.send(WorkerMessage::UpdateAggConfigs(HashMap::new())) + .await + .unwrap(); + + tx.send(WorkerMessage::Shutdown).await.unwrap(); + // If the finite bound regressed to i64::MAX this join would hang or + // panic on overflow instead of completing. + handle.await.unwrap(); + + let captured = sink.drain(); + assert_eq!( + captured.len(), + 1, + "removing the agg_id must force-close the single open window" + ); + let (output, acc) = &captured[0]; + let window_ms = window_secs as i64 * 1_000; + let expected_start = base_ms - (base_ms % window_ms); + assert_eq!(output.start_timestamp as i64, expected_start); + assert_eq!(output.end_timestamp as i64, expected_start + window_ms); + let sum_acc = acc + .as_any() + .downcast_ref::() + .expect("should be SumAccumulator"); + assert!( + (sum_acc.sum - 7.0).abs() < 1e-10, + "force-closed window should hold the ingested sample (val=7.0), got {}", + sum_acc.sum + ); + } + // ----------------------------------------------------------------------- // Test: wall-clock fallback closes an idle window when event-time freezes //