Exactly-once windowed aggregates

Design a stateful stream-processing job that computes real-time per-merchant fraud-risk aggregates (e.g., count + sum of transactions per merchant over sliding 5-minute and 1-hour windows) off a 1M-events/sec transaction stream, emitting an alert when a window crosses a threshold. The aggregates must be correct under failures, a job restart must not double-count nor lose events, and must handle late and out-of-order events (a transaction event arriving 20 s after its actual time). How do you compute exactly-once windowed aggregates that survive crashes and late data?