diff --git a/data_plane/src/precompute_engine/engine.rs b/data_plane/src/precompute_engine/engine.rs index 085d7275..c1edf95b 100644 --- a/data_plane/src/precompute_engine/engine.rs +++ b/data_plane/src/precompute_engine/engine.rs @@ -138,7 +138,6 @@ impl PrecomputeEngine { }, self.diagnostics.worker_group_counts[id].clone(), self.diagnostics.worker_watermarks[id].clone(), - self.diagnostics.worker_watermarks.to_vec(), ); let handle = tokio::spawn(async move { worker.run().await; diff --git a/data_plane/src/precompute_engine/precompute_engine_design_doc.md b/data_plane/src/precompute_engine/precompute_engine_design_doc.md index ca9f746d..cfa301ea 100644 --- a/data_plane/src/precompute_engine/precompute_engine_design_doc.md +++ b/data_plane/src/precompute_engine/precompute_engine_design_doc.md @@ -98,69 +98,21 @@ Late data handling: 3 < 6 → YES, late → DROP (or ForwardToStore) ``` -### Cross-group watermark propagation +### Per-group event-time watermarks -Without cross-group propagation, each group tracks its own watermark -independently. If a group stops receiving data, its watermark freezes and its -windows never close. Cross-group propagation solves this with two layers: +Each group tracks its own maximum observed event timestamp. Its event-time +watermark is `max_event_time_ms - allowed_lateness_ms`. Periodic flushes never +modify observed event time. -**Layer 1 — Intra-worker (max):** Each worker computes its worker watermark as -`max(all group watermarks)`. This represents "time has progressed to at least -here on this worker." During each flush, idle groups are advanced to the worker -watermark. +A timestamp observed for one group does not advance another group. Independent +sources can progress at different rates, so cross-group propagation can close a +live window prematurely and turn valid input into late data. The worker-level +watermark atomic reports the maximum group watermark for diagnostics only; it +is not fed back into group closure. -**Layer 2 — Cross-worker (min):** Each worker publishes its worker watermark to -a shared `Arc`. The global watermark is `min(all worker watermarks)`, -ignoring workers that have not yet started. This becomes the floor for all group -watermarks across all workers. - -``` - ┌──────────────────────────────────────────┐ - │ Shared Atomics │ - │ AtomicI64[0] AtomicI64[1] AtomicI64[2]│ - │ 100s 80s 90s │ - └──┬──────────────┬──────────────┬─────────┘ - │ store │ store │ store - │ (Release) │ (Release) │ (Release) - ┌────────┴───┐ ┌───────┴────┐ ┌─────┴──────┐ - │ Worker 0 │ │ Worker 1 │ │ Worker 2 │ - │ │ │ │ │ │ - │ Groups: │ │ Groups: │ │ Groups: │ - │ A: wm=100s│ │ C: wm=80s│ │ E: wm=90s│ - │ B: wm=50s │ │ D: wm=80s│ │ F: wm=30s│ - │ │ │ │ │ │ - │ worker_wm │ │ worker_wm │ │ worker_wm │ - │ = max(A,B) │ │ = max(C,D) │ │ = max(E,F) │ - │ = 100s │ │ = 80s │ │ = 90s │ - └────────────┘ └────────────┘ └────────────┘ - │ load all │ load all │ load all - │ (Acquire) │ (Acquire) │ (Acquire) - ▼ ▼ ▼ - global_wm = min(100s, 80s, 90s) = 80s - - On flush, each group's effective watermark becomes: - max(group_wm, global_wm) + 1ms - - Worker 0: Group B (50s) → advanced to 80s → closes [50s, 80s] windows - Worker 2: Group F (30s) → advanced to 80s → closes [30s, 80s] windows -``` - -**Why max within a worker?** We want to propagate forward progress from active -groups to idle groups on the same worker. - -**Why min across workers?** Conservative: only advance as far as ALL workers -agree time has progressed. If worker 1 is behind at 80s, we should not close -windows at 90s on worker 2 because worker 1 might still send data for those -windows. - -**Staleness:** Because workers read each other's atomics during flush, the -global watermark may be up to one `flush_interval_ms` (default 1s) stale. -This is acceptable — it only means idle groups close windows one flush cycle -later than they theoretically could. - -**Unstarted workers:** Workers that have not yet received any data remain at -`i64::MIN` and are excluded from the global watermark min calculation. This -prevents a cold worker from blocking the entire system. +Idle and bounded-time closure use a separate wall-clock policy. Distributed +completion requires an explicit source/window barrier; an unrelated group's +timestamp is not such a barrier. ## 3. Components @@ -384,15 +336,16 @@ from `active_panes`. Remaining panes are read non-destructively via ``` 1. Match series to AggregationConfigs (by metric name / spatial_filter) -2. Insert samples into SeriesBuffer, update watermark -3. Drop samples beyond allowed_lateness_ms behind watermark +2. Insert samples and update maximum observed event time +3. Compute event watermark as `max_event_time_ms - allowed_lateness_ms` 4. For each sample × each aggregation: a. Compute pane_start = pane_start_for(ts) b. If pane was evicted (late data for closed window): → late_data_policy == Drop: skip → late_data_policy == ForwardToStore: create mini-accumulator, emit c. Else: get-or-create pane in active_panes, feed value (1 update per sample) -5. Detect newly closed windows via closed_windows(prev_wm, current_wm) +5. Detect newly closed windows via + `closed_windows(previous_closure_watermark, event_watermark)` 6. For each closed window: a. Get pane starts via panes_for_window(window_start) b. Oldest pane: take_accumulator() + remove from active_panes (destructive) @@ -400,7 +353,7 @@ from `active_panes`. Remaining panes are read non-destructively via d. Merge all pane accumulators via AggregateCore::merge_with() e. Emit merged result as PrecomputedOutput + AggregateCore 7. Emit batch to OutputSink -8. Update previous_watermark_ms +8. Update observed event time and the separate closure watermark ``` #### Raw mode diff --git a/data_plane/src/precompute_engine/worker.rs b/data_plane/src/precompute_engine/worker.rs index 94722914..d8a8534f 100644 --- a/data_plane/src/precompute_engine/worker.rs +++ b/data_plane/src/precompute_engine/worker.rs @@ -57,9 +57,13 @@ struct GroupState { /// `Box` objects and do not share the updater /// machinery used for incremental sample updates. sketch_panes: BTreeMap>, - /// Per-group watermark: tracks the maximum timestamp seen across all - /// series in this group on this worker. - previous_watermark_ms: i64, + /// Maximum event timestamp actually observed from input for this group. + /// Periodic flushes never modify this value. + max_event_time_ms: i64, + /// Monotonic watermark through which windows have already been closed. + /// This is separate from observed event time because wall-clock policies + /// may close a window without manufacturing a later input timestamp. + closure_watermark_ms: i64, /// Wall-clock time (ms since epoch) at which each currently-open pane /// was last touched by an input. Used by `flush_all`'s wall-clock fallback /// to close panes that have been idle too long when event-time has @@ -126,11 +130,9 @@ pub struct Worker { raw_mode_aggregation_id: u64, /// Policy for handling late samples that arrive after their window has closed. late_data_policy: LateDataPolicy, - /// This worker's watermark atomic, shared with engine for cross-worker reads. - /// Updated during flush with max(all group watermarks). + /// This worker's maximum observed event-time watermark, exposed only for + /// diagnostics. It is never propagated into another group's closure state. worker_watermark: Arc, - /// All worker watermark atomics (including self), for computing global watermark. - all_worker_watermarks: Vec>, /// Externally-readable group count for diagnostics. group_count: Arc, /// Grace period (ms) for the wall-clock fallback in `flush_all`. @@ -155,7 +157,6 @@ impl Worker { runtime_config: WorkerRuntimeConfig, group_count: Arc, worker_watermark: Arc, - all_worker_watermarks: Vec>, ) -> Self { let WorkerRuntimeConfig { max_buffer_per_series: _, @@ -176,7 +177,6 @@ impl Worker { raw_mode_aggregation_id, late_data_policy, worker_watermark, - all_worker_watermarks, group_count, wall_clock_grace_period_ms, now_ms_fn: Box::new(default_now_ms), @@ -299,11 +299,9 @@ impl Worker { 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. + // The wall-clock fallback may not yet be due for a one-shot + // batch, so the trailing window can remain open. No more + // samples will arrive after shutdown; close every pane. if let Err(e) = self.force_close_all() { warn!("Worker {} shutdown force-close error: {}", self.id, e); } @@ -349,7 +347,8 @@ impl Worker { group_key: group_key.to_string(), active_panes: BTreeMap::new(), sketch_panes: BTreeMap::new(), - previous_watermark_ms: i64::MIN, + max_event_time_ms: i64::MIN, + closure_watermark_ms: i64::MIN, pane_wall_clock_last_touches_ms: BTreeMap::new(), }; self.group_states.insert(sid, gs); @@ -397,40 +396,41 @@ impl Worker { .map(|(_, ts, _)| *ts) .max() .unwrap_or(i64::MIN); - let previous_wm = state.previous_watermark_ms; - let current_wm = if batch_max_ts > previous_wm { + let previous_event_time = state.max_event_time_ms; + let current_event_time = if batch_max_ts > previous_event_time { batch_max_ts } else { - previous_wm + previous_event_time }; + let event_watermark = watermark_for_event_time(current_event_time, allowed_lateness_ms); + let previous_closure_watermark = state.closure_watermark_ms; let mut emit_batch: Vec<(PrecomputedOutput, Box)> = Vec::new(); // Route each sample to its pane for (series_key, ts, val) in &samples { - // Drop late samples - if previous_wm != i64::MIN && *ts < previous_wm - allowed_lateness_ms { - debug!( - "Worker {} dropping late sample for sid={} (group={}): ts={} watermark={}", - worker_id, sid, group_key, ts, previous_wm - ); - continue; - } - + let too_late = previous_event_time != i64::MIN + && *ts < watermark_for_event_time(previous_event_time, allowed_lateness_ms); let pane_start = state.window_manager.pane_start_for(*ts); let pane_end = pane_start + state.window_manager.slide_interval_ms(); + let pane_closed = !state.active_panes.contains_key(&pane_start) + && previous_closure_watermark >= pane_start + state.window_manager.window_size_ms(); - // Check if pane was already evicted (late data for a closed window) - if !state.active_panes.contains_key(&pane_start) - && current_wm >= pane_start + state.window_manager.window_size_ms() - { + if too_late || pane_closed { let window_start = pane_start; let window_end = pane_start + state.window_manager.window_size_ms(); match late_data_policy { LateDataPolicy::Drop => { debug!( - "Dropping late sample for evicted pane [{}, {})", - pane_start, pane_end + "Worker {} dropping late sample for sid={} (group={}): \ + ts={} observed_event_time={} pane=[{}, {})", + worker_id, + sid, + group_key, + ts, + previous_event_time, + pane_start, + pane_end ); continue; } @@ -470,7 +470,9 @@ impl Worker { } // Check for closed windows - let closed = state.window_manager.closed_windows(previous_wm, current_wm); + let closed = state + .window_manager + .closed_windows(previous_closure_watermark, event_watermark); for window_start in &closed { let (_, window_end) = state.window_manager.window_bounds(*window_start); @@ -489,7 +491,10 @@ impl Worker { } } - state.previous_watermark_ms = current_wm; + state.max_event_time_ms = current_event_time; + if event_watermark > state.closure_watermark_ms { + state.closure_watermark_ms = event_watermark; + } state.prune_pane_wall_clock_last_touches(); // Emit to output sink @@ -547,27 +552,30 @@ impl Worker { } let state = self.group_states.get_mut(&sid).unwrap(); - let previous_wm = state.previous_watermark_ms; - let current_wm = if timestamp_ms > previous_wm { + let previous_event_time = state.max_event_time_ms; + let current_event_time = if timestamp_ms > previous_event_time { timestamp_ms } else { - previous_wm + previous_event_time }; + let event_watermark = watermark_for_event_time(current_event_time, allowed_lateness_ms); + let previous_closure_watermark = state.closure_watermark_ms; let mut emit_batch: Vec<(PrecomputedOutput, Box)> = Vec::new(); // Late-arrival check against the existing watermark. - let too_late = previous_wm != i64::MIN && timestamp_ms < previous_wm - allowed_lateness_ms; + let too_late = previous_event_time != i64::MIN + && timestamp_ms < watermark_for_event_time(previous_event_time, allowed_lateness_ms); let pane_start = state.window_manager.pane_start_for(timestamp_ms); let pane_closed = !state.sketch_panes.contains_key(&pane_start) - && current_wm >= pane_start + state.window_manager.window_size_ms(); + && previous_closure_watermark >= pane_start + state.window_manager.window_size_ms(); if too_late || pane_closed { match late_data_policy { LateDataPolicy::Drop => { debug!( "Worker {} dropping late accumulator input for sid={} (group={}): ts={} watermark={}", - worker_id, sid, group_key, timestamp_ms, previous_wm + worker_id, sid, group_key, timestamp_ms, previous_event_time ); } LateDataPolicy::ForwardToStore => { @@ -612,7 +620,9 @@ impl Worker { } // Check for closed windows and emit merged outputs. - let closed = state.window_manager.closed_windows(previous_wm, current_wm); + let closed = state + .window_manager + .closed_windows(previous_closure_watermark, event_watermark); 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); @@ -646,7 +656,10 @@ impl Worker { } } - state.previous_watermark_ms = current_wm; + state.max_event_time_ms = current_event_time; + if event_watermark > state.closure_watermark_ms { + state.closure_watermark_ms = event_watermark; + } state.prune_pane_wall_clock_last_touches(); if !emit_batch.is_empty() { @@ -755,27 +768,24 @@ impl Worker { let now_ms = (self.now_ms_fn)(); let grace_ms = self.wall_clock_grace_period_ms; - // Step 1: Compute worker watermark = max of all group watermarks. + // Publish the largest observed event-time watermark for diagnostics. + // It must not be fed back into another group's closure decision: group + // timestamps are independent unless an explicit source barrier says + // otherwise. let worker_wm = self .group_states .values() - .map(|s| s.previous_watermark_ms) + .map(|s| watermark_for_event_time(s.max_event_time_ms, self.allowed_lateness_ms)) .filter(|&wm| wm != i64::MIN) .max() .unwrap_or(i64::MIN); - - // Step 2: Publish our worker watermark for cross-worker reads. self.worker_watermark.store(worker_wm, Ordering::Release); - // Step 3: Compute global watermark = min(all worker watermarks). - let global_wm = self.compute_global_watermark(); - - // Step 4: For each bucket, advance watermark and close due windows. let mut emit_batch: Vec<(PrecomputedOutput, Box)> = Vec::new(); for (&sid, state) in &mut self.group_states { let _ = sid; // sid is the bucket key; group_key/policy_fp live on `state` - if state.previous_watermark_ms == i64::MIN { + if state.max_event_time_ms == i64::MIN { continue; // No samples received yet — no panes to close. } // group_key/policy_fp travelled in on the message and are @@ -784,18 +794,13 @@ impl Worker { // `state` mutably for pane drains. let group_key = state.group_key.clone(); - // Effective watermark: max(group's own, global) + 1ms for boundary. - let propagated_wm = if global_wm != i64::MIN { - state.previous_watermark_ms.max(global_wm) - } else { - state.previous_watermark_ms - }; - let mut effective_wm = propagated_wm.saturating_add(1); + // Start from this group's existing closure watermark. A timer tick + // alone must not advance event time. + let mut effective_wm = state.closure_watermark_ms; // Wall-clock fallback for stuck event-time. If the agent // stamps every sketch with the same `time_unix_nano` - // (e.g. window-start), `previous_watermark_ms` freezes - // and `closed_windows(prev, prev+1)` returns empty + // (e.g. window-start), its event watermark freezes // forever — the 30s window never closes and ASAP-tier // queries come back empty even though sketches keep // arriving (sweep blocker #2). Force `effective_wm` past @@ -817,7 +822,7 @@ impl Worker { let closed = state .window_manager - .closed_windows(state.previous_watermark_ms, effective_wm); + .closed_windows(state.closure_watermark_ms, effective_wm); for window_start in &closed { let (_, window_end) = state.window_manager.window_bounds(*window_start); @@ -850,11 +855,8 @@ impl Worker { } } - // 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; + if effective_wm > state.closure_watermark_ms { + state.closure_watermark_ms = effective_wm; } state.prune_pane_wall_clock_last_touches(); @@ -874,11 +876,10 @@ impl Worker { /// 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 + /// Unlike `flush_all` — which preserves observed event time and only applies + /// configured wall-clock closure — this emits every remaining pane 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. Covers both `active_panes` (sample /// aggregation) and `sketch_panes` (OTLP-delivered sketches). @@ -901,7 +902,7 @@ impl Worker { for (&sid, state) in &mut self.group_states { let _ = sid; // sid is the bucket key; group_key/policy_fp live on `state` - if state.previous_watermark_ms == i64::MIN { + if state.max_event_time_ms == i64::MIN { continue; // never received data — nothing to close } @@ -921,7 +922,7 @@ impl Worker { let group_key = state.group_key.clone(); let closed = state .window_manager - .closed_windows(state.previous_watermark_ms, force_wm); + .closed_windows(state.closure_watermark_ms, force_wm); for window_start in &closed { let (_, window_end) = state.window_manager.window_bounds(*window_start); @@ -954,8 +955,8 @@ impl Worker { } } - if force_wm > state.previous_watermark_ms { - state.previous_watermark_ms = force_wm; + if force_wm > state.closure_watermark_ms { + state.closure_watermark_ms = force_wm; } state.prune_pane_wall_clock_last_touches(); } @@ -971,25 +972,6 @@ impl Worker { 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 { - let mut global_wm = i64::MAX; - let mut any_started = false; - for wm_atomic in &self.all_worker_watermarks { - let wm = wm_atomic.load(Ordering::Acquire); - if wm != i64::MIN { - global_wm = global_wm.min(wm); - any_started = true; - } - } - if any_started { - global_wm - } else { - i64::MIN - } - } } /// Build a `KeyByLabelValues` from a semicolon-delimited group key string. @@ -1009,6 +991,17 @@ fn default_now_ms() -> i64 { .unwrap_or(0) } +/// Convert observed event time to the watermark that is safe to close through. +/// A negative configured lateness is treated as zero rather than advancing the +/// watermark beyond any timestamp actually observed. +fn watermark_for_event_time(max_event_time_ms: i64, allowed_lateness_ms: i64) -> i64 { + if max_event_time_ms == i64::MIN { + i64::MIN + } else { + max_event_time_ms.saturating_sub(allowed_lateness_ms.max(0)) + } +} + fn build_group_key_label_values(group_key: &str) -> KeyByLabelValues { let labels: Vec = group_key.split(';').map(|s| s.to_string()).collect(); KeyByLabelValues::new_with_labels(labels) @@ -1437,6 +1430,17 @@ mod tests { pass_raw: bool, raw_agg_id: u64, late_policy: LateDataPolicy, + ) -> Worker { + make_worker_with_lateness(agg_configs, sink, pass_raw, raw_agg_id, late_policy, 0) + } + + fn make_worker_with_lateness( + agg_configs: HashMap, + sink: Arc, + pass_raw: bool, + raw_agg_id: u64, + late_policy: LateDataPolicy, + allowed_lateness_ms: i64, ) -> Worker { let (_tx, rx) = tokio::sync::mpsc::channel(1); let wm = Arc::new(AtomicI64::new(i64::MIN)); @@ -1447,15 +1451,14 @@ mod tests { make_hot_reload(agg_configs), WorkerRuntimeConfig { max_buffer_per_series: 10_000, - allowed_lateness_ms: 0, + allowed_lateness_ms, 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(), - vec![wm], + wm, ) } @@ -1982,8 +1985,7 @@ mod tests { wall_clock_grace_period_ms: 0, }, Arc::new(AtomicUsize::new(0)), - wm.clone(), - vec![wm], + wm, ); // Establish watermark at t=20000ms @@ -2036,17 +2038,17 @@ mod tests { wall_clock_grace_period_ms: 0, }, Arc::new(AtomicUsize::new(0)), - wm.clone(), - vec![wm], + wm, ); - // Seed then advance watermark to 20000 + // Seed then advance max event time far enough that the 15s lateness + // budget permits closing [0, 10s). let pf = PolicyFingerprint(5); worker .process_group_samples(5, pf, "", group_samples("cpu", vec![(500, 1.0)])) .unwrap(); worker - .process_group_samples(5, pf, "", group_samples("cpu", vec![(20_000, 0.0)])) + .process_group_samples(5, pf, "", group_samples("cpu", vec![(30_000, 0.0)])) .unwrap(); let _ = sink.drain(); @@ -2214,10 +2216,10 @@ aggregations: // ----------------------------------------------------------------------- #[test] - fn test_intra_worker_watermark_propagation() { - // Two groups on the same worker. Group A advances to t=100s. - // Group B has data at t=10s and then goes idle. - // After flush, group B's idle windows should close via propagation. + fn test_group_watermarks_are_isolated() { + // Two groups on the same worker. Group A advances to t=100s while + // group B remains active at t=5s. Group A is not evidence that group + // B's source has completed its earlier window. let config = make_agg_config( 1, "cpu", @@ -2268,12 +2270,11 @@ aggregations: .unwrap(); let _ = sink.drain(); - // Group B has NOT received new data — its watermark is still at 5s. - // Flush should propagate group A's watermark to group B. + // Group B has NOT received new data — its event-time watermark is + // still at 5s. Flushing must not borrow group A's timestamp. worker.flush_all().unwrap(); let flushed = sink.drain(); - // Group B's window [0, 10s) should now be closed via propagation. let group_b_outputs: Vec<_> = flushed .iter() .filter(|(out, _)| { @@ -2284,102 +2285,120 @@ aggregations: }) .collect(); assert!( - !group_b_outputs.is_empty(), - "idle group B should have windows closed via watermark propagation" + group_b_outputs.is_empty(), + "one group must not force-close another group's event-time window" ); + + worker + .process_group_samples( + sid_b, + pf, + "groupB", + group_samples("cpu", vec![(5_000, 4.0)]), + ) + .unwrap(); + worker.force_close_all().unwrap(); + let group_b = sink + .drain() + .into_iter() + .find(|(out, _)| { + out.key + .as_ref() + .map(|k| k.labels == vec!["groupB".to_string()]) + .unwrap_or(false) + }) + .expect("group B should emit on shutdown"); + let sum = group_b + .1 + .as_any() + .downcast_ref::() + .expect("must emit SumAccumulator"); + assert_eq!(sum.sum, 6.0, "group B's second sample must not be late"); } #[test] - fn test_compute_global_watermark_min_of_started() { - let wm0 = Arc::new(AtomicI64::new(100_000)); - let wm1 = Arc::new(AtomicI64::new(80_000)); - let wm2 = Arc::new(AtomicI64::new(90_000)); - let all = vec![wm0.clone(), wm1.clone(), wm2.clone()]; - - let (_tx, rx) = tokio::sync::mpsc::channel(1); - let worker = Worker::new( + fn repeated_flushes_do_not_make_fixed_timestamp_input_late() { + let config = make_agg_config( + 1, + "cpu", + AggregationType::SingleSubpopulation, + "Sum", + 1, 0, - rx, - Arc::new(CapturingOutputSink::new()), - make_hot_reload(HashMap::new()), - 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)), - wm0, - all, + vec![], ); - - assert_eq!(worker.compute_global_watermark(), 80_000); - } - - #[test] - fn test_compute_global_watermark_ignores_unstarted() { - let wm0 = Arc::new(AtomicI64::new(100_000)); - let wm1 = Arc::new(AtomicI64::new(i64::MIN)); // not started - let all = vec![wm0.clone(), wm1.clone()]; - - let (_tx, rx) = tokio::sync::mpsc::channel(1); - let worker = Worker::new( + let sink = Arc::new(CapturingOutputSink::new()); + let mut worker = make_worker_with_lateness( + HashMap::from([(1, config)]), + sink.clone(), + false, 0, - rx, - Arc::new(CapturingOutputSink::new()), - make_hot_reload(HashMap::new()), - 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)), - wm0, - all, + LateDataPolicy::Drop, + 1, ); + let pf = PolicyFingerprint(1); - assert_eq!( - worker.compute_global_watermark(), - 100_000, - "unstarted workers (i64::MIN) should be ignored" - ); + worker + .process_group_samples(1, pf, "", group_samples("cpu", vec![(0, 1.0)])) + .unwrap(); + worker.flush_all().unwrap(); + worker.flush_all().unwrap(); + worker + .process_group_samples(1, pf, "", group_samples("cpu", vec![(0, 2.0)])) + .unwrap(); + worker.force_close_all().unwrap(); + + let emitted = sink.drain(); + assert_eq!(emitted.len(), 1); + let sum = emitted[0] + .1 + .as_any() + .downcast_ref::() + .expect("must emit SumAccumulator"); + assert_eq!(sum.sum, 3.0, "flush must not manufacture event time"); } #[test] - fn test_compute_global_watermark_all_unstarted() { - let wm0 = Arc::new(AtomicI64::new(i64::MIN)); - let wm1 = Arc::new(AtomicI64::new(i64::MIN)); - let all = vec![wm0.clone(), wm1.clone()]; - - let (_tx, rx) = tokio::sync::mpsc::channel(1); - let worker = Worker::new( + fn allowed_lateness_delays_event_time_window_close() { + let config = make_agg_config( + 1, + "cpu", + AggregationType::SingleSubpopulation, + "Sum", + 10, 0, - rx, - Arc::new(CapturingOutputSink::new()), - make_hot_reload(HashMap::new()), - 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)), - wm0, - all, + vec![], ); + let sink = Arc::new(CapturingOutputSink::new()); + let mut worker = make_worker_with_lateness( + HashMap::from([(1, config)]), + sink.clone(), + false, + 0, + LateDataPolicy::Drop, + 5_000, + ); + let pf = PolicyFingerprint(1); + worker + .process_group_samples(1, pf, "", group_samples("cpu", vec![(5_000, 1.0)])) + .unwrap(); + worker + .process_group_samples(1, pf, "", group_samples("cpu", vec![(10_000, 2.0)])) + .unwrap(); assert_eq!( - worker.compute_global_watermark(), - i64::MIN, - "all unstarted should return i64::MIN" + sink.len(), + 0, + "window [0, 10s) remains open until max event time reaches 15s" ); + + worker + .process_group_samples(1, pf, "", group_samples("cpu", vec![(15_000, 3.0)])) + .unwrap(); + let emitted = sink.drain(); + assert_eq!(emitted.len(), 1); + assert_eq!(emitted[0].0.start_timestamp, 0); + assert_eq!(emitted[0].0.end_timestamp, 10_000); } #[test] @@ -2396,7 +2415,6 @@ aggregations: let agg_configs = HashMap::from([(1, config)]); let sink = Arc::new(CapturingOutputSink::new()); let wm = Arc::new(AtomicI64::new(i64::MIN)); - let all = vec![wm.clone()]; let (_tx, rx) = tokio::sync::mpsc::channel(1); let mut worker = Worker::new( 0, @@ -2413,7 +2431,6 @@ aggregations: }, Arc::new(AtomicUsize::new(0)), wm.clone(), - all, ); assert_eq!(wm.load(Ordering::Acquire), i64::MIN); @@ -2697,8 +2714,7 @@ aggregations: wall_clock_grace_period_ms, }, Arc::new(AtomicUsize::new(0)), - wm.clone(), - vec![wm], + wm, ) } @@ -2770,7 +2786,7 @@ aggregations: !captured.is_empty(), "wall-clock fallback failed to close the idle window — \ this is the live sweep blocker #2 root cause: with frozen event-time, \ - flush_all's +1ms event-time advance is a no-op and the 30s window \ + the 30s window \ never closes." );