No. Network and queue delays can reorder events.

Define event identity, timestamp semantics, aggregation keys, windows, permitted lateness, and correction policy. Partition by aggregate key to keep updates coordinated, while recognizing hot keys may need partial aggregation and a merge step. Deduplication state and offsets/checkpoints must recover consistently with output effects. Watermarks are protocol assumptions, not a universal guarantee that no older event will ever arrive. Keep provisional and finalized results distinguishable. Global exactly-once effects depend on the full source-state-sink protocol and replay behavior.

Contract and event schema

Define event_id, aggregate key, event timestamp, producer version, and payload. Choose processing-time or event-time windows, allowed lateness, and whether late data is dropped, corrected, or sent for offline recomputation.

Processing flow

Ingest into a durable partitioned log, validate and deduplicate, assign windows, update keyed state, and publish provisional or final aggregates. Partition ordering is not global ordering, and hot keys may require partial aggregation.

Recovery

Coordinate checkpointed offsets, dedup state, aggregate state, and sink updates. Replaying input must not duplicate committed effects. A transactional sink or idempotent versioned aggregate write may support the required guarantee.

Serving and operations

Readers need freshness and finality metadata, not only a number. Monitor lag, late-event rate, state size, hot partitions, and correction backlog. Retention must cover the promised replay and correction windows.

What the function promises: WindowCounter is a local event-time model with fixed-size windows, stable event IDs, and a max-observed-time minus lateness watermark. It retains all state for demonstration and drops too-late events.

class WindowCounter:
    def __init__(self, width, lateness):
        if width <= 0 or lateness < 0:
            raise ValueError("Invalid window")
        self.width, self.lateness = width, lateness
        self.maximum = 0
        self.seen, self.counts = set(), {}
    def add(self, event_id, event_time):
        if event_time < 0:
            raise ValueError("Invalid event time")
        if event_id in self.seen or event_time < self.maximum - self.lateness:
            return False
        self.maximum = max(self.maximum, event_time)
        self.seen.add(event_id)
        window = event_time // self.width * self.width
        self.counts[window] = self.counts.get(window, 0) + 1
        return True

counter = WindowCounter(10, 5)
assert counter.add("a", 12)
assert counter.add("b", 18)
assert not counter.add("a", 12)
assert not counter.add("late", 9)
assert counter.counts == {10: 2}

Work and storage: Expected O(1) local event updates. Production state must be retired according to finalized windows, replay, and dedup retention rules.

When to recognize it: A watermark is an explicit progress assumption. Events older than the accepted lateness boundary may be dropped or handled through a correction path.

Check before applying it: Window finality, late-event policy, dedup retention, and checkpoint recovery must agree.

A replayed old event can be counted again unless finalized-window policy prevents it. Coordinate retention across source offsets, aggregate state, and dedup records.