From 79ff2dab4bbc926015aa145640b97e4d01caa258 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 23 Jul 2026 09:26:57 -0400 Subject: [PATCH] test(precompute): add real-pipeline e2e regression test for #474 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Stacked on the wall-clock override hook. Adds JsonFileIngestConfig:: batch_delay_ms (default 0, a no-op everywhere in production and the existing netflow_single_second e2e test) — a real, explicit delay after each batch send, so an integration test can force ingest to span real wall-clock time deterministically instead of hoping incidental scheduling overhead is enough. New e2e test drives the real JsonFileIngestSource -> PrecomputeEngine -> sink pipeline with a bulk single-timestamp load spread across 10 batches, using the wall-clock override (unscaled — an earlier scaled version amplified ordinary scheduling jitter into spurious failures) and generous, seconds-scale margins so the periodic flush timer gets many real chances to land mid-ingest without the test depending on tight timing. Verified red against the pre-fix birth-time semantics (temporarily reverted, ran, confirmed rows were dropped mid-ingest, restored the fix) before counting it as a real regression test. Also fixed allowed_lateness_ms in the new test's config: every flush_all() call unconditionally nudges the event-time watermark forward by 1ms (the "boundary advance" that lets an idle stream make progress), independent of the wall-clock fallback. With allowed_lateness_ms=0 and multiple batches spaced across several flush ticks, that 1ms drift alone was enough to mark the next same-timestamp batch "late" via a wholly different code path, unrelated to #474. A generous, production-realistic value avoids that. Co-Authored-By: Claude Sonnet 5 --- asap-query-engine/src/main.rs | 1 + .../src/precompute_engine/json_ingest.rs | 16 ++ .../tests/e2e_netflow_single_second.rs | 1 + ...mpute_wall_clock_fallback_active_ingest.rs | 203 ++++++++++++++++++ 4 files changed, 221 insertions(+) create mode 100644 asap-query-engine/tests/e2e_precompute_wall_clock_fallback_active_ingest.rs diff --git a/asap-query-engine/src/main.rs b/asap-query-engine/src/main.rs index f00a113..95ec12e 100644 --- a/asap-query-engine/src/main.rs +++ b/asap-query-engine/src/main.rs @@ -263,6 +263,7 @@ async fn main() -> Result<()> { timestamp_col: timestamp_col.clone(), timestamp_unit: unit, batch_size: *batch_size, + batch_delay_ms: 0, }))] } IngestConfig::HttpRemoteWrite { port } => { diff --git a/asap-query-engine/src/precompute_engine/json_ingest.rs b/asap-query-engine/src/precompute_engine/json_ingest.rs index 598596a..121ecb6 100644 --- a/asap-query-engine/src/precompute_engine/json_ingest.rs +++ b/asap-query-engine/src/precompute_engine/json_ingest.rs @@ -42,6 +42,14 @@ pub struct JsonFileIngestConfig { pub timestamp_col: String, pub timestamp_unit: TimestampUnit, pub batch_size: usize, + /// Test-support only: real delay after sending each batch, in + /// milliseconds. `0` (the only value any production config uses) is a + /// no-op — the reader sends batches as fast as it can, unchanged from + /// before this field existed. A nonzero value exists so integration + /// tests can force ingest to span real wall-clock time deterministically + /// (e.g. to give a periodic flush timer real chances to fire mid-ingest), + /// instead of hoping incidental scheduling overhead is enough. + pub batch_delay_ms: u64, } pub struct JsonFileIngestSource { @@ -192,11 +200,19 @@ impl IngestSource for JsonFileIngestSource { if tx.blocking_send(send_batch).is_err() { break; } + if config.batch_delay_ms > 0 { + std::thread::sleep(std::time::Duration::from_millis( + config.batch_delay_ms, + )); + } } } if !batch.is_empty() { let _ = tx.blocking_send(batch); + if config.batch_delay_ms > 0 { + std::thread::sleep(std::time::Duration::from_millis(config.batch_delay_ms)); + } } Ok(row_count) diff --git a/asap-query-engine/tests/e2e_netflow_single_second.rs b/asap-query-engine/tests/e2e_netflow_single_second.rs index b4dd93e..604f64b 100644 --- a/asap-query-engine/tests/e2e_netflow_single_second.rs +++ b/asap-query-engine/tests/e2e_netflow_single_second.rs @@ -108,6 +108,7 @@ async fn netflow_single_second_batch_is_not_lost_on_shutdown() { timestamp_col: "timestamp".to_string(), timestamp_unit: TimestampUnit::Seconds, batch_size: 1024, + batch_delay_ms: 0, }; let sink = Arc::new(CapturingOutputSink::new()); diff --git a/asap-query-engine/tests/e2e_precompute_wall_clock_fallback_active_ingest.rs b/asap-query-engine/tests/e2e_precompute_wall_clock_fallback_active_ingest.rs new file mode 100644 index 0000000..550e97f --- /dev/null +++ b/asap-query-engine/tests/e2e_precompute_wall_clock_fallback_active_ingest.rs @@ -0,0 +1,203 @@ +//! End-to-end regression test for issue #474: a bulk one-shot ingest whose +//! rows all share one event-time (so the watermark never advances) can take +//! longer, in real wall-clock time, than `window_size_ms + +//! wall_clock_grace_period_ms` to fully land. Before the fix, the wall-clock +//! fallback in `flush_all` aged a pane by time-since-*first-touch*, so it +//! force-closed the window mid-ingest and every row arriving afterward hit +//! the (hardcoded-Drop) late-data path — a silent, uniform undercount. +//! +//! This drives the real `JsonFileIngestSource → PrecomputeEngine → sink` +//! path, not `Worker` directly, to catch wiring gaps the `worker.rs` unit +//! test (`wall_clock_fallback_does_not_close_a_pane_still_receiving_samples`) +//! can't — e.g. `PrecomputeEngineConfig` not actually reaching the workers. +//! That unit test remains the precise, fully deterministic proof of the fix; +//! this test is a wiring sanity check on top, and needs two pieces of +//! test-only support to be deterministic-by-construction rather than +//! best-effort: +//! +//! 1. `PrecomputeEngine::with_now_ms_fn` (a real, not `#[cfg(test)]`-gated, +//! hook) — used here with an *unscaled* real clock (no multiplier). An +//! earlier version of this test scaled real elapsed time up 50x to reach +//! a multi-second-equivalent deadline in tens of real milliseconds, but +//! that amplifies ordinary scheduling jitter right along with intentional +//! delay — a batch arriving a few tens of ms late (a common, harmless +//! hiccup under load) reads as *seconds* of simulated time, enough to +//! spuriously cross the deadline between two consecutive touches. Using +//! the real clock as-is with seconds-scale delays (below) avoids that: +//! jitter stays a small fraction of the margin instead of dominating it. +//! 2. `JsonFileIngestConfig::batch_delay_ms` — a real, explicit delay after +//! each batch send, so ingest is *guaranteed* (not merely likely, given +//! incidental scheduling overhead) to span real wall-clock time. Combined +//! with a short `flush_interval_ms`, the periodic flush timer is given +//! many real chances to land mid-ingest, not just one narrow shot. +//! +//! Both default to off/zero and are unused by every production code path. +//! +//! Timing margins (deliberately generous, not tight): +//! - deadline (window_size_ms + wall_clock_grace_period_ms) = 3000ms. +//! - batch_delay_ms = 400ms nominal per-gap, ~7.5x under the deadline — even +//! a few hundred ms of scheduling jitter on any single gap is absorbed. +//! - 10 batches x 400ms ≈ 4000ms total ingest span, comfortably past the +//! 3000ms deadline, so the old birth-time bug (were it still present) +//! would have real room to force-close mid-stream and drop later batches. +//! +//! This trades test runtime (~4-5 real seconds) for robustness against jitter. + +use std::collections::HashMap; +use std::io::Write; +use std::sync::Arc; +use std::time::Instant; + +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_size_ms: 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_size_ms, + 0, + WindowType::Tumbling, + metric.to_string(), + metric.to_string(), + None, + None, + None, + None, + ) +} + +#[tokio::test] +async fn bulk_single_timestamp_ingest_is_not_dropped_while_still_active() { + // --- Synthesize a NetFlow JSONL where every record shares one second, + // spread across enough batches to give real ingest measurable span. + let num_rows = 20; + let rows: Vec<(&str, String, String, f64)> = (0..num_rows) + .map(|i| { + ( + "2024-06-01 12:00:00", + format!("10.0.0.{}", i * 2 + 1), + format!("10.0.0.{}", i * 2 + 2), + 100.0 + i as f64, + ) + }) + .collect(); + let expected_total_bytes: f64 = rows.iter().map(|r| r.3).sum(); + + let path = std::env::temp_dir().join(format!( + "wall_clock_fallback_active_ingest_{}.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_000)); + 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: 2, // 20 rows -> 10 batches, each a separate worker touch + // Force real wall-clock separation between batches so ingest is + // *guaranteed* (not merely likely) to span the flush timer below. + // See the module doc comment for the margin reasoning. + batch_delay_ms: 400, + }; + + let sink = Arc::new(CapturingOutputSink::new()); + let sources: Vec> = vec![Box::new(JsonFileIngestSource::new(json_cfg))]; + + let engine_config = PrecomputeEngineConfig { + num_workers: 1, + // Every flush_all() call unconditionally nudges the event-time + // watermark forward by 1ms (the "+1ms boundary advance" that lets an + // idle stream make progress), independent of the wall-clock + // fallback this test targets. With multiple batches spaced across + // several flush ticks, allowed_lateness_ms=0 would make that 1ms + // drift alone mark the next same-timestamp batch "late" and drop it + // via a completely different code path — not the bug under test. + // A generous, production-realistic value (default is 5000ms) keeps + // that unrelated mechanism from firing during this test's ~4s span. + allowed_lateness_ms: 10_000, + max_buffer_per_series: 10_000, + // Real flush ticks every 50ms, many chances to land during the + // ~4000ms (10 batches x 400ms) real span of ingest. + flush_interval_ms: 50, + channel_buffer_size: 10_000, + pass_raw_samples: false, + raw_mode_aggregation_id: 0, + late_data_policy: LateDataPolicy::Drop, + // window_size_ms (1000) + this (2000) = a 3000ms deadline. + wall_clock_grace_period_ms: 2_000, + }; + + // Fake clock anchored to a real Instant, unscaled — see the module doc + // comment for why this must NOT be amplified (jitter amplifies with it). + let start = Instant::now(); + let engine = PrecomputeEngine::new(engine_config, streaming_config, sink.clone(), sources) + .with_now_ms_fn(move || 1_000_000 + start.elapsed().as_millis() as i64); + + engine.run().await.expect("engine run failed"); + + let _ = std::fs::remove_file(&path); + + // --- Verify: no row was lost to a premature force-close, whether the + // window closed via a mid-ingest flush tick or the shutdown force-close. + let captured = sink.drain(); + assert_eq!( + captured.len(), + 1, + "the single-second window must be emitted exactly once (got {} outputs)", + captured.len() + ); + + let (output, acc) = &captured[0]; + assert_eq!( + output.end_timestamp - output.start_timestamp, + 1_000, + "expected a 1s window, got [{}, {})", + output.start_timestamp, + output.end_timestamp + ); + + 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 8 rows with none dropped \ + to a premature force-close: expected {}, got {}", + expected_total_bytes, + sum_acc.sum + ); +}