Streaming Semantics — Watermarks and Exactly-Once TODO
Concept
In stream processing, the time an event actually occurred (event time) differs from the time the system processes it (processing time), and network delays or retries can scramble the order further. A watermark is the system's estimate that 'essentially all events earlier than this timestamp have arrived by now' — it's the signal that decides when to close an event-time window and emit a result. Because a watermark is only an estimate, late data can still show up, which means you need a policy — an allowed lateness window — for updating results or routing late data down a separate path. This is where the tradeoff between completeness and latency appears: a conservative watermark is accurate but slow, an aggressive one is fast but lets more late data through. Exactly-once semantics keep operator state consistent via periodic checkpoints, and pair that with transactional commits or idempotent writes at the sink so results aren't duplicated after a restart.
Most cases of 'why are my aggregates slightly off' aren't bugs — they come from never having pinned down an event-time and watermark policy in the first place.
Code & Formula
# 스트리밍 처리 의미론 — 워터마크로 이벤트 시간 윈도우를 닫고, 늦은 데이터를 걸러낸다.
# event_time 이 뒤섞여 도착해도 워터마크(진행 추정치) 기준으로 윈도우 완결을 판단한다.
events = [
# (event_time, payload) — 도착 순서는 뒤섞여 있다 (네트워크 지연 시뮬레이션)
(1, "a"), (2, "b"), (5, "c"), (3, "d"), (9, "e"), (4, "late-for-w1"),
]
WINDOW = 5 # [0,5), [5,10) 처럼 크기 5 윈도우
ALLOWED_LATENESS = 1 # 워터마크를 지난 뒤에도 1만큼은 늦은 데이터로 받아준다
def window_of(t):
start = (t // WINDOW) * WINDOW
return (start, start + WINDOW)
state = {} # window -> 누적 payload 리스트
closed = set() # 이미 결과를 낸(닫힌) 윈도우
late_dropped = []
watermark = -1
for event_time, payload in events:
watermark = max(watermark, event_time - 1) # 단순화한 워터마크 추정: max(event_time) - 1
w = window_of(event_time)
if w in closed and watermark - ALLOWED_LATENESS >= w[1]:
late_dropped.append((event_time, payload))
continue
state.setdefault(w, []).append(payload)
# 워터마크가 윈도우 끝을 지나면 그 윈도우를 닫고 결과를 낸다(exactly-once: 한 번만 emit)
if watermark >= w[1] and w not in closed:
closed.add(w)
for w in sorted(state):
status = "closed" if w in closed else "open"
print(f"window {w} ({status}): {state[w]}")
print("late data (dropped after allowed lateness):", late_dropped)
print("final watermark:", watermark)
docs/code/algorithms/algorithms-77.py
Exercise
Build an event stream that's deliberately out of order with some events significantly delayed, then vary the watermark delay and the allowed lateness, and record how the windowed aggregate results and the count of late events change.
Practical Connection
Chain event indexing has its own gap between block time and receipt time, and reorgs can even flip the past, so Verex's volume and position aggregation pipeline needs exactly this kind of watermark and correction-handling design.
Where it lands in Jayverse
- Verex: set an explicit watermark delay and allowed-lateness window for volume/position aggregation. Write the completeness-vs-latency tradeoff as a number, not an assumption, since block time and receipt time already diverge.
- Verex: route late or reorged events down a correction path. Match receipt-is-not-settlement's reorged state instead of silently mutating an already-emitted aggregate when a reorg flips the past.
- Auditor: require exactly-once semantics for settlement-relevant aggregation. Checkpointed state and idempotent sink writes matter here because a duplicated write after a restart is exactly the silent inconsistency an audit should catch.
Key expressions
| Expression | 뜻 · 쓰이는 자리 |
|---|---|
| scramble the order | 순서를 뒤섞다 · 네트워크 지연으로 이벤트 도착 순서가 엉킬 때. "network delays or retries can scramble the order further" |
| essentially all | 사실상 전부의 · 거의 다 도착했다고 추정할 때. "essentially all events earlier than this timestamp have arrived" |
| allowed lateness | 허용 지연(윈도우가 닫힌 뒤에도 봐주는 시간) · 늦게 도착한 데이터를 처리하는 정책. "you need a policy — an allowed lateness window" |
| the tradeoff between completeness and latency | 완전성과 지연시간 사이의 트레이드오프 · 정확함과 빠름을 동시에 가질 수 없는 딜레마. "the tradeoff between completeness and latency appears" |
| pin down | 명확히 정해두다·못박다 · 정책이나 기준을 애매하게 두지 않는 것. "never having pinned down an event-time and watermark policy" |
| slightly off | 살짝 어긋난·조금 맞지 않는 · 집계 결과가 미묘하게 틀릴 때. "why are my aggregates slightly off" |
| flip the past | 과거를 뒤집다(이미 지난 사실을 바꿔놓다) · 리오그처럼 확정된 줄 알았던 게 바뀔 때. "reorgs can even flip the past" |
If you study this on a given day, add a note link and a ✅ to this line in the source curriculum (docs/knowledge/dev-100-curriculum.md) and this spot will lead straight to the note body. You can also write directly on this page — but regenerating overwrites it, so it's safer to keep anything you want to save as markdown under docs/algorithms/.