Backfilling a missed week of county feeds without double-counting

The scheduler’s worker node was drained during a cluster upgrade on the 3rd and nobody noticed until the 11th. Eight days, eleven counties, and a change table with a hole in it. The instinct is to run the pipeline now and let it catch up, and that instinct produces two specific wrong outcomes: every parcel that changed during the gap is recorded as changing today, and every intermediate state — the parcel that went R-1 → MU-2 → MU-3 across those eight days — collapses into one change that never happened. This guide replays the gap properly. It is the recovery procedure that scheduling & orchestration for municipal feed runs assumes exists, and it depends entirely on having kept the source archive.

Diagnosis: what a naive catch-up actually loses

Run today’s pipeline against today’s payload after an eight-day gap and three things go wrong at once.

Effective dates collapse to today. The pipeline stamps its own run date onto changes whose ordinances took effect during the gap. Every downstream question of the form “what was this parcel zoned on the 6th?” now answers wrongly, and the bitemporal history records a belief you never held.

What a naive catch-up loses, and whether it can be recovered later A four-row by three-column matrix of what a single catch-up run costs after an eight-day gap. Rows are effective dates collapsing to today, intermediate states vanishing, change volume arriving as one burst, and notifications firing for historical changes. Columns are what the pipeline records, whether the damage is recoverable afterwards, and what prevents it. Collapsed effective dates are recoverable by retraction and replay. Vanished intermediate states are recoverable only from the source archive, and not at all without it. What a naive catch-up loses, and whether it can be recovered later what gets recorded recoverable later? prevented by Effective dates collapse to today today, for every change yes — retract by run_id valid_from from the payload Intermediate states vanish only the end state only from the archive replay each payload in order Eight days arrive as one burst one enormous change set yes one run per payload Notifications fire for old changes stale alerts as news no — they already left suppress, then summarise fine repairable lost or already out
The second row is the one that cannot be undone by better code later: a parcel that went R-1 to MU-2 to MU-3 inside the gap has a middle state that exists only in the payload published on the 6th. Without the archive it is gone, which is what makes archiving raw payloads the load-bearing decision here.

Intermediate states vanish. A parcel rezoned twice inside the gap appears to have gone straight from its old code to its newest one. That is not a smaller error than the first — it is a permanently unrecoverable one, because the middle state exists only in the payload published on the 6th, and you are no longer looking at that payload.

Change volume is misattributed. One run emits eight days of change as a single burst, which trips every volume-based check you have and looks exactly like the restatement described in managing watermarks when a portal restates history. Your own recovery now needs recovering from.

Before doing anything, establish what you can actually replay:

def assess_gap(store, archive, source_id, gap_start, gap_end):
    """What is replayable, and what is simply lost. Run this BEFORE deciding."""
    archived = archive.list_payloads(source_id, gap_start, gap_end)   # sorted by published_at
    wm = store.load(source_id)

    return {
        "watermark_at": wm.published_at,
        "archived_payloads": len(archived),
        "publisher_dates": [p.published_at.date() for p in archived],
        # A payload published inside the gap that we never fetched is NOT in the
        # archive — the archive only holds what we downloaded. Gaps in the archive
        # are gaps in the replay, and they cannot be manufactured.
        "replayable": [p for p in archived if p.published_at > wm.published_at],
        "unrecoverable_days": count_publication_days_missing(source_id, gap_start, gap_end,
                                                            archived),
    }

That last field is the honest part. If the archive holds nothing from the gap — because the scheduler that fetches is the same one that was down — then the intermediate states are gone and no procedure recovers them. What you can still do correctly is establish the end state with an accurate effective date, and record explicitly that the path was not observed.

Step-by-step implementation

1. Freeze the schedule before you start

Disable the normal schedule for the affected sources. A backfill racing the nightly run produces interleaved writes whose order depends on timing, and the whole point of this procedure is deterministic ordering.

2. Replay in publisher order, not fetch order 

def backfill(store, archive, differ, source_id, gap_start, gap_end, *, notify=False):
    """Replay archived payloads oldest-first, stamping each with ITS OWN publisher
    date. Notification is off by default: subscribers do not want eight days of
    history delivered as news."""
    plan = assess_gap(store, archive, source_id, gap_start, gap_end)
    if plan["unrecoverable_days"]:
        log.warning("%s: %d publication day(s) are not in the archive and cannot be "
                    "replayed", source_id, plan["unrecoverable_days"])

    run_id = uuid4()
    applied = []
    for payload in plan["replayable"]:            # ascending published_at
        changes = differ.diff(store.current_snapshot(source_id), payload)

        # The effective date comes from the PAYLOAD, never from now(). This is the
        # single line that separates a correct backfill from a wrong one.
        store.apply_versions(source_id, payload, changes,
                             valid_from=payload.published_at,
                             txn_at=datetime.now(timezone.utc),
                             run_id=run_id, backfill=True)

        store.record_changes(changes, run_id=run_id, suppress_notification=not notify)
        store.advance_watermark(source_id, payload)   # one hop at a time
        applied.append((payload.published_at.date(), len(changes)))

    return {"run_id": run_id, "steps": applied,
            "unrecoverable_days": plan["unrecoverable_days"]}

Two properties matter. The watermark advances one payload at a time, so an interrupted backfill resumes from where it stopped rather than restarting. And valid_from comes from the payload, which is what keeps the bitemporal history honest: the transaction time says “we learned this on the 11th,” the valid time says “it took effect on the 6th,” and both are true. That is exactly the double-bind the model in temporal versioning & snapshots exists to hold.

Versions written per publication date, replay against catch-up Line chart over the four publication dates in an eight-day gap, showing how many parcel versions each approach writes with that date as the effective date. An ordered replay writes one version on each of the 4th, 6th and 9th, matching what the county actually published. A single catch-up run writes nothing on those dates and three versions dated the 11th, which is the day the pipeline ran rather than the day anything took effect. Versions written per publication date, replay against catch-up 0 1 2 3 4 4 Aug 6 Aug 9 Aug 11 Aug Publication date inside the gap Versions written with that effective date ordered replay (valid_from from payload) single catch-up run (valid_from = today)
The replay curve reproduces what the county published; the catch-up curve records three versions on a day nothing happened. Only the replay can answer "what was this parcel zoned on the 6th?", and the difference costs one loop over archived payloads.

3. Suppress notification, then send one summary

Replaying eight days with notifications enabled sends subscribers a week of stale alerts, and the idempotency key will not save you — these are genuinely distinct changes that were genuinely never sent. The correct behaviour is to suppress per-change notification during the backfill and send one summary afterwards.

def summarise_backfill(result, source_id):
    total = sum(n for _d, n in result["steps"])
    return (f"Backfill for {source_id}: {total} changes recovered across "
            f"{len(result['steps'])} publication days "
            f"({result['steps'][0][0]} to {result['steps'][-1][0]}). "
            f"{result['unrecoverable_days']} day(s) had no archived payload and were "
            f"not replayed. Change details are queryable by run_id "
            f"{result['run_id']}.")

One message naming the run id is more useful to a subscriber than 340 individual alerts about last week, and it is honest about the days that could not be replayed.

4. Re-enable the schedule and verify the handoff

The first normal run after a backfill should report few changes. If it reports many, the backfill left the watermark behind the true current state and the normal run is now re-doing the gap — which is the double-count this whole guide exists to prevent.

Verification & testing 

def test_backfill_preserves_intermediate_states(store, archive, differ):
    archive.seed("larimer", [
        payload(published_at="2026-08-04", parcel="0714-22-1", code="R-1"),
        payload(published_at="2026-08-06", parcel="0714-22-1", code="MU-2"),
        payload(published_at="2026-08-09", parcel="0714-22-1", code="MU-3"),
    ])
    backfill(store, archive, differ, "larimer", date(2026, 8, 3), date(2026, 8, 11))

    history = store.versions("larimer", "0714-22-1")
    assert [v.zoning_code for v in history] == ["R-1", "MU-2", "MU-3"]
    assert [v.valid_from.date() for v in history] == [
        date(2026, 8, 4), date(2026, 8, 6), date(2026, 8, 9)]      # payload dates
    assert all(v.txn_at.date() == date.today() for v in history)   # learned today


def test_backfill_then_normal_run_does_not_double_count(store, archive, differ, publisher):
    backfill(store, archive, differ, "larimer", date(2026, 8, 3), date(2026, 8, 11))
    before = publisher.published_change_count()
    consume(source("larimer"), store, Fetcher(), differ, publisher)   # the normal run
    assert publisher.published_change_count() == before   # nothing left to find


def test_interrupted_backfill_resumes(store, archive, differ):
    with pytest.raises(InjectedFailure):
        backfill(store, archive, differ, "larimer", ..., fail_after_payloads=2)
    assert store.load("larimer").published_at.date() == date(2026, 8, 6)  # advanced twice
    backfill(store, archive, differ, "larimer", ...)                      # completes
    assert store.load("larimer").published_at.date() == date(2026, 8, 9)

Failure recovery

The backfill already ran with notifications on. Send one correction naming the run id and explaining that the alerts described historical changes. Do not attempt to individually retract 340 notifications; subscribers need the frame, not the list.

Three ways a backfill goes wrong, and the repair A three-row by three-column matrix of backfill failures. Rows are a backfill that ran with notifications enabled, a backfill that stamped today as the effective date, and an archive with missing days. Columns are how the problem is detected, the repair, and what remains permanently affected. Notifications cannot be recalled and are corrected with one message. Wrong effective dates are retracted by run id and replayed. Missing archive days cannot be reconstructed and are recorded as an explicit gap. Three ways a backfill goes wrong, and the repair detected by repair permanently affected Ran with notifications on subscriber complaints one correction message alerts already sent Stamped valid_from as today versions dated the run day retract by run_id, replay nothing Archive has missing days assess_gap before starting record an explicit gap those dates clean needs a notice irreversible
Two of the three are repairable because the whole backfill shares one run id, which makes selective retraction a single query. The third is not repairable at all — and recording an explicit "not observed" gap is a better answer to a query about those dates than a confident wrong one.

The backfill stamped valid_from as today. This is the damaging case, because the versions now assert effective dates that are wrong. Retract the backfill’s versions by run_id — which is why every version carries one — and replay correctly. The retraction is clean precisely because the whole backfill shares a single run id.

The archive has holes. Establish the current state with today’s payload and its own publisher date, and write an explicit gap record for the unobserved period: source, start, end, and the reason. A consumer asking “what changed on the 6th?” then gets “not observed” instead of a confident wrong answer, which is the only defensible option once the payloads are gone.

Two counties need different gaps. Backfill per source, never as one batch across sources. Their publication dates differ, their archives differ, and a shared run id across eleven counties makes selective retraction impossible.

Frequently asked questions

Can I just run the pipeline once and let it catch up?

Only if you do not care about effective dates or intermediate states. A single catch-up run stamps eight days of change with today’s date and collapses any parcel that changed twice into one change that never happened. It also emits the whole gap as one burst, which trips volume checks and looks indistinguishable from a source restatement.

What if there is no source archive?

Then the intermediate states are unrecoverable and no procedure invents them. Establish the current state using today’s payload with its own publisher date, and write an explicit gap record for the unobserved period so queries about those dates return “not observed” rather than a wrong answer. This is the strongest practical argument for archiving raw payloads: the archive is what makes a scheduler outage a replay instead of a permanent hole.

Should the watermark advance once at the end or once per payload?

Once per payload. Advancing per payload makes an interrupted backfill resumable — it restarts from the last replayed publication rather than from the beginning — and it keeps the watermark consistent with the versions actually written. A single advance at the end means an interruption leaves the watermark describing a state you have only partly applied.

How do I tell a subscriber that a week of history arrived at once?

One summary message naming the source, the date range, the change count, the number of days that could not be replayed, and the run id to query for detail. Subscribers can act on that; they cannot act on 340 individual notifications about last week, and after the first dozen they will filter the channel.