From 0751a2d883d1644b45e100745c2074e7f68c24b7 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 3 Sep 2026 12:27:58 -0600 Subject: [PATCH] fix(precompute): preserve deadline corrections --- data_plane/src/main.rs | 6 +- data_plane/src/precompute_engine/config.rs | 21 +- data_plane/src/precompute_engine/metrics.rs | 37 ++++ data_plane/src/precompute_engine/mod.rs | 1 + .../precompute_engine_design_doc.md | 45 ++--- data_plane/src/precompute_engine/worker.rs | 181 ++++++++++++++++-- 6 files changed, 241 insertions(+), 50 deletions(-) create mode 100644 data_plane/src/precompute_engine/metrics.rs diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs index f13c65d2..cb6209cf 100644 --- a/data_plane/src/main.rs +++ b/data_plane/src/main.rs @@ -512,11 +512,9 @@ async fn main() -> Result<()> { channel_buffer_size: args.precompute_channel_buffer_size, pass_raw_samples: false, raw_mode_aggregation_id: 0, - late_data_policy: LateDataPolicy::Drop, + late_data_policy: LateDataPolicy::ForwardToStore, wall_clock_idle_grace_period_ms: 5_000, - // Enabled only after deadline-triggered late corrections are - // guaranteed to append rather than drop. - wall_clock_max_open_grace_period_ms: 0, + wall_clock_max_open_grace_period_ms: 5_000, schema_persist_path: args.schema_persist_path.clone(), }; // M2.3.6 — sketch-only sink. Precompute writes now go to diff --git a/data_plane/src/precompute_engine/config.rs b/data_plane/src/precompute_engine/config.rs index 5c803ec1..4a18e544 100644 --- a/data_plane/src/precompute_engine/config.rs +++ b/data_plane/src/precompute_engine/config.rs @@ -40,9 +40,10 @@ pub struct PrecomputeEngineConfig { pub wall_clock_idle_grace_period_ms: i64, /// Additional grace for the absolute wall-clock deadline. A pane closes /// after `window_size + max_open_grace` from its first input even if it is - /// still active. Non-positive disables the deadline. It stays disabled by - /// default until late corrections are guaranteed not to be dropped. - #[serde(default)] + /// still active. Non-positive disables the deadline. The worker applies an + /// absolute deadline only with `ForwardToStore`, ensuring later inputs are + /// emitted as mergeable corrections. Default: 5000 ms. + #[serde(default = "default_wall_clock_max_open_grace_period_ms")] pub wall_clock_max_open_grace_period_ms: i64, /// Optional path where the `SchemaRegistry` persists per-`agg_id` /// lifecycle state across restarts (sketch DB Phase 2c). When @@ -65,9 +66,9 @@ impl Default for PrecomputeEngineConfig { channel_buffer_size: 10_000, pass_raw_samples: false, raw_mode_aggregation_id: 0, - late_data_policy: LateDataPolicy::Drop, + late_data_policy: LateDataPolicy::ForwardToStore, wall_clock_idle_grace_period_ms: default_wall_clock_idle_grace_period_ms(), - wall_clock_max_open_grace_period_ms: 0, + wall_clock_max_open_grace_period_ms: default_wall_clock_max_open_grace_period_ms(), schema_persist_path: None, } } @@ -77,6 +78,10 @@ fn default_wall_clock_idle_grace_period_ms() -> i64 { 5_000 } +fn default_wall_clock_max_open_grace_period_ms() -> i64 { + 5_000 +} + #[cfg(test)] mod tests { use super::*; @@ -91,9 +96,9 @@ mod tests { assert_eq!(config.channel_buffer_size, 10_000); assert!(!config.pass_raw_samples); assert_eq!(config.raw_mode_aggregation_id, 0); - assert_eq!(config.late_data_policy, LateDataPolicy::Drop); + assert_eq!(config.late_data_policy, LateDataPolicy::ForwardToStore); assert_eq!(config.wall_clock_idle_grace_period_ms, 5_000); - assert_eq!(config.wall_clock_max_open_grace_period_ms, 0); + assert_eq!(config.wall_clock_max_open_grace_period_ms, 5_000); } #[test] @@ -113,6 +118,6 @@ wall_clock_grace_period_ms: 7000 ) .expect("legacy config should deserialize"); assert_eq!(config.wall_clock_idle_grace_period_ms, 7_000); - assert_eq!(config.wall_clock_max_open_grace_period_ms, 0); + assert_eq!(config.wall_clock_max_open_grace_period_ms, 5_000); } } diff --git a/data_plane/src/precompute_engine/metrics.rs b/data_plane/src/precompute_engine/metrics.rs new file mode 100644 index 00000000..71d8dab0 --- /dev/null +++ b/data_plane/src/precompute_engine/metrics.rs @@ -0,0 +1,37 @@ +use lazy_static::lazy_static; +use prometheus::{register_counter_vec, CounterVec}; + +lazy_static! { + static ref LATE_INPUTS_TOTAL: CounterVec = register_counter_vec!( + "asap_precompute_late_inputs_total", + "Precompute inputs handled after their event-time window was late or closed", + &["action", "input_kind"] + ) + .expect("late-input counter registration must succeed"); +} + +pub fn record_late_input(action: &'static str, input_kind: &'static str) { + LATE_INPUTS_TOTAL + .with_label_values(&[action, input_kind]) + .inc(); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn late_input_actions_are_exported() { + record_late_input("drop", "raw_sample"); + let family = prometheus::gather() + .into_iter() + .find(|family| family.get_name() == "asap_precompute_late_inputs_total") + .expect("late-input counter must be registered"); + assert!(family.get_metric().iter().any(|metric| { + metric + .get_label() + .iter() + .any(|label| label.get_name() == "action" && label.get_value() == "drop") + })); + } +} diff --git a/data_plane/src/precompute_engine/mod.rs b/data_plane/src/precompute_engine/mod.rs index e97f7ecd..ce74b89b 100644 --- a/data_plane/src/precompute_engine/mod.rs +++ b/data_plane/src/precompute_engine/mod.rs @@ -2,6 +2,7 @@ pub mod accumulator_factory; pub mod config; mod engine; pub mod ingest_handler; +mod metrics; pub mod operators; pub mod output_sink; pub mod series_buffer; 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 0dc4f1ea..b0792cf3 100644 --- a/data_plane/src/precompute_engine/precompute_engine_design_doc.md +++ b/data_plane/src/precompute_engine/precompute_engine_design_doc.md @@ -146,9 +146,9 @@ pub struct PrecomputeEngineConfig { pub channel_buffer_size: usize, // default: 10,000 pub pass_raw_samples: bool, // default: false pub raw_mode_aggregation_id: u64, // default: 0 - pub late_data_policy: LateDataPolicy, // default: Drop + pub late_data_policy: LateDataPolicy, // default: ForwardToStore pub wall_clock_idle_grace_period_ms: i64, // default: 5,000 - pub wall_clock_max_open_grace_period_ms: i64, // default: 0 (disabled) + pub wall_clock_max_open_grace_period_ms: i64, // default: 5,000 } pub enum LateDataPolicy { @@ -163,6 +163,12 @@ without a touch. When enabled, the absolute deadline fires after wall-clock decisions advance only the closure watermark; they never modify the maximum observed event timestamp. +The absolute deadline is applied only with `ForwardToStore`: input arriving +after a deadline close is appended as a mergeable correction for the same +logical window. With explicit `Drop`, the absolute deadline is disabled so a +timer cannot introduce silent loss. Both actions increment +`asap_precompute_late_inputs_total{action,input_kind}`. + ### 3.3 SeriesRouter (`series_router.rs`) Deterministic hash-based routing using XXHash64: @@ -1032,19 +1038,12 @@ automatically combined with original window data at query time. ## 7. Late Data Handling -Two checks determine whether a sample is "late": - -1. **Watermark check** (sample-level): `ts < watermark - allowed_lateness_ms` → - sample is dropped entirely before reaching any aggregation logic. - -2. **Window closure check** (window-level): the sample passes the watermark check - but targets a window that is already closed - (`window not in active_windows && watermark >= window_end`). +Two checks classify an input as late: its timestamp is behind the event-time +watermark, or its target window has already closed. `LateDataPolicy` applies to +both cases and to both raw samples and prebuilt sketches: -For case 2, the `LateDataPolicy` controls behavior: - -- **Drop**: log at debug level and skip. No ghost accumulator is created - (fixing the original bug where `or_insert_with` would create orphaned entries). +- **Drop**: increment `asap_precompute_late_inputs_total{action="drop",...}`, + log at debug level, and skip. No ghost accumulator is created. - **ForwardToStore**: create a fresh `AccumulatorUpdater`, feed the single late sample, wrap as `PrecomputedOutput`, and push into the same `emit_batch` @@ -1129,7 +1128,7 @@ store with the Kafka consumer path. | `test_tumbling_window_correctness` | Samples at t=1s/5s/9s; window [0,10s) closes on t=10s; `sum=6` | | `test_sliding_window_pane_sharing` | Sample at t=15s in 30s/10s window -> 2 emits for [0,30s) and [10s,40s), both `sum=42` via shared pane snapshot/take | | `test_groupby_separate_emits_per_series` | Two series (`host=A`, `host=B`) on same worker -> 2 independent `MultipleSumAccumulator` emits (no ingest-time cross-series merge) | - | `test_late_data_drop` | Sample behind `watermark - allowed_lateness_ms` with `Drop` policy -> 0 emits | + | `test_late_data_drop` | Sample behind the event watermark with `Drop` policy -> 0 emits and records the action | | `test_late_data_forward_to_store` | Late sample for evicted pane with `ForwardToStore` -> 1 emit as mini-accumulator with correct window bounds and sum | - **Unit tests -- other modules**: `window_manager.rs` (tumbling/sliding arithmetic, pane enumeration, closure detection), `series_buffer.rs` (ordering, watermark), `accumulator_factory.rs` (updater creation and reset), `series_router.rs` (consistent hash routing), `config.rs` (defaults). @@ -1149,18 +1148,12 @@ The engine is currently in-memory and single-process with no persistence of in-f | # | Case | When it occurs | Mitigation status | |---|---|---|---| -| 1 | **Explicit late drop** | `LateDataPolicy::Drop` + `ts < watermark - allowed_lateness_ms` | Intended; use `ForwardToStore` to avoid | -| 2 | **Intra-batch lateness** | Within a single `process_samples` call, `current_wm` is set to the batch's max timestamp before pane routing; with `allowed_lateness_ms=0` every sample below the batch max is dropped | Set `allowed_lateness_ms` >= max timestamp spread within a producer batch | -| 3 | **Evicted pane + Drop** | Sample passes watermark check but its pane was already evicted (window closed); `Drop` policy discards it | Use `ForwardToStore` | -| 4 | **No matching config** | `matching_agg_configs` returns empty -- metric name in the series key does not match any config's `metric` or `spatial_filter`; worker silently returns `Ok(())` | No warning is logged. TODO: emit a metric or log at warn level for unmatched series | -| 5 | **Open panes on shutdown** | `flush_all` only emits windows already closed by the watermark; panes that are still open at shutdown are discarded | TODO (see below) | -| 6 | **Worker panic** | Tokio task dies; all series owned by that worker lose their pane state; subsequent sends log a warning and drop | TODO (see below) | - -### TODO: open-pane flush on shutdown - -`flush_all` currently only closes windows whose `end <= watermark`. On graceful shutdown it should optionally force-close all open panes by advancing each series watermark to `i64::MAX` (or to `current_wm + window_size_ms`) before the final flush. This would emit partial windows with whatever samples have accumulated, allowing downstream consumers to decide whether to use them. +| 1 | **Explicit late drop** | `LateDataPolicy::Drop` handles a late or already-closed-window input | Intended and counted; use `ForwardToStore` to retain corrections | +| 2 | **No matching config** | Metric name does not match a configured policy | Warn and expose the existing policy-miss diagnostics | +| 3 | **Worker panic** | Tokio task dies with in-memory panes | TODO: worker supervision and durable pane recovery | -This behaviour should be opt-in (a `force_flush_on_shutdown: bool` config flag) because partial windows can be misleading for consumers that expect complete windows. +On graceful shutdown, `force_close_all` emits all remaining raw and sketch +panes. These are partial windows when the event-time boundary was not reached. ### TODO: warn on unmatched series diff --git a/data_plane/src/precompute_engine/worker.rs b/data_plane/src/precompute_engine/worker.rs index b1fdd588..72828eee 100644 --- a/data_plane/src/precompute_engine/worker.rs +++ b/data_plane/src/precompute_engine/worker.rs @@ -2,6 +2,7 @@ use crate::precompute_engine::accumulator_factory::{ create_accumulator_updater, AccumulatorUpdater, }; use crate::precompute_engine::config::LateDataPolicy; +use crate::precompute_engine::metrics::record_late_input; use crate::precompute_engine::operators::sum_accumulator::SumAccumulator; use crate::precompute_engine::output_sink::OutputSink; use crate::precompute_engine::series_router::WorkerMessage; @@ -425,6 +426,7 @@ impl Worker { let window_end = pane_start + state.window_manager.window_size_ms(); match late_data_policy { LateDataPolicy::Drop => { + record_late_input("drop", "raw_sample"); debug!( "Worker {} dropping late sample for sid={} (group={}): \ ts={} observed_event_time={} pane=[{}, {})", @@ -439,6 +441,7 @@ impl Worker { continue; } LateDataPolicy::ForwardToStore => { + record_late_input("append_correction", "raw_sample"); let mut updater = create_accumulator_updater(&state.config); apply_sample(&mut *updater, series_key, *val, *ts, &state.config); let key = build_group_key_label_values(group_key); @@ -575,12 +578,14 @@ impl Worker { if too_late || pane_closed { match late_data_policy { LateDataPolicy::Drop => { + record_late_input("drop", "prebuilt_sketch"); debug!( "Worker {} dropping late accumulator input for sid={} (group={}): ts={} watermark={}", worker_id, sid, group_key, timestamp_ms, previous_event_time ); } LateDataPolicy::ForwardToStore => { + record_late_input("append_correction", "prebuilt_sketch"); let window_start = pane_start; let window_end = pane_start + state.window_manager.window_size_ms(); let key = build_group_key_label_values(group_key); @@ -808,7 +813,11 @@ impl Worker { let idle_due = idle_grace_ms > 0 && now_ms.saturating_sub(clock.last_touch_ms) >= window_size_ms.saturating_add(idle_grace_ms); - let deadline_due = max_open_grace_ms > 0 + // An absolute close can be followed by more input for the + // same event-time window. Enable it only when those inputs + // are emitted as mergeable corrections. + let deadline_due = self.late_data_policy == LateDataPolicy::ForwardToStore + && max_open_grace_ms > 0 && now_ms.saturating_sub(clock.first_touch_ms) >= window_size_ms.saturating_add(max_open_grace_ms); if idle_due || deadline_due { @@ -2693,6 +2702,7 @@ aggregations: fn make_worker_with_wall_clock_policy( agg_configs: HashMap, sink: Arc, + late_data_policy: LateDataPolicy, idle_grace_period_ms: i64, max_open_grace_period_ms: i64, ) -> Worker { @@ -2708,7 +2718,7 @@ aggregations: allowed_lateness_ms: 0, pass_raw_samples: false, raw_mode_aggregation_id: 0, - late_data_policy: LateDataPolicy::Drop, + late_data_policy, wall_clock_idle_grace_period_ms: idle_grace_period_ms, wall_clock_max_open_grace_period_ms: max_open_grace_period_ms, }, @@ -2737,7 +2747,13 @@ aggregations: let agg_configs = HashMap::from([(1, cfg)]); let sink = Arc::new(CapturingOutputSink::new()); // 5s grace period — production default. - let mut worker = make_worker_with_wall_clock_policy(agg_configs, sink.clone(), 5_000, 0); + let mut worker = make_worker_with_wall_clock_policy( + agg_configs, + sink.clone(), + LateDataPolicy::Drop, + 5_000, + 0, + ); // Pin clock at t_wall = 1_000_000 ms during ingest. Every // sketch arrives stamped with the SAME event-time @@ -2841,8 +2857,13 @@ aggregations: vec![], ); let sink = Arc::new(CapturingOutputSink::new()); - let mut worker = - make_worker_with_wall_clock_policy(HashMap::from([(7, cfg)]), sink.clone(), 5_000, 0); + let mut worker = make_worker_with_wall_clock_policy( + HashMap::from([(7, cfg)]), + sink.clone(), + LateDataPolicy::Drop, + 5_000, + 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))); @@ -2897,6 +2918,7 @@ aggregations: let mut worker = make_worker_with_wall_clock_policy( HashMap::from([(9, cfg)]), sink.clone(), + LateDataPolicy::ForwardToStore, 5_000, 5_000, ); @@ -2921,14 +2943,72 @@ aggregations: // 6.5s: longer than window_size + max_open_grace = 6s. wall_clock.store(3_006_500, Ordering::Relaxed); worker.flush_all().unwrap(); - let captured = sink.drain(); + let mut captured = sink.drain(); assert_eq!(captured.len(), 1, "absolute deadline must bound freshness"); - let sum = captured[0] - .1 + let initial = captured.pop().expect("deadline output").1; + let sum = initial .as_any() .downcast_ref::() .expect("must emit SumAccumulator"); assert_eq!(sum.sum, 28.0); + + // Continuing input for the already-closed event-time window becomes a + // mergeable correction instead of being silently dropped. + wall_clock.store(3_007_000, Ordering::Relaxed); + worker + .process_group_samples(9, pf, "", group_samples("netflow_bytes", vec![(0, 8.0)])) + .unwrap(); + let mut corrections = sink.drain(); + assert_eq!(corrections.len(), 1, "late input must emit a correction"); + let correction = corrections.pop().expect("correction output").1; + let merged = initial + .merge_with(correction.as_ref()) + .expect("deadline output and correction must merge"); + let merged_sum = merged + .as_any() + .downcast_ref::() + .expect("merged output must remain SumAccumulator"); + assert_eq!(merged_sum.sum, 36.0); + } + + #[test] + fn absolute_deadline_is_disabled_for_drop_policy() { + let cfg = make_agg_config( + 11, + "netflow_bytes", + AggregationType::SingleSubpopulation, + "Sum", + 1, + 0, + vec![], + ); + let sink = Arc::new(CapturingOutputSink::new()); + let mut worker = make_worker_with_wall_clock_policy( + HashMap::from([(11, cfg)]), + sink.clone(), + LateDataPolicy::Drop, + 0, + 5_000, + ); + let wall_clock = Arc::new(AtomicI64::new(5_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( + 11, + PolicyFingerprint(11), + "", + group_samples("netflow_bytes", vec![(0, 1.0)]), + ) + .unwrap(); + wall_clock.store(5_006_500, Ordering::Relaxed); + worker.flush_all().unwrap(); + assert_eq!( + sink.len(), + 0, + "a deadline must not close a pane when later input would be dropped" + ); } #[test] @@ -2943,8 +3023,13 @@ aggregations: vec!["zone"], ); let sink = Arc::new(CapturingOutputSink::new()); - let mut worker = - make_worker_with_wall_clock_policy(HashMap::from([(8, cfg)]), sink.clone(), 5_000, 0); + let mut worker = make_worker_with_wall_clock_policy( + HashMap::from([(8, cfg)]), + sink.clone(), + LateDataPolicy::Drop, + 5_000, + 0, + ); let wall_clock = Arc::new(AtomicI64::new(2_000_000)); let wc_clone = wall_clock.clone(); worker.set_now_ms_fn(Box::new(move || wc_clone.load(Ordering::Relaxed))); @@ -2984,6 +3069,66 @@ aggregations: assert_eq!(dd.inner.total_count(), 8); } + #[test] + fn absolute_deadline_forwards_late_sketch_correction() { + let cfg = make_agg_config( + 10, + "latency", + AggregationType::DDSketch, + "", + 1, + 0, + vec!["zone"], + ); + let sink = Arc::new(CapturingOutputSink::new()); + let mut worker = make_worker_with_wall_clock_policy( + HashMap::from([(10, cfg)]), + sink.clone(), + LateDataPolicy::ForwardToStore, + 5_000, + 5_000, + ); + let wall_clock = Arc::new(AtomicI64::new(4_000_000)); + let wc_clone = wall_clock.clone(); + worker.set_now_ms_fn(Box::new(move || wc_clone.load(Ordering::Relaxed))); + + let pf = PolicyFingerprint(10); + for i in 0..7 { + wall_clock.store(4_000_000 + i * 1_000, Ordering::Relaxed); + worker + .process_accumulator_input( + 100, + pf, + "us-east", + 0, + Box::new(make_ddsketch(0.01, &[1.0 + i as f64])), + ) + .unwrap(); + } + + wall_clock.store(4_006_500, Ordering::Relaxed); + worker.flush_all().unwrap(); + let mut deadline_outputs = sink.drain(); + assert_eq!(deadline_outputs.len(), 1); + let initial = deadline_outputs.pop().expect("deadline output").1; + + wall_clock.store(4_007_000, Ordering::Relaxed); + worker + .process_accumulator_input(100, pf, "us-east", 0, Box::new(make_ddsketch(0.01, &[8.0]))) + .unwrap(); + let mut corrections = sink.drain(); + assert_eq!(corrections.len(), 1, "late sketch must emit a correction"); + let correction = corrections.pop().expect("correction output").1; + let merged = initial + .merge_with(correction.as_ref()) + .expect("deadline output and sketch correction must merge"); + let dd = merged + .as_any() + .downcast_ref::() + .expect("merged output must remain DDSketchAccumulator"); + assert_eq!(dd.inner.total_count(), 8); + } + /// Pin the wall-clock-fallback opt-out: setting /// Disabling both wall-clock grace values preserves event-time-only /// semantics, matching pre-fix behaviour. This keeps @@ -3003,7 +3148,13 @@ aggregations: let agg_configs = HashMap::from([(1, cfg)]); let sink = Arc::new(CapturingOutputSink::new()); // grace=0 disables the fallback entirely. - let mut worker = make_worker_with_wall_clock_policy(agg_configs, sink.clone(), 0, 0); + let mut worker = make_worker_with_wall_clock_policy( + agg_configs, + sink.clone(), + LateDataPolicy::Drop, + 0, + 0, + ); let wall_clock = Arc::new(AtomicI64::new(1_000_000)); let wc_clone = wall_clock.clone(); @@ -3119,7 +3270,13 @@ aggregations: ); let agg_configs = HashMap::from([(1, cfg)]); let sink = Arc::new(CapturingOutputSink::new()); - let mut worker = make_worker_with_wall_clock_policy(agg_configs, sink.clone(), 0, 0); + let mut worker = make_worker_with_wall_clock_policy( + agg_configs, + sink.clone(), + LateDataPolicy::Drop, + 0, + 0, + ); // 10 sketches, all stamped at frozen event-time 0 → window [0, 30_000). let pf = PolicyFingerprint(1);