fix(precompute): close idle/trailing windows for one-shot batch ingest (wall-clock fallback + shutdown force-close) (#400)

* fix(precompute): port wall-clock grace fallback from backend to close idle windows

Sync the precompute-engine window-closing fix from ASAPQuery-backend.

Problem: with strict event-time semantics, a tumbling window [T, T+size)
only closes when the watermark reaches T+size. For a one-shot batch ingest
where every record carries (nearly) the same timestamp, the watermark
freezes and flush_all's +1ms advance is a no-op, so the trailing window
never closes — emit_batch is never called and the store stays empty even
though data was ingested.

Fix (ported from the backend's flush_all "sweep blocker #2" fallback):
track each pane's wall-clock birth time and, in flush_all, force the
effective watermark past pane_start + window_size_ms for any pane that has
been alive longer than window_size_ms + wall_clock_grace_period_ms of
wall-clock time, regardless of event-time. Set wall_clock_grace_period_ms
<= 0 to opt out and keep strict event-time-only semantics.

Changes:
- config: add wall_clock_grace_period_ms (serde default 5000ms)
- worker: GroupState.pane_wall_clock_starts_ms + prune; injectable now_ms_fn
  (default SystemTime::now, test override via set_now_ms_fn); record pane
  birth in process_group_samples; wall-clock fallback in flush_all
- engine/main/CLI: thread the setting through (PrecomputeSettings +
  --wall-clock-grace-period-ms flag)
- tests: wall_clock_fallback_closes_idle_window and the grace=0 opt-out;
  update existing config literals

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(precompute): force-close open windows on worker shutdown

Follow-up to the wall-clock grace fallback. The fallback only fires once a
pane has aged window_size + grace of wall-clock time on a flush tick, so a
one-shot batch that ingests and immediately shuts down still leaves its
trailing window open — flush_all's +1ms advance never reaches the window
end and the data never lands in the store.

Add Worker::force_close_all(), invoked from the Shutdown handler after the
final flush_all(): for every group with open panes, advance to a finite
bound (max_pane + window_size_ms) and emit every remaining window. A finite
bound is used deliberately — WindowManager::closed_windows enumerates window
starts one slide at a time, so passing i64::MAX would loop ~i64::MAX/slide
times and overflow. The pass is idempotent: drained panes are removed and
their wall-clock bookkeeping pruned.

- worker: force_close_all() + Shutdown handler wiring
- tests: shutdown_force_close_emits_trailing_window; update
  test_update_agg_configs_enables_new_aggregation_at_runtime to assert the
  trailing window [10_000, 20_000) is now emitted (previously lost)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* test(precompute): e2e regression for one-shot NetFlow single-second ingest

Drives the real JsonFileIngestSource -> PrecomputeEngine -> sink path with a
synthesized NetFlow JSONL whose records all fall in the same second. Sets
wall_clock_grace_period_ms=0 so only the shutdown force-close can close the
trailing window — a non-empty sink proves the force-close rescues the batch.

Verified both directions: with the Shutdown force-close the single-second
window is emitted with the summed bytes; with it disabled the run produces 0
outputs (reproducing the reported empty-store bug).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* style: cargo fmt for precompute worker tests and netflow e2e

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(precompute): use finite bound in UpdateAggConfigs removed-agg force-close

Addresses review feedback on #400. The removed-agg-id cleanup in the
UpdateAggConfigs handler advanced the watermark to i64::MAX, the exact hazard
force_close_all's docstring warns about: closed_windows enumerates window
starts one slide at a time, so with a real epoch-ms watermark it loops
~i64::MAX/slide times and overflows `start + window_size_ms` (panics in
debug). Use the same finite bound as force_close_all (max_pane +
window_size_ms).

Add test_update_agg_configs_removed_id_force_closes_at_epoch_ms_timestamps,
which opens a window at a realistic epoch-ms timestamp and removes the
agg_id; it hangs/overflows under the old i64::MAX code (verified: times out)
and passes with the finite bound.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: zzylol

asap-query-engine/src/bin/bench_precompute_sketch.rs

@@ -175,6 +175,7 @@ async fn start_engine(
         pass_raw_samples: false,
         raw_mode_aggregation_id: 0,
         late_data_policy: LateDataPolicy::Drop,
+        wall_clock_grace_period_ms: 5_000,
     };
     let sources: Vec<Box<dyn IngestSource>> =
         vec![Box::new(HttpIngestSource::new(HttpIngestConfig { port }))];

asap-query-engine/src/bin/e2e_quickstart_resource_test.rs

@@ -234,6 +234,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         pass_raw_samples: false,
         raw_mode_aggregation_id: 0,
         late_data_policy: LateDataPolicy::Drop,
+        wall_clock_grace_period_ms: 5_000,
     };
     let output_sink = Arc::new(StoreOutputSink::new(store.clone()));
     let sources: Vec<Box<dyn IngestSource>> =

asap-query-engine/src/bin/precompute_engine.rs

@@ -67,6 +67,11 @@ struct Args {
     #[arg(long, value_enum, default_value_t = LateDataPolicy::Drop)]
     late_data_policy: LateDataPolicy,
 
+    /// Wall-clock grace period (ms) for the flush fallback that force-closes
+    /// idle windows when event-time stagnates. Set to <= 0 to disable.
+    #[arg(long, default_value_t = 5000)]
+    wall_clock_grace_period_ms: i64,
+
     // --- CSV file ingest (alternative to HTTP) ---
     /// Path to a local CSV file to ingest instead of listening for HTTP writes
     #[arg(long)]
@@ -170,6 +175,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         pass_raw_samples: args.pass_raw_samples,
         raw_mode_aggregation_id: args.raw_mode_aggregation_id,
         late_data_policy: args.late_data_policy,
+        wall_clock_grace_period_ms: args.wall_clock_grace_period_ms,
     };
 
     // Create the output sink (writes directly to the store)

asap-query-engine/src/bin/test_e2e_precompute.rs

@@ -151,6 +151,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         pass_raw_samples: false,
         raw_mode_aggregation_id: 0,
         late_data_policy: LateDataPolicy::Drop,
+        wall_clock_grace_period_ms: 5000,
     };
     let output_sink = Arc::new(StoreOutputSink::new(store.clone()));
     let sources: Vec<Box<dyn IngestSource>> =
@@ -299,6 +300,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         pass_raw_samples: true,
         raw_mode_aggregation_id: raw_agg_id,
         late_data_policy: LateDataPolicy::Drop,
+        wall_clock_grace_period_ms: 5000,
     };
     let raw_sink = Arc::new(RawPassthroughSink::new(store.clone()));
     let raw_sources: Vec<Box<dyn IngestSource>> =
@@ -655,6 +657,7 @@ async fn run_single_bench(
         pass_raw_samples: false,
         raw_mode_aggregation_id: 0,
         late_data_policy: LateDataPolicy::Drop,
+        wall_clock_grace_period_ms: 5000,
     };
     let sources: Vec<Box<dyn IngestSource>> =
         vec![Box::new(HttpIngestSource::new(HttpIngestConfig { port }))];

asap-query-engine/src/engine_config.rs

@@ -310,6 +310,12 @@ pub struct PrecomputeSettings {
     pub flush_interval_ms: u64,
     pub channel_buffer_size: usize,
     pub dump_precomputes: bool,
+    /// Wall-clock grace period (ms) for the flush fallback that force-closes
+    /// idle windows when event-time stagnates (e.g. one-shot batch ingest
+    /// where every record shares a timestamp). Set to <= 0 to disable and
+    /// keep strict event-time-only semantics. See
+    /// `PrecomputeEngineConfig::wall_clock_grace_period_ms`.
+    pub wall_clock_grace_period_ms: i64,
 }
 
 impl Default for PrecomputeSettings {
@@ -321,6 +327,7 @@ impl Default for PrecomputeSettings {
             flush_interval_ms: 1000,
             channel_buffer_size: 10000,
             dump_precomputes: false,
+            wall_clock_grace_period_ms: 5000,
         }
     }
 }

asap-query-engine/src/main.rs

@@ -209,6 +209,7 @@ async fn main() -> Result<()> {
             pass_raw_samples: false,
             raw_mode_aggregation_id: 0,
             late_data_policy: LateDataPolicy::Drop,
+            wall_clock_grace_period_ms: config.precompute_engine.wall_clock_grace_period_ms,
         };
         let output_sink = Arc::new(StoreOutputSink::new(store.clone()));
         let sources: Vec<Box<dyn IngestSource>> = match &config.ingest {

asap-query-engine/src/precompute_engine/config.rs

@@ -30,6 +30,19 @@ 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. a one-shot batch where
+    /// every record carries the same timestamp), `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". Set to `<= 0` to opt out and keep strict
+    /// event-time-only semantics. Default: 5000 ms (matches
+    /// `allowed_lateness_ms` default).
+    #[serde(default = "default_wall_clock_grace_period_ms")]
+    pub wall_clock_grace_period_ms: i64,
 }
 
 impl Default for PrecomputeEngineConfig {
@@ -43,10 +56,15 @@ 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(),
         }
     }
 }
 
+fn default_wall_clock_grace_period_ms() -> i64 {
+    5_000
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -62,5 +80,6 @@ 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);
     }
 }

asap-query-engine/src/precompute_engine/engine.rs

@@ -171,6 +171,7 @@ 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,
                 },
                 self.diagnostics.worker_group_counts[id].clone(),
                 self.diagnostics.worker_watermarks[id].clone(),