From cfe0dfb6dea6f80d1b7770accde8fd67f9d40ac2 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 23 Jul 2026 09:26:57 -0400 Subject: [PATCH 1/3] 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 + ); +} From 209aceee87c17e85a87ec19de9499e4cdb576997 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Tue, 18 Aug 2026 11:39:12 -0400 Subject: [PATCH 2/3] docs(precompute): address PR #538 review feedback on framing Two review comments from @zzylol on #538: - batch_delay_ms's doc comment called it "test-support only," but it lives on JsonFileIngestConfig, a real production struct, and nothing about it is actually test-gated -- it's an ordinary throttling knob that just has no caller setting it nonzero yet. Reworded to describe what it does, not who currently uses it. - The new e2e test's module doc led with "regression test for issue #474." Reframed to lead with the behavior/property under test (bulk single- timestamp ingest must not lose data while still actively receiving samples), keeping the causal explanation of the prior bug as supporting context rather than the test's identity. Co-Authored-By: Claude Sonnet 5 --- .../src/precompute_engine/json_ingest.rs | 11 ++-- ...mpute_wall_clock_fallback_active_ingest.rs | 57 ++++++++++--------- 2 files changed, 35 insertions(+), 33 deletions(-) diff --git a/asap-query-engine/src/precompute_engine/json_ingest.rs b/asap-query-engine/src/precompute_engine/json_ingest.rs index 121ecb6..1d70e05 100644 --- a/asap-query-engine/src/precompute_engine/json_ingest.rs +++ b/asap-query-engine/src/precompute_engine/json_ingest.rs @@ -42,13 +42,10 @@ 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. + /// Delay after sending each batch, in milliseconds. `0` (the default) + /// sends batches as fast as possible, unchanged from before this field + /// existed. A nonzero value throttles ingestion to a real, deterministic + /// pace. pub batch_delay_ms: u64, } 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 index 550e97f..50ae5b8 100644 --- 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 @@ -1,37 +1,42 @@ -//! 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. +//! End-to-end test: JSON file ingestion of a bulk one-shot batch whose rows +//! all share one event-time (so the watermark never advances) must not lose +//! rows even when ingestion takes longer, in real wall-clock time, than +//! `window_size_ms + wall_clock_grace_period_ms`. The wall-clock fallback in +//! `flush_all` exists to force-close a window when event-time stagnates, but +//! it must only do so once a pane has gone genuinely idle — never while a +//! pane is still actively receiving samples, however long it's been open. +//! (A prior bug aged a pane by time-since-*first-touch* instead of +//! time-since-*last-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 +//! That unit test remains the precise, fully deterministic proof of the +//! fallback's behavior; this test is a wiring sanity check on top, and +//! relies on two things 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. +//! test-only 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` — an ordinary throttling knob +//! (not test-only; it just has no other caller yet) set here to 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. //! //! Timing margins (deliberately generous, not tight): //! - deadline (window_size_ms + wall_clock_grace_period_ms) = 3000ms. From 7ae3b2911476e8e0e65d780aea99c510d1df77c1 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Tue, 18 Aug 2026 12:05:49 -0400 Subject: [PATCH 3/3] docs(precompute): trim the e2e test's module doc comment Co-Authored-By: Claude Sonnet 5 --- ...mpute_wall_clock_fallback_active_ingest.rs | 25 +++---------------- 1 file changed, 4 insertions(+), 21 deletions(-) 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 index 50ae5b8..1911727 100644 --- 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 @@ -13,14 +13,9 @@ //! 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 -//! fallback's behavior; this test is a wiring sanity check on top, and -//! relies on two things to be deterministic-by-construction rather than -//! best-effort: +//! can't //! -//! 1. `PrecomputeEngine::with_now_ms_fn` (a real, not `#[cfg(test)]`-gated, -//! test-only hook) — used here with an *unscaled* real clock (no +//! 1. `PrecomputeEngine::with_now_ms_fn` — 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 @@ -31,20 +26,8 @@ //! delays (below) avoids that: jitter stays a small fraction of the //! margin instead of dominating it. //! 2. `JsonFileIngestConfig::batch_delay_ms` — an ordinary throttling knob -//! (not test-only; it just has no other caller yet) set here to 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. -//! -//! 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. +//! set here to a real, explicit delay after each batch send, so ingest is *guaranteed* to span real +//! wall-clock time. //! //! This trades test runtime (~4-5 real seconds) for robustness against jitter.