watermark_store.check_advance() states its contract in its own docstring:
Two refusals, both about a module that lies or breaks (§6):
BACKWARDS — would re-process everything in between, and on a workflow
with a payment node that means charging people twice;
a forward JUMP beyond max_watermark_jump_seconds — would skip every
record in the gap, invisibly. Surfaced for confirmation, not applied.
Neither refusal is qualified by watermark type. The implementation applies the
jump refusal only on the timestamp branch:
if wm_type == "timestamp":
... backwards check ...
limit = (contract_inc or {}).get("max_watermark_jump_seconds", ...)
if (n - c).total_seconds() > limit:
return False, "forward jump of …s exceeds …s — needs confirmation"
return True, "ok"
if wm_type == "integer":
try:
if int(candidate) < int(current):
return False, "refuses to move backwards"
except (TypeError, ValueError):
return False, "non-integer watermark"
return True, "ok" # <- no jump check at all
return True, "ok" # opaque: not comparable, accept
Integers are perfectly comparable, so the guard is implementable here — it was
simply not written. The operator's setting is not ignored for lack of a value
either: incremental_contract.parse() validates max_watermark_jump_seconds and
stores it in the normalised contract for EVERY watermark_type, including
integer. It is parsed, type-checked at install time, persisted, and then never
read on this branch.
The consequence is the one the docstring names. candidate_watermark() takes the
mark from a field on the items the module returned, so a module that returns a
single item whose watermark field is an inflated integer advances the mark
there; check_advance accepts it; and because the mark only ever moves forward
(watermark_store's own note: records "it skipped are never revisited because the
watermark only moves forward"), every record in the gap is skipped permanently
and silently.
Reproduction steps:
- Show the guard is stored for every type. For each of timestamp / integer /
opaque, parse a command declaring both blocks:
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"]
- Ask check_advance to make a large forward move under that same guard:
inc = {"max_watermark_jump_seconds": 3600}
# timestamp control — a ten-year jump
check_advance({"watermark_type": "timestamp",
"watermark": "2026-08-14T09:00:00Z"},
"2036-08-14T09:00:00Z", inc)
# integer — the same shape of lie
check_advance({"watermark_type": "integer", "watermark": 100},
999999999, inc)
- Control, to prove the integer branch is reached at all:
check_advance({"watermark_type": "integer", "watermark": 100}, 50, inc)
Expected:
Both moves are refused. The operator declared a maximum jump, the contract
validated and stored it, and the docstring says a forward jump beyond it is
"surfaced for confirmation, not applied".
Actual:
contract stores the guard for every type:
timestamp -> stored max_watermark_jump_seconds = 3600
integer -> stored max_watermark_jump_seconds = 3600
opaque -> stored max_watermark_jump_seconds = 3600
check_advance():
timestamp 100 -> +10 years
ok=False "forward jump of 315619200s exceeds 3600s — needs
confirmation, not applied silently"
integer 100 -> 999999999 (999,999,899 records skipped)
ok=True "ok"
integer 100 -> 50 (backwards control)
ok=False "refuses to move backwards (100 -> 50)"
The identical lie is refused on a timestamp watermark and accepted on an integer
one. The backwards control refuses, so the integer branch is genuinely reached
and simply omits the jump check.
Root cause:
primitives/watermark_store.py check_advance() — the integer branch implements
only the backwards comparison and returns "ok" without consulting
max_watermark_jump_seconds, while the timestamp branch implements both.
primitives/incremental_contract.py parse() validates and persists the field
irrespective of watermark_type, so an operator setting it on an integer
contract gets a guard that installs cleanly and never fires.
Suggested fix:
Apply the same bound on the integer branch. The field is named in seconds, so
either interpret it as a maximum id delta for integer marks, or add an explicit
sibling (e.g. max_watermark_jump_ids) and reject the seconds-named field on an
integer contract at parse time rather than storing a setting that cannot apply:
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"
Whichever shape is chosen, the validator and check_advance should agree: a
guard the contract accepts must be one some branch enforces.
Honest scope:
Requires a module whose contract declares watermark_type "integer" — one of the
three supported types, natural for APIs with monotonic ids — and a watermark
value that is wrong, whether through a buggy handler, a provider returning an
out-of-range id, or a hostile module. It is not remotely triggerable and needs
no forged signature.
The failure is silent data loss rather than a governance bypass: records between
the true position and the inflated mark are never fetched, and because the mark
only moves forward they are not recoverable by waiting — an operator would have
to reset the watermark deliberately. Nothing executes that should not, and no
duplicate delivery or double charge results.
I am NOT claiming the opaque branch is a defect. It also stores the setting and
never reads it, but opaque marks are explicitly documented as not comparable
("the provider's own cursor … we cannot compare them"), so no guard is possible
there. Integer is the clean case: it is comparable, the backwards half is
already implemented on it, and only the jump half is missing.
Counter-evidence checked:
- Ran the real check_advance() and incremental_contract.parse() from a clean
v0.99 tarball extraction, not a reimplementation.
- Ran a backwards control on the SAME integer record, which correctly refuses —
so the integer branch is reached and this is a missing check, not an
unreachable branch or a bad test fixture.
- Ran a timestamp control with an equivalent-magnitude jump under the identical
guard value, which correctly refuses — so the guard itself works and only its
application is type-dependent.
- Confirmed the field is not merely absent for integer contracts: parse() stores
max_watermark_jump_seconds = 3600 for all three types, so the enforcement side
had the value available.
- Checked the direction of harm rather than assuming: this over-advances (skips
records), it does not re-deliver, so it is not a double-charge risk.
Distinctness:
This is the integer branch of check_advance(). It is distinct from the earlier
fixed item where seen_window_seconds and max_watermark_jump_seconds skipped the
type validation lookback_seconds received — that concerned parse() accepting a
malformed value and is fixed (parse now type-checks both); this concerns
check_advance never reading the well-formed value on one branch. It is also
distinct from the seen_param/seen-window item filed alongside this one, which
concerns dedupe rather than the watermark. I did not find a community thread
about the integer watermark jump guard.