Scheduling & Orchestration for Municipal Feed Runs

A county publishes its parcel layer “weekly.” In practice it publishes on Tuesday most weeks, on Wednesday when Monday is a holiday, twice on the Thursday after a planning-commission meeting, and not at all during the last week of the fiscal year. Your scheduler, meanwhile, runs at 02:00 daily and does not care. The result is the failure mode this page addresses: a pipeline that is technically running on schedule and is nevertheless ingesting nothing new for nine days, then ingesting the same restated history three times in one morning. Scheduling municipal feeds is not a cron problem — it is a problem of tracking what you have already correctly consumed from sources that have no obligation to tell you when they changed. This topic sits at the front of the Automated Feed Ingestion & GIS Data Parsing area, upstream of everything else in it: the order and repeatability of runs is what makes async batch processing worth optimising at all.

Prerequisites and operational context jump to heading

Before orchestration is worth designing, three things have to exist. Each of them is a precondition rather than a nicety, and skipping any one of them produces a scheduler that appears to work and quietly loses data.

The first is a durable watermark per source. Not a global “last run” timestamp, but one record per jurisdiction naming what that jurisdiction had published the last time you successfully consumed it. A single global watermark is the most common design error in municipal ingestion, because sources move independently: advancing one watermark after a run in which three of eleven counties failed marks the failed counties as consumed.

The second is idempotent task bodies. Municipal orchestration involves reruns constantly — a portal times out, a schema check fails, a deploy lands mid-run — and a task that cannot be re-executed safely turns every one of those ordinary events into a data-repair job. Idempotency here means the task computes the same target state from the same inputs, not merely that it “does not crash twice.”

The third is a content digest for every fetched artifact. Municipal portals frequently republish identical content with a new timestamp, and just as frequently change content while leaving the timestamp alone. Neither an HTTP Last-Modified header nor a file name is reliable evidence that something changed; only a digest over the bytes is. This is the same digest that data lineage & provenance tracking needs downstream, so it is not extra work — it is work you were going to do anyway, done earlier.

With those in place, a run becomes describable as a pure function of (source, watermark, digest) rather than of the wall clock, and everything else on this page follows from that.

Architecture: a dependency graph over jurisdictions, not over stages jump to heading

The intuitive orchestration design puts pipeline stages in the graph: fetch, then normalise, then validate, then publish. It is intuitive and it is wrong for this domain, because it couples jurisdictions that have nothing to do with each other. If fetch is one task for all eleven counties, then one county’s outage blocks normalisation for the ten that succeeded.

The design that survives production inverts the graph. Each jurisdiction gets its own independent chain, and the only shared node is the publication gate at the end.

The publication gate at the end is deliberately shared, because publishing is the one operation where consistency across jurisdictions matters: a consumer reading a regional dataset should not see county A’s Tuesday data joined to county C’s data from three weeks ago without being told. The gate publishes what passed and records what was held, which is exactly the tiering that fallback routing logic then serves from.

Choosing a trigger, not just an interval jump to heading

A schedule answers “when do we look?” — it does not answer “is there anything new?” Those need separate mechanisms, and conflating them is what produces the nine-quiet-days-then-three-runs pattern.

The workable combination for most portals is a frequent cheap check with a conditional full fetch behind it: poll HEAD hourly, and only run the expensive chain when the validator says something moved. Where headers cannot be trusted — which is common — the cheap check has to be a small ranged request or a metadata endpoint, and the digest becomes the authority. Whatever the mechanism, the rate at which you check is governed by municipal API rate limit management, and an hourly HEAD across eleven counties is well inside any reasonable budget.

Production implementation jump to heading

The core of a municipal scheduler is smaller than most teams expect. What follows is the watermark-and-digest logic that decides whether a chain should run at all; it is deliberately framework-agnostic, because it is the part that has to be correct regardless of whether Airflow, Prefect, or a cron script is calling it.

import hashlib
import logging
from dataclasses import dataclass, replace
from datetime import datetime, timezone

log = logging.getLogger(__name__)


@dataclass(frozen=True)
class Watermark:
    """What we have already correctly consumed from one source."""
    source_id: str
    published_at: datetime | None   # the publisher's own timestamp, if it has one
    content_digest: str | None      # sha256 over the fetched bytes
    consumed_at: datetime | None    # when OUR run committed it
    run_id: str | None


class SkipRun(Exception):
    """Raised to mean 'nothing new' — not an error, and not retried."""


def decide(source, wm: Watermark, probe) -> str:
    """Return 'run', 'skip', or 'restated'.

    `probe` is a cheap HEAD-style result: (published_at, etag, size).
    The digest is only computed after a fetch, so this function decides
    whether the fetch is worth doing at all.
    """
    published_at, etag, size = probe

    if wm.content_digest is None:
        return "run"                       # never consumed: always run

    # A publisher timestamp that moved forward is the strongest cheap signal.
    if published_at and wm.published_at and published_at > wm.published_at:
        return "run"

    # Some portals never move the timestamp but do change the payload size.
    if size is not None and wm.size_hint is not None and size != wm.size_hint:
        log.info("%s: size changed with a static timestamp — fetching", source.id)
        return "run"

    if etag and etag == wm.etag:
        return "skip"

    # No usable signal at all. Fetch on a floor interval so a portal that
    # lies about everything is still picked up within one day.
    if wm.consumed_at and (datetime.now(timezone.utc) - wm.consumed_at).days >= 1:
        return "run"

    return "skip"


def consume(source, store, fetch) -> Watermark:
    """Fetch, compare digests, and advance the watermark only on success."""
    wm = store.load(source.id) or Watermark(source.id, None, None, None, None)

    verdict = decide(source, wm, fetch.probe(source))
    if verdict == "skip":
        raise SkipRun(f"{source.id}: no change since {wm.consumed_at}")

    payload = fetch.get(source)                       # the expensive call
    digest = hashlib.sha256(payload.body).hexdigest()

    if digest == wm.content_digest:
        # The publisher moved a timestamp without changing content. Record that
        # we looked, so the floor interval does not re-fetch in an hour, but do
        # NOT emit a change downstream.
        log.info("%s: republished identical content", source.id)
        return store.save(replace(wm, consumed_at=datetime.now(timezone.utc)))

    if wm.published_at and payload.published_at and payload.published_at < wm.published_at:
        # The publisher went BACKWARDS: a restatement of history.
        log.warning("%s: restated history (%s < %s)", source.id,
                    payload.published_at, wm.published_at)
        raise Restatement(source.id, payload)         # handled out of band

    # Only now is it safe to hand the payload downstream.
    return store.save(Watermark(
        source_id=source.id,
        published_at=payload.published_at,
        content_digest=digest,
        consumed_at=datetime.now(timezone.utc),
        run_id=payload.run_id,
    ))

Two properties of that code matter more than the details. First, the watermark is written after the downstream commit, never before — a watermark advanced ahead of the data it describes is indistinguishable from successful consumption, and the records it skipped are gone silently. Second, SkipRun is not an error. A scheduler that reports “nothing new” as a failure trains its operators to ignore failures, and a scheduler that reports it as success without distinguishing it from real work makes “did anything happen last night?” unanswerable.

What the run should record jump to heading

Every chain execution should leave a row that makes the decision auditable later, whether or not it did any work:

Edge cases and gotchas jump to heading

A portal restates history. A county reloads its parcel layer from an older backup, so the newest published file describes an older state. Handled naively, the next run treats months of stale records as fresh changes and emits thousands of spurious rezones. This is why decide() treats a backwards-moving publisher timestamp as a distinct verdict rather than as ordinary change — the run must stop and be looked at, because the correct response depends on why it happened.

Two runs overlap. A slow run is still going when the next fires. Without a per-source lock, both fetch, both write, and the watermark ends up describing whichever finished last — which may be the older one. The fix is a lease keyed on source_id, not a global lock: a global lock reintroduces exactly the cross-jurisdiction coupling the graph shape was designed to remove.

A schema change arrives mid-run. Half the counties are ingested against the old contract and half against the new one. Because the publication gate is shared, this is recoverable: hold the whole batch rather than publishing a mixed one, and let the schema validation & data quality checks tier decide which contract is authoritative.

The floor interval hides a dead portal. A source whose signals are all unusable gets fetched once a day by the floor interval. If that fetch always returns identical content, the pipeline is stable and the data is frozen. Track time since content last changed per source, not just time since last run — a parcel layer that has not changed in six weeks is either a very quiet county or a portal that has stopped updating, and the distinction is worth an alert.

Daylight-saving transitions. A schedule expressed in local time either runs twice or not at all on the transition day. Municipal effective dates are inherently local, so store and compare instants in UTC and convert only for display; the one place local time genuinely matters is interpreting an ordinance’s effective date, which is a data question rather than a scheduling one.

Integration points jump to heading

Orchestration is the layer everything else in the ingestion area hangs from, and it has two hard interfaces.

Downstream, each chain hands a validated batch plus a watermark candidate to the publication gate. The gate is where GIS export sync workflows take over: it is the point at which a delta becomes publishable, and the point at which a held jurisdiction becomes a fallback-tier decision rather than an ingestion problem.

Upstream, the scheduler is the only component that should know about wall-clock time at all. Fetch code takes a source and returns bytes; parsers take bytes and return records; none of them should ask what day it is. Keeping time in one layer is what makes the whole pipeline testable, which is the argument developed in testing spatial data pipelines — a task whose behaviour depends on datetime.now() inside its body cannot be exercised deterministically.

The retry relationship also runs through here. Orchestrator-level retries and in-task retries solve different problems and must not be layered without thought: an orchestrator retrying a task that already contains a five-attempt backoff loop produces twenty-five attempts against a struggling portal. Set one of them to retry and the other to fail fast, and let the error handling & retry logic design decide which.

Compliance and audit artifacts jump to heading

Because the scheduler decides what was consumed and when, its records are part of the audit trail whether or not they were designed to be.

Three artifacts should survive every run. The run row described above, which answers “why did this chain run, or not.” The watermark history — the watermark table should be append-only or at minimum versioned, because a watermark that was advanced incorrectly is the root cause of a whole class of missing-data incidents and the only way to prove that is to see its previous value. And the held-jurisdiction record: when the publication gate holds a county, that decision, its reason, and the tier consumers were served instead need to be recorded, because a consumer who made a decision during the hold will eventually ask what they were looking at.

Those three, joined on run_id, are what let an auditor ask “was this parcel’s zoning current on the day we priced this deal?” and receive an answer that includes “no — that county was held for eleven hours and you were served a snapshot from the previous evening, flagged.” That is a defensible answer. “Our pipeline runs nightly” is not.

Choosing a scheduler jump to heading

The framework matters far less than the two properties above, and the honest comparison is short. A cron script with a watermark table and a lease is entirely adequate for a dozen sources, and it has the enormous advantage that it can be read in one sitting. Airflow earns its operational weight when backfills over date ranges become routine and when the dependency graph genuinely branches — its scheduler is built around exactly the “run this logical date again” question that a municipal restatement forces. Prefect sits between them, with the practical benefit that tasks are ordinary Python functions, which keeps the watermark logic above testable without a scheduler running.

What none of them supply is the domain logic. Every one of these tools will happily run a task that advances a watermark before the data commits, and none of them knows that a publisher timestamp moving backwards is a restatement rather than an update. Choosing a scheduler is therefore a much smaller decision than it appears; the design on this page is the part that determines whether the pipeline loses data, and it is identical under all three.

A practical consequence is worth stating plainly: do not begin a municipal ingestion project by installing an orchestrator. Begin with the watermark table, the digest, and the lease, driven by whatever runs a command on a timer. Move to a heavier scheduler when a specific pain appears — usually the first time somebody has to backfill eleven counties across a fortnight by hand — and migrate the same task bodies into it unchanged.

FAQ jump to heading

Related jump to heading

• Section overview: Automated Feed Ingestion & GIS Data Parsing
• Municipal API Rate Limit Management — the budget that decides how often a cheap change check can run
• Error Handling & Retry Logic — where retries belong once the scheduler stops owning them
• Data Lineage & Provenance Tracking — consumes the run records and digests this layer produces
• Testing Spatial Data Pipelines — why the scheduler should be the only component that knows the time