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>

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(),