From c955e38129eac1ee296165246c13dc382598eca4 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 3 Sep 2026 12:20:42 -0600 Subject: [PATCH] feat(precompute): separate idle and deadline closure --- data_plane/src/main.rs | 5 +- data_plane/src/precompute_engine/config.rs | 54 +++-- data_plane/src/precompute_engine/engine.rs | 5 +- .../precompute_engine_design_doc.md | 8 + data_plane/src/precompute_engine/worker.rs | 196 +++++++++++------- ...e2e_controller_plans_and_backend_serves.rs | 3 +- .../tests/e2e_modified_otlp_sketch_path.rs | 3 +- 7 files changed, 182 insertions(+), 92 deletions(-) diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs index c69d15f4..f13c65d2 100644 --- a/data_plane/src/main.rs +++ b/data_plane/src/main.rs @@ -513,7 +513,10 @@ async fn main() -> Result<()> { pass_raw_samples: false, raw_mode_aggregation_id: 0, late_data_policy: LateDataPolicy::Drop, - wall_clock_grace_period_ms: 5_000, + 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, 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 93e14ffe..5c803ec1 100644 --- a/data_plane/src/precompute_engine/config.rs +++ b/data_plane/src/precompute_engine/config.rs @@ -30,18 +30,20 @@ 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. agents stamp every - /// sketch with the same `time_unix_nano`), `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". Default: 5000 ms (matches - /// `allowed_lateness_ms` default). - #[serde(default = "default_wall_clock_grace_period_ms")] - pub wall_clock_grace_period_ms: i64, + /// Additional idle grace after one window duration. A pane closes when it + /// has received no input for `window_size + idle_grace`. The legacy YAML + /// name is accepted as a serde alias. Non-positive disables idle closure. + #[serde( + default = "default_wall_clock_idle_grace_period_ms", + alias = "wall_clock_grace_period_ms" + )] + 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)] + 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 /// set, the registry loads prior `created_at_ms` / `retired_at_ms` @@ -64,13 +66,14 @@ 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(), + wall_clock_idle_grace_period_ms: default_wall_clock_idle_grace_period_ms(), + wall_clock_max_open_grace_period_ms: 0, schema_persist_path: None, } } } -fn default_wall_clock_grace_period_ms() -> i64 { +fn default_wall_clock_idle_grace_period_ms() -> i64 { 5_000 } @@ -89,6 +92,27 @@ 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); + assert_eq!(config.wall_clock_idle_grace_period_ms, 5_000); + assert_eq!(config.wall_clock_max_open_grace_period_ms, 0); + } + + #[test] + fn legacy_wall_clock_grace_deserializes_as_idle_grace() { + let config: PrecomputeEngineConfig = serde_yaml::from_str( + r#" +num_workers: 1 +allowed_lateness_ms: 100 +max_buffer_per_series: 10 +flush_interval_ms: 1000 +channel_buffer_size: 10 +pass_raw_samples: false +raw_mode_aggregation_id: 0 +late_data_policy: Drop +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); } } diff --git a/data_plane/src/precompute_engine/engine.rs b/data_plane/src/precompute_engine/engine.rs index c1edf95b..ad2dbb64 100644 --- a/data_plane/src/precompute_engine/engine.rs +++ b/data_plane/src/precompute_engine/engine.rs @@ -134,7 +134,10 @@ 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, + wall_clock_idle_grace_period_ms: self.config.wall_clock_idle_grace_period_ms, + wall_clock_max_open_grace_period_ms: self + .config + .wall_clock_max_open_grace_period_ms, }, self.diagnostics.worker_group_counts[id].clone(), self.diagnostics.worker_watermarks[id].clone(), 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 cfa301ea..0dc4f1ea 100644 --- a/data_plane/src/precompute_engine/precompute_engine_design_doc.md +++ b/data_plane/src/precompute_engine/precompute_engine_design_doc.md @@ -147,6 +147,8 @@ pub struct PrecomputeEngineConfig { pub pass_raw_samples: bool, // default: false pub raw_mode_aggregation_id: u64, // default: 0 pub late_data_policy: LateDataPolicy, // default: Drop + pub wall_clock_idle_grace_period_ms: i64, // default: 5,000 + pub wall_clock_max_open_grace_period_ms: i64, // default: 0 (disabled) } pub enum LateDataPolicy { @@ -155,6 +157,12 @@ pub enum LateDataPolicy { } ``` +For a window of duration `W`, idle closure fires after `W + idle_grace` +without a touch. When enabled, the absolute deadline fires after +`W + max_open_grace` from the first touch even if input remains active. These +wall-clock decisions advance only the closure watermark; they never modify the +maximum observed event timestamp. + ### 3.3 SeriesRouter (`series_router.rs`) Deterministic hash-based routing using XXHash64: diff --git a/data_plane/src/precompute_engine/worker.rs b/data_plane/src/precompute_engine/worker.rs index d8a8534f..b1fdd588 100644 --- a/data_plane/src/precompute_engine/worker.rs +++ b/data_plane/src/precompute_engine/worker.rs @@ -64,27 +64,32 @@ struct GroupState { /// 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 - /// stagnated. Refreshing on every touch prevents an active bulk ingest - /// with a fixed event timestamp from being force-closed mid-stream. - /// Keyed by `pane_start_ms` and covers entries in BOTH - /// `active_panes` and `sketch_panes` (a single pane_start may have - /// either or both populated). Entries are GC'd by - /// `prune_pane_wall_clock_last_touches` after each window-close cycle. - pane_wall_clock_last_touches_ms: BTreeMap, + /// First and latest wall-clock input times for each open pane. The former + /// enforces an absolute lifetime; the latter supports idle closure. + pane_wall_clock: BTreeMap, +} + +#[derive(Clone, Copy)] +struct PaneWallClock { + first_touch_ms: i64, + last_touch_ms: i64, } impl GroupState { - /// Drop wall-clock-last-touch entries whose pane no longer exists in - /// either pane map. Called after window-close cycles in - /// `process_group_samples`, `process_accumulator_input`, and - /// `flush_all` so the bookkeeping doesn't leak as panes turn over. - fn prune_pane_wall_clock_last_touches(&mut self) { + fn touch_pane(&mut self, pane_start_ms: i64, now_ms: i64) { + self.pane_wall_clock + .entry(pane_start_ms) + .and_modify(|clock| clock.last_touch_ms = now_ms) + .or_insert(PaneWallClock { + first_touch_ms: now_ms, + last_touch_ms: now_ms, + }); + } + + fn prune_pane_wall_clock(&mut self) { let active = &self.active_panes; let sketch = &self.sketch_panes; - self.pane_wall_clock_last_touches_ms + self.pane_wall_clock .retain(|ps, _| active.contains_key(ps) || sketch.contains_key(ps)); } } @@ -96,10 +101,8 @@ 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, + pub wall_clock_idle_grace_period_ms: i64, + pub wall_clock_max_open_grace_period_ms: i64, } /// Worker that processes samples for a shard of the sid space. @@ -135,9 +138,8 @@ pub struct Worker { worker_watermark: Arc, /// 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, + wall_clock_idle_grace_period_ms: i64, + wall_clock_max_open_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. The closure runs under @@ -164,7 +166,8 @@ impl Worker { pass_raw_samples, raw_mode_aggregation_id, late_data_policy, - wall_clock_grace_period_ms, + wall_clock_idle_grace_period_ms, + wall_clock_max_open_grace_period_ms, } = runtime_config; Self { id, @@ -178,7 +181,8 @@ impl Worker { late_data_policy, worker_watermark, group_count, - wall_clock_grace_period_ms, + wall_clock_idle_grace_period_ms, + wall_clock_max_open_grace_period_ms, now_ms_fn: Box::new(default_now_ms), } } @@ -349,7 +353,7 @@ impl Worker { sketch_panes: BTreeMap::new(), max_event_time_ms: i64::MIN, closure_watermark_ms: i64::MIN, - pane_wall_clock_last_touches_ms: BTreeMap::new(), + pane_wall_clock: BTreeMap::new(), }; self.group_states.insert(sid, gs); self.group_count @@ -458,9 +462,7 @@ impl Worker { // Refresh the pane's wall-clock last-touch time so the fallback // only closes an idle pane, not a long-running bulk ingest whose // records share one event timestamp. - state - .pane_wall_clock_last_touches_ms - .insert(pane_start, now_ms); + state.touch_pane(pane_start, now_ms); let updater = state .active_panes .entry(pane_start) @@ -495,7 +497,7 @@ impl Worker { if event_watermark > state.closure_watermark_ms { state.closure_watermark_ms = event_watermark; } - state.prune_pane_wall_clock_last_touches(); + state.prune_pane_wall_clock(); // Emit to output sink if !emit_batch.is_empty() { @@ -602,9 +604,7 @@ impl Worker { // Refresh the pane's wall-clock last-touch time so an active sketch // stream with a fixed event timestamp is not force-closed mid-ingest. - state - .pane_wall_clock_last_touches_ms - .insert(pane_start, now_ms); + state.touch_pane(pane_start, now_ms); // Merge into the sketch pane covering this timestamp. match state.sketch_panes.remove(&pane_start) { @@ -660,7 +660,7 @@ impl Worker { if event_watermark > state.closure_watermark_ms { state.closure_watermark_ms = event_watermark; } - state.prune_pane_wall_clock_last_touches(); + state.prune_pane_wall_clock(); if !emit_batch.is_empty() { debug!( @@ -766,7 +766,8 @@ impl Worker { } let now_ms = (self.now_ms_fn)(); - let grace_ms = self.wall_clock_grace_period_ms; + let idle_grace_ms = self.wall_clock_idle_grace_period_ms; + let max_open_grace_ms = self.wall_clock_max_open_grace_period_ms; // Publish the largest observed event-time watermark for diagnostics. // It must not be fed back into another group's closure decision: group @@ -798,20 +799,19 @@ impl Worker { // 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), 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 - // `pane_start + window_size_ms` for any pane untouched for - // `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 { + // Wall-clock policy is independent of observed event time. Idle + // closure handles completed finite input; the optional first-touch + // deadline bounds freshness even for a continuously touched pane. + if idle_grace_ms > 0 || max_open_grace_ms > 0 { let window_size_ms = state.window_manager.window_size_ms(); - for (&pane_start, &pane_last_touch_ms) in &state.pane_wall_clock_last_touches_ms { - if now_ms.saturating_sub(pane_last_touch_ms) >= window_size_ms + grace_ms { + for (&pane_start, &clock) in &state.pane_wall_clock { + 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 + && now_ms.saturating_sub(clock.first_touch_ms) + >= window_size_ms.saturating_add(max_open_grace_ms); + if idle_due || deadline_due { let force_to = pane_start.saturating_add(window_size_ms); if force_to > effective_wm { effective_wm = force_to; @@ -859,7 +859,7 @@ impl Worker { state.closure_watermark_ms = effective_wm; } - state.prune_pane_wall_clock_last_touches(); + state.prune_pane_wall_clock(); } if !emit_batch.is_empty() { @@ -958,7 +958,7 @@ impl Worker { if force_wm > state.closure_watermark_ms { state.closure_watermark_ms = force_wm; } - state.prune_pane_wall_clock_last_touches(); + state.prune_pane_wall_clock(); } if !emit_batch.is_empty() { @@ -1455,7 +1455,8 @@ 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, + wall_clock_idle_grace_period_ms: 0, + wall_clock_max_open_grace_period_ms: 0, }, Arc::new(AtomicUsize::new(0)), wm, @@ -1982,7 +1983,8 @@ mod tests { pass_raw_samples: false, raw_mode_aggregation_id: 0, late_data_policy: LateDataPolicy::Drop, - wall_clock_grace_period_ms: 0, + wall_clock_idle_grace_period_ms: 0, + wall_clock_max_open_grace_period_ms: 0, }, Arc::new(AtomicUsize::new(0)), wm, @@ -2035,7 +2037,8 @@ mod tests { pass_raw_samples: false, raw_mode_aggregation_id: 0, late_data_policy: LateDataPolicy::ForwardToStore, - wall_clock_grace_period_ms: 0, + wall_clock_idle_grace_period_ms: 0, + wall_clock_max_open_grace_period_ms: 0, }, Arc::new(AtomicUsize::new(0)), wm, @@ -2427,7 +2430,8 @@ aggregations: pass_raw_samples: false, raw_mode_aggregation_id: 0, late_data_policy: LateDataPolicy::Drop, - wall_clock_grace_period_ms: 0, + wall_clock_idle_grace_period_ms: 0, + wall_clock_max_open_grace_period_ms: 0, }, Arc::new(AtomicUsize::new(0)), wm.clone(), @@ -2685,14 +2689,12 @@ aggregations: // `std::thread::sleep(35s)`. // ----------------------------------------------------------------------- - /// Build a Worker for the wall-clock fallback test that is otherwise - /// identical to `make_worker`, but takes an explicit - /// `wall_clock_grace_period_ms`. Test-local helper so existing - /// callsites stay untouched. - fn make_worker_with_grace( + /// Build a worker with explicit wall-clock closure grace values. + fn make_worker_with_wall_clock_policy( agg_configs: HashMap, sink: Arc, - wall_clock_grace_period_ms: i64, + idle_grace_period_ms: i64, + max_open_grace_period_ms: i64, ) -> Worker { let (_tx, rx) = tokio::sync::mpsc::channel(1); let wm = Arc::new(AtomicI64::new(i64::MIN)); @@ -2703,15 +2705,12 @@ aggregations: make_hot_reload(agg_configs), WorkerRuntimeConfig { max_buffer_per_series: 10_000, - // `flush_all` advances a stagnant event-time watermark by - // 1ms. Permit that one boundary nudge so a subsequent - // same-timestamp touch exercises wall-clock idleness rather - // than the unrelated late-data path. - allowed_lateness_ms: 1, + allowed_lateness_ms: 0, pass_raw_samples: false, raw_mode_aggregation_id: 0, late_data_policy: LateDataPolicy::Drop, - wall_clock_grace_period_ms, + wall_clock_idle_grace_period_ms: idle_grace_period_ms, + wall_clock_max_open_grace_period_ms: max_open_grace_period_ms, }, Arc::new(AtomicUsize::new(0)), wm, @@ -2738,7 +2737,7 @@ 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_grace(agg_configs, sink.clone(), 5_000); + let mut worker = make_worker_with_wall_clock_policy(agg_configs, sink.clone(), 5_000, 0); // Pin clock at t_wall = 1_000_000 ms during ingest. Every // sketch arrives stamped with the SAME event-time @@ -2818,7 +2817,7 @@ aggregations: // Calling flush_all again with no new data must NOT re-emit // the same window — once a window has been closed via the // fallback, its pane is drained from `sketch_panes` and from - // `pane_wall_clock_last_touches_ms`, so there's nothing left to + // `pane_wall_clock`, so there's nothing left to // close. This pins the monotonicity invariant: emitted // windows have monotonically non-decreasing close times and // each window is emitted at most once. @@ -2842,7 +2841,8 @@ aggregations: vec![], ); let sink = Arc::new(CapturingOutputSink::new()); - let mut worker = make_worker_with_grace(HashMap::from([(7, cfg)]), sink.clone(), 5_000); + let mut worker = + make_worker_with_wall_clock_policy(HashMap::from([(7, cfg)]), sink.clone(), 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))); @@ -2882,6 +2882,55 @@ aggregations: assert!((sum.sum - expected_sum).abs() < 1e-9); } + #[test] + fn absolute_wall_clock_deadline_closes_active_raw_ingest() { + let cfg = make_agg_config( + 9, + "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([(9, cfg)]), + sink.clone(), + 5_000, + 5_000, + ); + let wall_clock = Arc::new(AtomicI64::new(3_000_000)); + let wc_clone = wall_clock.clone(); + worker.set_now_ms_fn(Box::new(move || wc_clone.load(Ordering::Relaxed))); + + let pf = PolicyFingerprint(9); + for i in 0..7 { + wall_clock.store(3_000_000 + i * 1_000, Ordering::Relaxed); + worker + .process_group_samples( + 9, + pf, + "", + group_samples("netflow_bytes", vec![(0, 1.0 + i as f64)]), + ) + .unwrap(); + } + + // The last touch was only 500ms ago, but the pane has been open for + // 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(); + assert_eq!(captured.len(), 1, "absolute deadline must bound freshness"); + let sum = captured[0] + .1 + .as_any() + .downcast_ref::() + .expect("must emit SumAccumulator"); + assert_eq!(sum.sum, 28.0); + } + #[test] fn wall_clock_fallback_does_not_close_active_sketch_ingest() { let cfg = make_agg_config( @@ -2894,7 +2943,8 @@ aggregations: vec!["zone"], ); let sink = Arc::new(CapturingOutputSink::new()); - let mut worker = make_worker_with_grace(HashMap::from([(8, cfg)]), sink.clone(), 5_000); + let mut worker = + make_worker_with_wall_clock_policy(HashMap::from([(8, cfg)]), sink.clone(), 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))); @@ -2935,7 +2985,7 @@ aggregations: } /// Pin the wall-clock-fallback opt-out: setting - /// `wall_clock_grace_period_ms = 0` reverts to event-time-only + /// Disabling both wall-clock grace values preserves event-time-only /// semantics, matching pre-fix behaviour. This keeps /// `flush_all`'s contract backward-compatible for callers that /// want strict event-time (e.g. deterministic replays). @@ -2953,7 +3003,7 @@ 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_grace(agg_configs, sink.clone(), 0); + let mut worker = make_worker_with_wall_clock_policy(agg_configs, sink.clone(), 0, 0); let wall_clock = Arc::new(AtomicI64::new(1_000_000)); let wc_clone = wall_clock.clone(); @@ -3069,7 +3119,7 @@ aggregations: ); let agg_configs = HashMap::from([(1, cfg)]); let sink = Arc::new(CapturingOutputSink::new()); - let mut worker = make_worker_with_grace(agg_configs, sink.clone(), 0); + let mut worker = make_worker_with_wall_clock_policy(agg_configs, sink.clone(), 0, 0); // 10 sketches, all stamped at frozen event-time 0 → window [0, 30_000). let pf = PolicyFingerprint(1); diff --git a/data_plane/tests/e2e_controller_plans_and_backend_serves.rs b/data_plane/tests/e2e_controller_plans_and_backend_serves.rs index cd828744..13d2e921 100644 --- a/data_plane/tests/e2e_controller_plans_and_backend_serves.rs +++ b/data_plane/tests/e2e_controller_plans_and_backend_serves.rs @@ -313,7 +313,8 @@ async fn start_full_stack(otlp_http_port: u16, otlp_grpc_port: u16) -> FullStack pass_raw_samples: false, raw_mode_aggregation_id: 0, late_data_policy: LateDataPolicy::Drop, - wall_clock_grace_period_ms: 5_000, + wall_clock_idle_grace_period_ms: 5_000, + wall_clock_max_open_grace_period_ms: 0, schema_persist_path: None, }; let engine = PrecomputeEngine::new( diff --git a/data_plane/tests/e2e_modified_otlp_sketch_path.rs b/data_plane/tests/e2e_modified_otlp_sketch_path.rs index fa2ed1a4..8f5e3c74 100644 --- a/data_plane/tests/e2e_modified_otlp_sketch_path.rs +++ b/data_plane/tests/e2e_modified_otlp_sketch_path.rs @@ -101,7 +101,8 @@ fn engine_config() -> PrecomputeEngineConfig { pass_raw_samples: false, raw_mode_aggregation_id: 0, late_data_policy: LateDataPolicy::Drop, - wall_clock_grace_period_ms: 5_000, + wall_clock_idle_grace_period_ms: 5_000, + wall_clock_max_open_grace_period_ms: 0, schema_persist_path: None, } }