FA-45676 / Data systems / Open access
Watermark release retains arrival order instead of event-time order · case 01
Watermark release retains arrival order instead of event-time order.
ROOT CAUSE
watermark-buffer-release: Watermark release retains arrival order instead of event-time order.
VERIFIED REPAIR
Preserve the stated physical representation and operation order: Process event actions [event,time,id] and watermark actions [watermark,time,None]. Maintain monotone watermark; events older than it go to late output. On a watermark, release buffered events with time strictly below watermark ordered by (time,id). Return [emitted,late,sorted-pending].
Unsuccessful approach: Identity order is not event-time order.
Case contract
Process event actions [event,time,id] and watermark actions [watermark,time,None]. Maintain monotone watermark; events older than it go to late output. On a watermark, release buffered events with time strictly below watermark ordered by (time,id). Return [emitted,late,sorted-pending].
Why this case matters
A bounded deterministic data engine model makes representation and changelog faults reproducible.
1 / The failure
Exit 1"""Failure Map reference implementation. Python standard library only."""
import json
N = 1
observations = []
def solve(d):
try:
watermark=None; pending=[]; emitted=[]; late=[]
for kind,time,ident in d:
if kind=='event':
if watermark is not None and time<watermark: late.append([time,ident])
else: pending.append([time,ident])
else:
watermark=time if watermark is None else max(watermark,time)
ready=[e for e in pending if e[0]<watermark]
emitted.extend(ready)
pending=[e for e in pending if e[0]>=watermark]
return [emitted,late,sorted(pending)]
except (IndexError, KeyError, ValueError, StopIteration) as exc:
return {"representation_error": type(exc).__name__}
def check(label, actual, expected):
observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
if N == 1:
check('frontier equality', solve([['watermark', 1, None], ['event', 1, 10]]), [[], [], [[1, 10]]])
check('regression', solve([['watermark', 4, None], ['watermark', 1, None], ['event', 2, 10]]), [[], [[2, 10]], []])
check('frontier stays pending', solve([['event', 1, 10], ['watermark', 1, None]]), [[], [], [[1, 10]]])
check('event time ordering', solve([['event', 3, 10], ['event', 1, 11], ['watermark', 4, None]]), [[[1, 11], [3, 10]], [], []])
check('emit once', solve([['event', 1, 10], ['watermark', 2, None], ['watermark', 3, None]]), [[[1, 10]], [], []])
check('late row', solve([['watermark', 3, None], ['event', 1, 10]]), [[], [[1, 10]], []])
check('empty input', solve([]), [[], [], []])
elif N == 2:
check('frontier equality', solve([['watermark', 2, None], ['event', 2, 10]]), [[], [], [[2, 10]]])
check('regression', solve([['watermark', 5, None], ['watermark', 2, None], ['event', 3, 10]]), [[], [[3, 10]], []])
check('frontier stays pending', solve([['event', 2, 10], ['watermark', 2, None]]), [[], [], [[2, 10]]])
check('event time ordering', solve([['event', 4, 10], ['event', 2, 11], ['watermark', 5, None]]), [[[2, 11], [4, 10]], [], []])
check('emit once', solve([['event', 2, 10], ['watermark', 3, None], ['watermark', 4, None]]), [[[2, 10]], [], []])
check('late row', solve([['watermark', 4, None], ['event', 2, 10]]), [[], [[2, 10]], []])
check('empty input', solve([]), [[], [], []])
elif N == 3:
check('frontier equality', solve([['watermark', 3, None], ['event', 3, 10]]), [[], [], [[3, 10]]])
check('regression', solve([['watermark', 6, None], ['watermark', 3, None], ['event', 4, 10]]), [[], [[4, 10]], []])
check('frontier stays pending', solve([['event', 3, 10], ['watermark', 3, None]]), [[], [], [[3, 10]]])
check('event time ordering', solve([['event', 5, 10], ['event', 3, 11], ['watermark', 6, None]]), [[[3, 11], [5, 10]], [], []])
check('emit once', solve([['event', 3, 10], ['watermark', 4, None], ['watermark', 5, None]]), [[[3, 10]], [], []])
check('late row', solve([['watermark', 5, None], ['event', 3, 10]]), [[], [[3, 10]], []])
check('empty input', solve([]), [[], [], []])
elif N == 4:
check('frontier equality', solve([['watermark', 4, None], ['event', 4, 10]]), [[], [], [[4, 10]]])
check('regression', solve([['watermark', 7, None], ['watermark', 4, None], ['event', 5, 10]]), [[], [[5, 10]], []])
check('frontier stays pending', solve([['event', 4, 10], ['watermark', 4, None]]), [[], [], [[4, 10]]])
check('event time ordering', solve([['event', 6, 10], ['event', 4, 11], ['watermark', 7, None]]), [[[4, 11], [6, 10]], [], []])
check('emit once', solve([['event', 4, 10], ['watermark', 5, None], ['watermark', 6, None]]), [[[4, 10]], [], []])
check('late row', solve([['watermark', 6, None], ['event', 4, 10]]), [[], [[4, 10]], []])
check('empty input', solve([]), [[], [], []])
elif N == 5:
check('frontier equality', solve([['watermark', 5, None], ['event', 5, 10]]), [[], [], [[5, 10]]])
check('regression', solve([['watermark', 8, None], ['watermark', 5, None], ['event', 6, 10]]), [[], [[6, 10]], []])
check('frontier stays pending', solve([['event', 5, 10], ['watermark', 5, None]]), [[], [], [[5, 10]]])
check('event time ordering', solve([['event', 7, 10], ['event', 5, 11], ['watermark', 8, None]]), [[[5, 11], [7, 10]], [], []])
check('emit once', solve([['event', 5, 10], ['watermark', 6, None], ['watermark', 7, None]]), [[[5, 10]], [], []])
check('late row', solve([['watermark', 7, None], ['event', 5, 10]]), [[], [[5, 10]], []])
check('empty input', solve([]), [[], [], []])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)
| Boundary fixture | Actual | Expected | Outcome |
|---|---|---|---|
| frontier equality | [[], [], [[1, 10]]] | [[], [], [[1, 10]]] | Passed |
| regression | [[], [[2, 10]], []] | [[], [[2, 10]], []] | Passed |
| frontier stays pending | [[], [], [[1, 10]]] | [[], [], [[1, 10]]] | Passed |
| event time ordering | [[[3, 10], [1, 11]], [], []] | [[[1, 11], [3, 10]], [], []] | Failed |
| emit once | [[[1, 10]], [], []] | [[[1, 10]], [], []] | Passed |
| late row | [[], [[1, 10]], []] | [[], [[1, 10]], []] | Passed |
| empty input | [[], [], []] | [[], [], []] | Passed |
SHA-256 / ef14ee67b1bf042336604caaef8fb1ee5affc9a50162b685445a6c2e94fe92ba
2 / The unsuccessful fix
Exit 1"""Failure Map reference implementation. Python standard library only."""
import json
N = 1
observations = []
def solve(d):
try:
watermark=None; pending=[]; emitted=[]; late=[]
for kind,time,ident in d:
if kind=='event':
if watermark is not None and time<watermark: late.append([time,ident])
else: pending.append([time,ident])
else:
watermark=time if watermark is None else max(watermark,time)
ready=sorted((e for e in pending if e[0]<watermark),key=lambda e:e[1])
emitted.extend(ready)
pending=[e for e in pending if e[0]>=watermark]
return [emitted,late,sorted(pending)]
except (IndexError, KeyError, ValueError, StopIteration) as exc:
return {"representation_error": type(exc).__name__}
def check(label, actual, expected):
observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
if N == 1:
check('frontier equality', solve([['watermark', 1, None], ['event', 1, 10]]), [[], [], [[1, 10]]])
check('regression', solve([['watermark', 4, None], ['watermark', 1, None], ['event', 2, 10]]), [[], [[2, 10]], []])
check('frontier stays pending', solve([['event', 1, 10], ['watermark', 1, None]]), [[], [], [[1, 10]]])
check('event time ordering', solve([['event', 3, 10], ['event', 1, 11], ['watermark', 4, None]]), [[[1, 11], [3, 10]], [], []])
check('emit once', solve([['event', 1, 10], ['watermark', 2, None], ['watermark', 3, None]]), [[[1, 10]], [], []])
check('late row', solve([['watermark', 3, None], ['event', 1, 10]]), [[], [[1, 10]], []])
check('empty input', solve([]), [[], [], []])
elif N == 2:
check('frontier equality', solve([['watermark', 2, None], ['event', 2, 10]]), [[], [], [[2, 10]]])
check('regression', solve([['watermark', 5, None], ['watermark', 2, None], ['event', 3, 10]]), [[], [[3, 10]], []])
check('frontier stays pending', solve([['event', 2, 10], ['watermark', 2, None]]), [[], [], [[2, 10]]])
check('event time ordering', solve([['event', 4, 10], ['event', 2, 11], ['watermark', 5, None]]), [[[2, 11], [4, 10]], [], []])
check('emit once', solve([['event', 2, 10], ['watermark', 3, None], ['watermark', 4, None]]), [[[2, 10]], [], []])
check('late row', solve([['watermark', 4, None], ['event', 2, 10]]), [[], [[2, 10]], []])
check('empty input', solve([]), [[], [], []])
elif N == 3:
check('frontier equality', solve([['watermark', 3, None], ['event', 3, 10]]), [[], [], [[3, 10]]])
check('regression', solve([['watermark', 6, None], ['watermark', 3, None], ['event', 4, 10]]), [[], [[4, 10]], []])
check('frontier stays pending', solve([['event', 3, 10], ['watermark', 3, None]]), [[], [], [[3, 10]]])
check('event time ordering', solve([['event', 5, 10], ['event', 3, 11], ['watermark', 6, None]]), [[[3, 11], [5, 10]], [], []])
check('emit once', solve([['event', 3, 10], ['watermark', 4, None], ['watermark', 5, None]]), [[[3, 10]], [], []])
check('late row', solve([['watermark', 5, None], ['event', 3, 10]]), [[], [[3, 10]], []])
check('empty input', solve([]), [[], [], []])
elif N == 4:
check('frontier equality', solve([['watermark', 4, None], ['event', 4, 10]]), [[], [], [[4, 10]]])
check('regression', solve([['watermark', 7, None], ['watermark', 4, None], ['event', 5, 10]]), [[], [[5, 10]], []])
check('frontier stays pending', solve([['event', 4, 10], ['watermark', 4, None]]), [[], [], [[4, 10]]])
check('event time ordering', solve([['event', 6, 10], ['event', 4, 11], ['watermark', 7, None]]), [[[4, 11], [6, 10]], [], []])
check('emit once', solve([['event', 4, 10], ['watermark', 5, None], ['watermark', 6, None]]), [[[4, 10]], [], []])
check('late row', solve([['watermark', 6, None], ['event', 4, 10]]), [[], [[4, 10]], []])
check('empty input', solve([]), [[], [], []])
elif N == 5:
check('frontier equality', solve([['watermark', 5, None], ['event', 5, 10]]), [[], [], [[5, 10]]])
check('regression', solve([['watermark', 8, None], ['watermark', 5, None], ['event', 6, 10]]), [[], [[6, 10]], []])
check('frontier stays pending', solve([['event', 5, 10], ['watermark', 5, None]]), [[], [], [[5, 10]]])
check('event time ordering', solve([['event', 7, 10], ['event', 5, 11], ['watermark', 8, None]]), [[[5, 11], [7, 10]], [], []])
check('emit once', solve([['event', 5, 10], ['watermark', 6, None], ['watermark', 7, None]]), [[[5, 10]], [], []])
check('late row', solve([['watermark', 7, None], ['event', 5, 10]]), [[], [[5, 10]], []])
check('empty input', solve([]), [[], [], []])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)
| Boundary fixture | Actual | Expected | Outcome |
|---|---|---|---|
| frontier equality | [[], [], [[1, 10]]] | [[], [], [[1, 10]]] | Passed |
| regression | [[], [[2, 10]], []] | [[], [[2, 10]], []] | Passed |
| frontier stays pending | [[], [], [[1, 10]]] | [[], [], [[1, 10]]] | Passed |
| event time ordering | [[[3, 10], [1, 11]], [], []] | [[[1, 11], [3, 10]], [], []] | Failed |
| emit once | [[[1, 10]], [], []] | [[[1, 10]], [], []] | Passed |
| late row | [[], [[1, 10]], []] | [[], [[1, 10]], []] | Passed |
| empty input | [[], [], []] | [[], [], []] | Passed |
SHA-256 / 3ff2e39b38eb2eb7cdbbc0f5a8ab421078d8d3c1687efcd9e1a07eb18889fc68
3 / The verified repair
Exit 0"""Failure Map reference implementation. Python standard library only."""
import json
N = 1
observations = []
def solve(d):
try:
watermark=None; pending=[]; emitted=[]; late=[]
for kind,time,ident in d:
if kind=='event':
if watermark is not None and time<watermark: late.append([time,ident])
else: pending.append([time,ident])
else:
watermark=time if watermark is None else max(watermark,time)
ready=sorted(e for e in pending if e[0]<watermark)
emitted.extend(ready)
pending=[e for e in pending if e[0]>=watermark]
return [emitted,late,sorted(pending)]
except (IndexError, KeyError, ValueError, StopIteration) as exc:
return {"representation_error": type(exc).__name__}
def check(label, actual, expected):
observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
if N == 1:
check('frontier equality', solve([['watermark', 1, None], ['event', 1, 10]]), [[], [], [[1, 10]]])
check('regression', solve([['watermark', 4, None], ['watermark', 1, None], ['event', 2, 10]]), [[], [[2, 10]], []])
check('frontier stays pending', solve([['event', 1, 10], ['watermark', 1, None]]), [[], [], [[1, 10]]])
check('event time ordering', solve([['event', 3, 10], ['event', 1, 11], ['watermark', 4, None]]), [[[1, 11], [3, 10]], [], []])
check('emit once', solve([['event', 1, 10], ['watermark', 2, None], ['watermark', 3, None]]), [[[1, 10]], [], []])
check('late row', solve([['watermark', 3, None], ['event', 1, 10]]), [[], [[1, 10]], []])
check('empty input', solve([]), [[], [], []])
elif N == 2:
check('frontier equality', solve([['watermark', 2, None], ['event', 2, 10]]), [[], [], [[2, 10]]])
check('regression', solve([['watermark', 5, None], ['watermark', 2, None], ['event', 3, 10]]), [[], [[3, 10]], []])
check('frontier stays pending', solve([['event', 2, 10], ['watermark', 2, None]]), [[], [], [[2, 10]]])
check('event time ordering', solve([['event', 4, 10], ['event', 2, 11], ['watermark', 5, None]]), [[[2, 11], [4, 10]], [], []])
check('emit once', solve([['event', 2, 10], ['watermark', 3, None], ['watermark', 4, None]]), [[[2, 10]], [], []])
check('late row', solve([['watermark', 4, None], ['event', 2, 10]]), [[], [[2, 10]], []])
check('empty input', solve([]), [[], [], []])
elif N == 3:
check('frontier equality', solve([['watermark', 3, None], ['event', 3, 10]]), [[], [], [[3, 10]]])
check('regression', solve([['watermark', 6, None], ['watermark', 3, None], ['event', 4, 10]]), [[], [[4, 10]], []])
check('frontier stays pending', solve([['event', 3, 10], ['watermark', 3, None]]), [[], [], [[3, 10]]])
check('event time ordering', solve([['event', 5, 10], ['event', 3, 11], ['watermark', 6, None]]), [[[3, 11], [5, 10]], [], []])
check('emit once', solve([['event', 3, 10], ['watermark', 4, None], ['watermark', 5, None]]), [[[3, 10]], [], []])
check('late row', solve([['watermark', 5, None], ['event', 3, 10]]), [[], [[3, 10]], []])
check('empty input', solve([]), [[], [], []])
elif N == 4:
check('frontier equality', solve([['watermark', 4, None], ['event', 4, 10]]), [[], [], [[4, 10]]])
check('regression', solve([['watermark', 7, None], ['watermark', 4, None], ['event', 5, 10]]), [[], [[5, 10]], []])
check('frontier stays pending', solve([['event', 4, 10], ['watermark', 4, None]]), [[], [], [[4, 10]]])
check('event time ordering', solve([['event', 6, 10], ['event', 4, 11], ['watermark', 7, None]]), [[[4, 11], [6, 10]], [], []])
check('emit once', solve([['event', 4, 10], ['watermark', 5, None], ['watermark', 6, None]]), [[[4, 10]], [], []])
check('late row', solve([['watermark', 6, None], ['event', 4, 10]]), [[], [[4, 10]], []])
check('empty input', solve([]), [[], [], []])
elif N == 5:
check('frontier equality', solve([['watermark', 5, None], ['event', 5, 10]]), [[], [], [[5, 10]]])
check('regression', solve([['watermark', 8, None], ['watermark', 5, None], ['event', 6, 10]]), [[], [[6, 10]], []])
check('frontier stays pending', solve([['event', 5, 10], ['watermark', 5, None]]), [[], [], [[5, 10]]])
check('event time ordering', solve([['event', 7, 10], ['event', 5, 11], ['watermark', 8, None]]), [[[5, 11], [7, 10]], [], []])
check('emit once', solve([['event', 5, 10], ['watermark', 6, None], ['watermark', 7, None]]), [[[5, 10]], [], []])
check('late row', solve([['watermark', 7, None], ['event', 5, 10]]), [[], [[5, 10]], []])
check('empty input', solve([]), [[], [], []])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)
| Boundary fixture | Actual | Expected | Outcome |
|---|---|---|---|
| frontier equality | [[], [], [[1, 10]]] | [[], [], [[1, 10]]] | Passed |
| regression | [[], [[2, 10]], []] | [[], [[2, 10]], []] | Passed |
| frontier stays pending | [[], [], [[1, 10]]] | [[], [], [[1, 10]]] | Passed |
| event time ordering | [[[1, 11], [3, 10]], [], []] | [[[1, 11], [3, 10]], [], []] | Passed |
| emit once | [[[1, 10]], [], []] | [[[1, 10]], [], []] | Passed |
| late row | [[], [[1, 10]], []] | [[], [[1, 10]], []] | Passed |
| empty input | [[], [], []] | [[], [], []] | Passed |
SHA-256 / 15b2eb00a8b04c69e6f07835ab64ddf0303420db98d87d21c7e99daef68f39a2
Verification & scope
Offline stipulated semantics over valid small inputs; no performance, concurrency, or production-engine conformance claim. This reproducer isolates one failure mechanism. Results cover the supplied fixtures. Variants within a family share a test contract and should remain grouped when constructing evaluation splits. Related mechanisms with a shared evaluation_group must also remain together; these controlled models are not independent production incidents.
Observations recorded using Python 3.12.14 at 2026-09-29T14:44:24.621513+00:00.
Case digest / e61441e1b060771e0327b62d39bbb5f65a9ee95f9e38ea2113ac62016e2b8839