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

[zzylol]

Syncs the precompute-engine window-closing fixes from ASAPQuery-backend into ASAPQuery, and adds a shutdown force-close so a one-shot batch can never lose its trailing window.

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 — e.g. NetFlow records that all fall in the same second — the per-group watermark freezes at max_ts < T+size. flush_all only advances by +1ms, so closed_windows(prev, prev+1) returns empty forever: the trailing window never closes, emit_batch is never called, and the store stays empty even though data was ingested.

Commit 1 — port the wall-clock grace fallback (backend → frontend)

Track each pane's wall-clock birth time; in flush_all, force the effective watermark past pane_start + window_size_ms for any pane alive longer than window_size_ms + wall_clock_grace_period_ms of wall-clock time, regardless of event-time. wall_clock_grace_period_ms <= 0 opts out (strict event-time only). Default 5000ms, matching the backend.

• config: add wall_clock_grace_period_ms (serde default 5000)
• worker: GroupState.pane_wall_clock_starts_ms + prune; injectable now_ms_fn (set_now_ms_fn for tests); record pane birth in process_group_samples; fallback block in flush_all
• engine/main/CLI: thread the setting through (PrecomputeSettings + --wall-clock-grace-period-ms)

Commit 2 — force-close open windows on shutdown

The fallback is wall-clock gated, so a batch that ingests and immediately shuts down hasn't aged past the grace period and still leaves its trailing window open. Worker::force_close_all() (called from the Shutdown handler after the final flush_all) closes every remaining open window unconditionally, since no further samples will arrive.

It advances to a finite bound (max_pane + window_size_ms), not i64::MAX: WindowManager::closed_windows enumerates window starts one slide at a time, so i64::MAX would loop ~i64::MAX/slide times and overflow. Idempotent — drained panes are removed and their bookkeeping pruned.

Tests

• wall_clock_fallback_closes_idle_window, wall_clock_fallback_disabled_preserves_event_time_only_semantics
• shutdown_force_close_emits_trailing_window
• test_update_agg_configs_enables_new_aggregation_at_runtime updated: the trailing window [10_000, 20_000) is now emitted on shutdown (previously silently lost)

Verification

• cargo check --lib --tests --bins — clean
• precompute_engine::worker — 28 passed · precompute_engine::config — passed
• engine_config — 20 passed · e2e_precompute_equivalence — 2 passed

🤖 Generated with Claude Code

[milindsrivastava1997]

The PR's own docstring for force_close_all explains why it avoids i64::MAX (closed_windows would loop ~i64::MAX/slidetimes and overflow/hang). But there's a sibling code path with that exact bug, untouched by this PR: in theUpdateAggConfigs handler (worker.rs:206, pre-existing, not in this diff), the removed-agg-id cleanup calls:
let closed = state.window_manager.closed_windows(state.previous_watermark_ms, i64::MAX);

[zzylol]

Good catch — fixed in 23e2bde.

The UpdateAggConfigs removed-agg cleanup (worker.rs:206) now uses the same finite bound as force_close_all instead of i64::MAX:

let closed = match active_panes.keys().next_back() {
    Some(&max_pane) => {
        let force_wm = max_pane.saturating_add(state.window_manager.window_size_ms());
        state.window_manager.closed_windows(state.previous_watermark_ms, force_wm)
    }
    None => Vec::new(), // no open panes
};

I also added test_update_agg_configs_removed_id_force_closes_at_epoch_ms_timestamps, which opens a window at a realistic epoch-ms timestamp (1_700_000_000_000) and removes the agg_id. I verified it hangs/times out under the old i64::MAX code (exit 124 at a 45s timeout — the ~i64::MAX/slide loop) and passes instantly with the finite bound.