Confusion: flush_all's unconditional +1ms watermark bump creeps forward on idle ticks — separate late-data-drop path from #474/#536

[milindsrivastava1997]

Context

While code-reviewing PR #536 (fix(precompute): keep wall-clock fallback from force-closing panes still under active ingest, part of the #474 fix stack —
asap-query-engine/src/precompute_engine/worker.rs), I found something that
still doesn't fully make sense to me and want a second pair of eyes on it,
since I'm not the original author of this engine. Writing out the full train
of thought below rather than jumping to the punchline, since the confusion
is as much about why the code is shaped this way as it is about behavior.

Commit reviewed: 380936401107d93934b6c91cbf6d3d7a9150337d on branch
474-precompute-wall-clock-e2e-test.

Step 1 — reviewing the regression test in PR #536

PR #536 adds wall_clock_fallback_does_not_close_a_pane_still_receiving_samples
(worker.rs, in the tests module). It touches a pane 8 times in a for i in 0..8 loop, then calls flush_all() and asserts the pane is not
force-closed, with a comment claiming the check happens "500ms after the
pane's last touch (i=6, at 1_006_000ms)".

That comment doesn't match the loop: 0..8 runs 8 iterations (i = 0..7), so
the actual last touch lands at i=7 → wall-clock 1_007_000ms, one whole
second later than the comment says. All 8 touches finish before
flush_all() is called even once, so the "mid-ingest" flush described in
the comment isn't really mid-ingest — it's post-ingest, with the fake clock
wound backward 500ms to construct the assertion. I confirmed this by
temporarily adding a debug print inside the wall-clock-fallback loop in
flush_all and re-running the test:

DEBUG flush_all: pane_start=0 pane_last_touch_ms=1007000 now_ms=1006500 delta=-500 threshold=6000

delta is negative — the assertion passes because the flush is (fictionally)
timestamped before the last touch, not because it's exercising a
"500ms-since-last-touch, still fresh" boundary. So the test passes, but not
for the reason its own comments describe.

Step 2 — fixing the test's loop bound surfaced a second, unrelated failure

To make the test actually call flush_all() mid-ingest (matching the
comment's intent — and matching what I understood production to do via its
periodic flush timer), I split the loop: 7 touches, then the "still fresh"
flush check, then an 8th touch, then the final closing flush.

That restructured test fails:

thread '...' panicked: expected all 8 touches merged: expected 36, got 28

28 = 1+2+...+7 — the 8th touch's value is missing from the final merged
output. The pane wasn't force-closed (sink.len() was 0 at the mid-ingest
check, as expected) — the 8th touch's sample was silently dropped by a
completely different mechanism than the one this PR touches.

Step 3 — tracing where the 8th touch went

Traced it to the "late data" guard in process_group_samples
(worker.rs:428):

if previous_wm != i64::MIN && *ts < previous_wm - allowed_lateness_ms {
    // dropped as "late"
}

The 8th touch's timestamp is 0 (all samples in this test share one
event-time, deliberately, since that's the degenerate case #474 is about).
For it to be dropped as "late," state.previous_watermark_ms must have
already advanced past allowed_lateness_ms by the time it arrived — even
though the event-time watermark should be frozen at 0 the whole time,
since every sample carries the same timestamp.

Looking at flush_all (worker.rs:601-607):

let mut effective_wm = propagated_wm.saturating_add(1);
...
if effective_wm > state.previous_watermark_ms {
    state.previous_watermark_ms = effective_wm;
}

propagated_wm is read from state.previous_watermark_ms itself, and the
result (+1) gets written straight back into that same field — every time
flush_all runs, regardless of whether any new data arrived or the
wall-clock fallback actually fired. So previous_watermark_ms creeps
forward by 1ms on every call to flush_all, even during a fully idle tick
where nothing happened.

Step 4 — this isn't a one-off; flush_all runs continuously in production

engine.rs:216-229 spawns a background timer that calls flush_all() on
every worker every flush_interval_ms (default 1000, config.rs:58), for
the entire lifetime of the pipeline — not just at shutdown. So the 1ms creep
described in Step 3 isn't confined to my test scenario; it happens on every
tick of that timer, the whole time a group has any open pane, whether or not
the group is receiving new data.

With the production default allowed_lateness_ms = 5000 (config.rs:56),
a stagnant-timestamp pane's own new samples would start getting dropped as
"late" once the creep pushes previous_watermark_ms past
ts + allowed_lateness_ms, i.e. roughly 5000 flush cycles ≈ 83 minutes
of continuous ingest with a frozen timestamp. My test's helper sets
allowed_lateness_ms: 0 (worker.rs:2870), which is why it only took one
flush cycle to manifest there.

This is the same symptom #474 originally described (silent row loss on a
bulk load whose rows share one event-time) but via a completely different
mechanism than the one PR #536 fixes, and it isn't mentioned anywhere in
that PR's design doc.

Step 5 — trying to understand why the +1 is there at all

The only comment on it is // Effective watermark: max(group's own, global) + 1ms for boundary. (worker.rs:601). My first assumption was that
closed_windows (window_manager.rs) requires the watermark to be
strictly greater than a window's end to close it, and that the +1 was
needed to cross that boundary. I checked, and that's wrong — the doc comment
on closed_windows (window_manager.rs:45-46) is explicit that it already
uses >=:

A window [start, start + window_size_ms) is closed when
current_wm >= start + window_size_ms.

What closed_windows does have is a separate early-return guard
(window_manager.rs:50): if current_wm <= previous_wm { return Vec::new(); } — i.e. it requires forward progress (current_wm strictly
greater than previous_wm) just to enter the closing logic at all, distinct
from the >= boundary check inside.

To test whether the +1 is load-bearing anywhere, I temporarily changed
effective_wm = propagated_wm.saturating_add(1) to effective_wm = propagated_wm (no nudge) and ran the full precompute_engine unit suite
(81 tests) plus the three related e2e tests
(e2e_precompute_wall_clock_fallback_active_ingest,
e2e_netflow_single_second, e2e_precompute_equivalence). Everything
passed, including the restructured mid-ingest test from Step 2. I reverted
this change immediately after — it was only a local experiment to answer
the question, not a proposed fix.

So, as far as I can tell from the test suite, the +1 isn't protecting any
currently-tested behavior — the two real cases that need forward progress on
an otherwise-idle tick (the wall-clock fallback's own force_to value, and
a slower worker catching up to a faster worker's globally-propagated
watermark) both already produce a value strictly greater than
state.previous_watermark_ms on their own, without help from the +1.

What I'm confused about / asking here

I don't have a confident answer for why the +1 exists, or whether there's
a scenario (maybe not covered by the current test suite, maybe involving
some interaction I'm not seeing between multi-worker coordination and window
alignment) where it actually matters. I also don't know whether the
83-minute-creep-into-late-data-drop behavior described in Steps 3-4 is a
known, accepted characteristic of this engine that just isn't written down
anywhere, or a genuine gap in the #474 fix. I'd like someone who knows this
engine's watermark/windowing design better than I do to weigh in on both
before anyone (myself included) tries to change anything here.

[milindsrivastava1997]

@zzylol The issue description is AI generated and long but I think it makes sense. The crux is that while trying to fix issue 474, Claude discovered this +1 in the watermark that is called when calling flush_all() and it's not obvious why the +1 is needed. The full train of thought of how Claude discovered this accidentally is written above.