python-watermark-retraction-window-merge

Answer

The solution is written to ~/SOLUTION.md (749 lines), with runnable source in ~/watermark_engine/.

What was delivered

Root-cause analysis — six failure modes that break a naive implementation:

1. A single global watermark stalls forever on a silent partition.
2. Recomputing the min watermark by scanning is O(P) per event.
3. Watermarks regress (new/idle-resumed partitions), corrupting emission/compaction.
4. Late corrections to fired windows silently corrupt results.
5. Duplicate event IDs double-count under at-least-once delivery.
6. Retaining all events to support corrections is unbounded.

Exact fix — WatermarkAggregator with four invariants:

• I1 effective watermark is a monotonic high-water mark (watermark_hwm = max(hwm, min_active)), while raw min is maintained with a versioned min-heap (lazy deletion).
• I2 windows fire at end <= hwm; corrections accepted until end + lateness <= hwm and emitted as retract(old) → upsert(new).
• I3 exactly-once contributions via (partition, event_id) dedupe; re-arrival is either a no-op duplicate or a delta correction (which can even move an event across windows).
• I4 windows, per-event contributions, and dedupe rows are compacted once their grace elapses, so state is bounded by the lateness horizon, never by history.

Idle partitions are excluded from the min (via mark_idle), resume automatically on the next event, and heartbeat advances time without adding a contribution. Output is deterministically sorted.

Verification — 13 tests run clean:

ALL TESTS PASSED
state_is_bounded_and_deterministic OK (peak_state=16, peak_dedupe=33, records=82)
replay_equals_bruteforce_exactly_once OK (60 final windows)

The strongest check is the end-to-end one: replaying the emitted retract/upsert changelog reproduces a brute-force computation over the raw latest-value history exactly. I also extracted the two Python blocks back out of the markdown into a clean temp dir and ran the suite — it reproduces the documented output verbatim, so the markdown is directly usable.

Run it:

cd ~/watermark_engine && python3 test_engine.py