Managing watermarks when a portal restates history

At 02:15 the nightly run finishes and the change table has 6 214 new rows for a county that averages forty a week. Nothing failed. The payload was well-formed, the schema matched, the geometry was valid, and every one of those 6 214 differences is real in the sense that the bytes genuinely differ from yesterday’s. What happened is that the county restored its parcel layer from an August backup, so the file you fetched describes a state from three months ago — and your pipeline, which was built to notice change, noticed all of it. This guide covers detecting that situation before it reaches the change table, and reconciling it afterwards if it already has. It is the recovery-path companion to scheduling & orchestration for municipal feed runs.

Diagnosis: telling a restatement from a change

A restatement and a genuine bulk update look identical to a content digest, because both produce a payload whose bytes differ from the last one. Four signals separate them, and only the first is conclusive on its own.

The publisher’s own timestamp moving backwards is the definitive signal. If the feed’s updated field, its metadata endpoint, or its file name encodes a date earlier than the one stored in your watermark, the publisher has told you it went back in time. Nothing else needs checking.

Four signals, and what each one can and cannot conclude A four-row by three-column matrix of restatement signals. Rows are the publisher timestamp moving backwards, a change set dominated by reversals, change volume far above the norm, and a falling record count. Columns are what the signal proves, how often it is available, and what it can be confused with. Only the backwards timestamp is conclusive, and it is unavailable on many portals. The reversal ratio is nearly as strong and is always computable from your own history. High volume alone is confusable with a genuine county-wide re-code, and a falling record count indicates truncation rather than restatement. Four signals, and what each one can and cannot conclude what it proves available? confusable with Publisher timestamp went backwards conclusive many portals omit it nothing Most changes revert recent values strong always — it is your history a genuine reversal ordinance Volume far above the norm suggestive only always a real county-wide re-code Record count fell a different fault always truncation, not restatement reliable partial ambiguous
The top row is conclusive and often missing; the second is computable from your own history and nearly as strong. That is why the classifier uses both — a portal that restates is also a portal whose metadata you cannot trust, so the check that does not depend on the publisher is the one that fires.

A change set dominated by reversals is the strong secondary signal. In a restatement, most differences are your recent knowledge being undone: parcels revert to codes they held before, and the changes you recorded last month appear as changes back. A genuine bulk re-code moves parcels to new values, not to values you previously held.

Change volume far outside the norm is suggestive and not sufficient — a county-wide re-code after an ordinance rewrite is also enormous, and that one is real.

Record count falling points at a truncated or partial publication rather than a restatement, which needs a different response: hold and re-fetch, rather than hold and reconcile.

def classify_bulk_change(changes, watermark, payload, history) -> str:
    """Distinguish a restatement from a genuine bulk update before publishing."""
    if payload.published_at and watermark.published_at \
            and payload.published_at < watermark.published_at:
        return "restated"                       # conclusive: the publisher went backwards

    if payload.record_count < watermark.record_count * 0.98:
        return "truncated"                      # a different problem, different fix

    # How many of these "changes" revert a value we recorded in the last 90 days?
    reverting = 0
    for ch in changes:
        prior = history.value_before_last_change(ch.parcel_id, within_days=90)
        if prior is not None and ch.new_value == prior:
            reverting += 1

    if changes and reverting / len(changes) > 0.6:
        return "restated"                       # most of it undoes our recent knowledge

    return "bulk_update"                        # large, and apparently real

The reversal ratio is the check worth having even when the publisher supplies a timestamp, because portals that restate frequently are also the portals whose metadata is least trustworthy. Sixty per cent is a deliberately loose threshold: a genuine bulk re-code will sit near zero, and a restatement near ninety.

Step-by-step implementation

1. Store enough in the watermark to detect it

A watermark holding only a content digest cannot detect a restatement, because the digest of an older state differs from the current one exactly as a new state would. Three fields have to be there.

@dataclass(frozen=True)
class Watermark:
    source_id: str
    content_digest: str            # detects "did anything change?"
    published_at: datetime | None  # detects "did time go backwards?"
    record_count: int              # detects truncation
    consumed_at: datetime
    run_id: str

2. Make the verdict a first-class outcome, not an exception

A restatement is neither success nor failure, and forcing it into either produces the wrong behaviour: treated as success, it publishes; treated as a transient error, it retries and publishes on the second attempt.

def consume(source, store, fetch, differ, publisher) -> str:
    wm = store.load(source.id)
    payload = fetch.get(source)
    digest = sha256(payload.body).hexdigest()

    if wm and digest == wm.content_digest:
        store.touch(source.id)
        return "skip"

    changes = differ.diff(store.current_snapshot(source.id), payload)
    verdict = classify_bulk_change(changes, wm, payload, store.history) if wm else "bulk_update"

    if verdict in ("restated", "truncated"):
        # Hold: keep the payload for inspection, do NOT advance the watermark, and
        # do NOT publish. The previous snapshot stays authoritative meanwhile.
        store.quarantine_payload(source.id, payload, digest, reason=verdict,
                                 change_count=len(changes))
        publisher.hold(source.id, reason=verdict)
        return verdict

    publisher.publish(source.id, changes)
    store.advance_watermark(source.id, digest, payload)
    return "published"

3. Reconcile the held payload

Holding buys time; it does not answer the question. There are three legitimate resolutions and the choice belongs to a person, once, per incident.

The restatement is an accident and the portal will fix it. Do nothing but keep holding. Your last good snapshot remains authoritative, consumers are served from it with a staleness flag as fallback routing logic describes, and the next correct publication resumes normally. This is the most common case.

The restatement is authoritative — the county says the older state was right. Then your recent change history is wrong and has to be retracted rather than overwritten. Write the reversals as explicit corrections with a new transaction time so both beliefs survive, which is precisely what the bitemporal model in temporal versioning & snapshots exists for.

The restatement is partial. Some parcels reverted because of the backup; others changed legitimately in the same publication. Split the change set by whether each change reverts a recent value, publish the non-reverting subset, and hold the rest.

def split_restatement(changes, history, within_days=90):
    """Separate 'undoes what we knew' from 'genuinely new', so a partial
    restatement can be published in part rather than held whole."""
    reverting, novel = [], []
    for ch in changes:
        prior = history.value_before_last_change(ch.parcel_id, within_days=within_days)
        (reverting if prior is not None and ch.new_value == prior else novel).append(ch)
    return reverting, novel

4. Never advance the watermark past a held payload

The one irreversible mistake is advancing the watermark while holding, because the next run then sees no change and the restatement becomes invisible — you are now serving the older state with no record of how you got there. The watermark advances only when a payload is published.

Verification & testing

The scenario is straightforward to build as a fixture, and it is worth having because the code path only runs during incidents.

Change volume against its own trailing median Line chart of daily change volume for one county across twelve days, against the trailing median of about forty changes a day. Eleven days sit between 36 and 44. Day seven reaches 6 214, roughly 155 times the median, which is the restatement. The chart shows that a per-source ratio to its own median is a far better detector than any absolute threshold, because forty is normal here and would be a crisis for a quiet county. Change volume against its own trailing median 0 2000 4000 6000 8000 1 2 3 4 5 6 7 8 9 10 11 12 Day Changes in the run changes published trailing median
A ratio against the source's own trailing median is what makes this detectable: 6 214 is 155× normal for this county and would be an ordinary Tuesday for a metropolitan one. Alarm above roughly 20× and a human looks at both restatements and genuine bulk re-codes before subscribers do.
def test_backwards_timestamp_is_held(store, fetch, differ, publisher):
    fetch.queue(payload(published_at="2026-08-24", records=41_000, body=b"...current"))
    assert consume(source, store, fetch, differ, publisher) == "published"
    wm = store.load("larimer")

    fetch.queue(payload(published_at="2026-05-06", records=41_000, body=b"...older"))
    assert consume(source, store, fetch, differ, publisher) == "restated"

    assert store.load("larimer") == wm              # watermark untouched
    assert publisher.published_change_count() == 0  # nothing reached subscribers
    assert store.quarantined("larimer").reason == "restated"


def test_reversal_ratio_catches_a_silent_restatement(store, fetch, differ, publisher):
    """A portal that restates WITHOUT moving its timestamp — the common case."""
    seed_history(store, parcel="0714-22-1", values=["AG-2", "R-1"])
    fetch.queue(payload(published_at="2026-08-25", records=41_000,
                        changes_reverting_fraction=0.91))
    assert consume(source, store, fetch, differ, publisher) == "restated"

In production, the standing check is a per-source ratio of change volume to its own trailing median, alarming above roughly twenty times. That threshold catches restatements and genuine bulk re-codes alike, which is correct: both deserve a human glance before they reach subscribers.

Failure recovery

The restatement already published. Retract the published change set by run_id, then decide the resolution as above. Retraction has to be visible to subscribers — a notification that says “the 6 214 changes sent at 02:15 were a source restatement and are withdrawn” is far better than silence, because subscribers have already acted on some of them.

Three resolutions for a held payload A three-row by four-column matrix of restatement resolutions. Rows are the portal will fix it, the restatement is authoritative, and the restatement is partial. Columns are what to do with the watermark, what to publish, what consumers are served meanwhile, and who decides. In the first case nothing is published and the last good snapshot continues to serve. In the second the reversals are written as bitemporal corrections. In the third the change set is split and only the non-reverting part is published. In all three the decision belongs to a person rather than to the pipeline. Three resolutions for a held payload watermark what is published consumers see decided by The portal will fix it unchanged nothing last good, flagged a person The restatement is authoritative advance after correcting reversals as corrections corrected values a person The restatement is partial advance partially the non-reverting subset mixed, flagged a person safe needs a decision never automatic
The last column is the same in every row on purpose. A pipeline can detect a restatement reliably, and it cannot know whether the county meant it — so the correct automated behaviour is to hold and ask, once, per incident.

The watermark advanced before the hold was implemented. Restore it from the watermark history to the value it held before the restated run, then re-run. If the watermark table is not versioned, reconstruct it from the last published payload’s digest and timestamp — which is the argument for making that table append-only.

The portal restates repeatedly. Some counties do this every time they rebuild an index. Once a source has restated twice, stop treating it as an anomaly: require positive forward movement in the publisher timestamp before consuming at all, and accept that this source will occasionally be a day behind rather than occasionally wrong.

Frequently asked questions

Why not just always trust the newest payload?

Because “newest fetched” and “most recent state” are different things, and a restatement is exactly the case where they diverge. Trusting fetch order means a backup restore silently rewrites your current state and emits thousands of false changes to subscribers. The pipeline’s job is to represent what the source knows, and a source that published an August state in November has not told you anything new about November.

Should a restatement be treated as an error and retried?

No — retrying fetches the same restated payload again and, on a scheduler configured for retries, publishes it on the second attempt. A restatement is a verdict, not a failure: the run completed correctly and reached a conclusion that requires a decision. Model it as its own outcome so it can be alerted on without being retried.

How long should a held payload be kept?

Until the incident is resolved, plus the retention you would apply to any source archive. The held payload is the evidence for what the portal published on that date, so discarding it after a fixed window removes the ability to prove the restatement happened. It is also the fixture you will want when writing the test that stops the next one reaching subscribers.

What if the restatement is genuinely authoritative?

Then record the reversals as corrections with a new transaction time rather than overwriting history. Your March report stays reproducible as it was, and your current best knowledge reflects the county’s correction — the double-bind bitemporal storage exists to resolve. Overwriting is the one option that destroys information, and it is the one a naive publish does automatically.