Reproduction steps:
- Confirm the guard is stored for every watermark_type. For each of timestamp /
integer / opaque:
cmd = {"id":"crm.scan",
"schedulable":{"min_interval_minutes":5,"max_runtime_minutes":10},
"incremental":{"watermark_type":<type>,"since_param":"since",
"items_field":"items","cursor_field":"id",
"watermark_from":"mark","max_watermark_jump_seconds":3600}}
incremental_contract.parse(cmd)["incremental"]["max_watermark_jump_seconds"]
-> 3600 for all three types.
- Ask watermark_store.check_advance to make a large forward move:
inc = {"max_watermark_jump_seconds": 3600}
check_advance({"watermark_type":"timestamp","watermark":"2026-08-14T09:00:00Z"},
"2036-08-14T09:00:00Z", inc) # +10 years
check_advance({"watermark_type":"integer","watermark":100}, 999999999, inc)
- Control (proves the integer branch is reached):
check_advance({"watermark_type":"integer","watermark":100}, 50, inc)
Expected:
Both large moves are refused — the operator declared a maximum jump, the contract
validated and stored it, and check_advance's docstring says a forward jump beyond
it is "surfaced for confirmation, not applied".
Actual:
timestamp 100 -> +10 years : ok=False "forward jump ... exceeds 3600s"
integer 100 -> 999999999 : ok=True "ok" (999,999,899 rows skipped)
integer 100 -> 50 : ok=False "refuses to move backwards"
The integer branch implements only the backwards refusal and returns ok without
reading max_watermark_jump_seconds. Because the mark only moves forward, every
record between the true position and the inflated mark is skipped and never
revisited — silent data loss. (Opaque marks are documented as non-comparable, so
only the integer case is affected; timestamp is guarded correctly.)
Suggested fix:
Apply the bound on the integer branch of check_advance():
if wm_type == "integer":
try:
cur_i, cand_i = int(current), int(candidate)
except (TypeError, ValueError):
return False, f"non-integer watermark ({current!r} -> {candidate!r})"
if cand_i < cur_i:
return False, f"refuses to move backwards ({current} -> {candidate})"
limit = (contract_inc or {}).get("max_watermark_jump")
if limit and (cand_i - cur_i) > limit:
return False, (f"forward jump of {cand_i - cur_i} exceeds {limit} — "
f"needs confirmation, not applied silently")
return True, "ok"
max_watermark_jump_seconds is a seconds-named field; for integer marks either add
an explicit max_watermark_jump (id delta), or reject the seconds-named field on
an integer contract at parse time rather than storing a guard that never fires.
The validator and check_advance must agree: a guard the contract accepts must be
one some branch enforces.
Station version (railcall version): station-v0.99
Module slug + version: N/A (platform — workbench/primitives/watermark_store.py)