Scheduling Incremental Geocode Refresh with Cron

TL;DR: Schedule a cron job that re-geocodes only the rows whose canonical address hash changed or whose geocoded_at timestamp is older than the cache TTL, bounded by a persisted watermark and made idempotent with an upsert keyed by record id. This page is part of automating CI/CD sync for spatial enrichment.

The selection predicate

A row needs re-geocoding for exactly one of two reasons: its source address changed, or its cached result aged past the TTL. Both are queryable if each row carries a canonical hash of its normalized address and a geocoded_at timestamp. The change-data-capture query selects the union of the two, ordered so the oldest work drains first.

-- Select rows to re-geocode: new, TTL-expired, or changed since the watermark.
SELECT a.id, a.raw_address, a.addr_hash, a.updated_at
FROM address_source AS a
LEFT JOIN geocode_enriched AS g ON g.id = a.id
WHERE
    -- 1. New rows never geocoded
    g.id IS NULL
    -- 2. Cache TTL expired (re-verify against provider drift), any age
    OR g.geocoded_at < now() - INTERVAL '30 days'
    -- 3. Source address changed since last enrichment AND touched since the
    --    watermark (the watermark keeps this changed-row scan cheap).
    OR (g.addr_hash IS DISTINCT FROM a.addr_hash AND a.updated_at > :watermark)
ORDER BY COALESCE(g.geocoded_at, 'epoch'::timestamptz) ASC
LIMIT :batch_size;

The addr_hash is computed once at ingestion over the canonical address only — never over volatile fields like a load timestamp — so a row that was merely re-loaded without a real change does not appear stale.

What Qualifies a Row for Refresh

The selection predicate is the whole job. Too broad and every run re-enriches the world; too narrow and stale rows never get revisited. Four conditions cover the cases that genuinely deserve another lookup, and they compose as an OR.

Four conditions that make a row eligible Four conditions with their typical nightly volume. Rows never enriched are the largest and most obvious group. Low-precision rows past an age threshold are revisited in case a better match now exists. Rows whose address text changed must be re-enriched. Rows explicitly marked for recall after a provider or data change form the fourth group. never enriched geocoded_at IS NULL new rows since the last run the bulk of a healthy nightly run low precision and old tier < 'parcel' AND age > 30 d a better match may exist now cap this group per run address text changed key <> last_geocoded_key the canonical key moved, so the stored coordinate is for another address marked for recall recall_requested = true set in bulk after a provider change gives you a deliberate replay lever

Cap the second group explicitly. Without a per-run limit, the first night after the threshold is introduced sweeps up every low-precision row ever stored and turns a routine refresh into the most expensive job of the quarter. A limit spreads the same work over a fortnight at a predictable cost.

Predicate and edge-case breakdown

Predicate / concern Selects / handles Why it matters
g.id IS NULL Never-geocoded new rows Backfills freshly ingested addresses
addr_hash IS DISTINCT FROM Rows whose canonical address changed Catches corrections and re-normalizations; IS DISTINCT FROM treats NULLs safely
geocoded_at < now() - INTERVAL '30 days' Cache TTL expiry Re-verifies against provider reference-data drift
updated_at > :watermark Recently changed rows only Keeps the changed-row scan cheap on large tables
ORDER BY geocoded_at ASC Oldest results first Bounds worst-case staleness under a batch limit
LIMIT :batch_size Caps work per run Prevents a backfill storm from exhausting quota
Clock skew between DB and job host Watermark comparisons Use the database clock (now()), not the host clock, for all timestamps
Backfill storm after a bulk re-normalization Millions of hash changes at once The LIMIT plus oldest-first ordering spreads the work across runs

The idempotent refresh job

The job reads the watermark, pulls one bounded batch of stale rows, geocodes them, upserts the results keyed by id, then advances the watermark only after the batch commits. Because writes are upserts and the watermark advances last, a crash mid-run simply reprocesses the same batch on the next invocation — no duplicates, no skipped rows.

from __future__ import annotations

import hashlib
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Callable, Optional

import psycopg
from psycopg.rows import dict_row

# Geocode function: address -> (lat, lon, precision) or None on no-match.
GeocodeFn = Callable[[str], Optional[tuple]]

_BATCH_SIZE = 2000
_TTL_DAYS = 30


def canonical_hash(address: str) -> str:
    """Stable SHA-256 over the canonical address, used for change detection."""
    return hashlib.sha256(address.strip().lower().encode("utf-8")).hexdigest()


@dataclass
class RefreshStats:
    """Outcome of one incremental run."""

    selected: int
    geocoded: int
    failed: int


def read_watermark(conn: psycopg.Connection) -> datetime:
    """Fetch the persisted high-water timestamp, defaulting to the epoch."""
    row = conn.execute("SELECT ts FROM sync_watermark WHERE job = 'geocode'").fetchone()
    if row and row[0]:
        return row[0]
    return datetime(1970, 1, 1, tzinfo=timezone.utc)


def run_incremental_refresh(dsn: str, geocode: GeocodeFn) -> RefreshStats:
    """Re-geocode one bounded batch of stale rows and advance the watermark.

    Idempotent: results are upserted by id, and the watermark advances only
    after a successful commit, so a crash reprocesses rather than skips.
    """
    with psycopg.connect(dsn, row_factory=dict_row) as conn:
        watermark = read_watermark(conn)

        rows = conn.execute(
            """
            SELECT a.id, a.raw_address, a.addr_hash, a.updated_at
            FROM address_source AS a
            LEFT JOIN geocode_enriched AS g ON g.id = a.id
            WHERE g.id IS NULL
               OR g.geocoded_at < now() - (%(ttl)s || ' days')::interval
               OR (g.addr_hash IS DISTINCT FROM a.addr_hash
                   AND a.updated_at > %(watermark)s)
            ORDER BY COALESCE(g.geocoded_at, 'epoch'::timestamptz) ASC
            LIMIT %(batch)s
            """,
            {"ttl": _TTL_DAYS, "watermark": watermark, "batch": _BATCH_SIZE},
        ).fetchall()

        geocoded = 0
        failed = 0
        max_updated = watermark
        for row in rows:
            result = geocode(row["raw_address"])
            if result is None:
                failed += 1
                continue
            lat, lon, precision = result
            conn.execute(
                """
                INSERT INTO geocode_enriched
                    (id, lat, lon, precision, addr_hash, geocoded_at)
                VALUES (%(id)s, %(lat)s, %(lon)s, %(precision)s, %(hash)s, now())
                ON CONFLICT (id) DO UPDATE SET
                    lat = EXCLUDED.lat,
                    lon = EXCLUDED.lon,
                    precision = EXCLUDED.precision,
                    addr_hash = EXCLUDED.addr_hash,
                    geocoded_at = EXCLUDED.geocoded_at
                """,
                {
                    "id": row["id"],
                    "lat": lat,
                    "lon": lon,
                    "precision": precision,
                    "hash": row["addr_hash"],
                },
            )
            geocoded += 1
            if row["updated_at"] and row["updated_at"] > max_updated:
                max_updated = row["updated_at"]

        # Advance the watermark last, in the same transaction as the writes.
        conn.execute(
            """
            INSERT INTO sync_watermark (job, ts) VALUES ('geocode', %(ts)s)
            ON CONFLICT (job) DO UPDATE SET ts = EXCLUDED.ts
            """,
            {"ts": max_updated},
        )
        conn.commit()

    return RefreshStats(selected=len(rows), geocoded=geocoded, failed=failed)

Vectorized pandas variant

When the geocoder is a batch API, pull the stale delta into a DataFrame, geocode the column in one call, and write results back with an upsert. This amortizes provider round-trips across the whole batch.

import pandas as pd


def refresh_frame(stale: pd.DataFrame, batch_geocode) -> pd.DataFrame:
    """Geocode a DataFrame of stale rows in bulk and return rows ready to upsert.

    stale needs columns 'id', 'raw_address', 'addr_hash'. batch_geocode maps a
    list of addresses to a list of (lat, lon, precision) or None.
    """
    results = batch_geocode(stale["raw_address"].tolist())
    stale = stale.assign(
        lat=[r[0] if r else None for r in results],
        lon=[r[1] if r else None for r in results],
        precision=[r[2] if r else None for r in results],
    )
    ready = stale.dropna(subset=["lat", "lon"]).copy()
    ready["geocoded_at"] = pd.Timestamp.now(tz="UTC")
    return ready[["id", "lat", "lon", "precision", "addr_hash", "geocoded_at"]]

Idempotent by Construction

Cron will run the job twice — after a manual retry, a duplicated schedule, or a node that did not die cleanly. The job must produce the same end state either way, which means the work must be claimed atomically and the write must be an upsert rather than an insert.

Claim, enrich, upsert, release A four-step sequence with a lease. The job claims a batch by updating rows with a lease identifier and expiry in one statement. It enriches the claimed rows, upserts the results keyed on the row identifier, and clears the lease. A concurrent second run claims nothing because the rows are already leased, and an expired lease returns rows to the pool. claim one UPDATE sets a lease id and expiry enrich cache first, provider only for the misses upsert keyed on row id, never a bare insert release clear the lease in the same commit A second run starting mid-flight claims nothing, because the rows already carry an unexpired lease. A run that dies without releasing leaves rows leased until expiry, after which they return to the pool automatically — no manual cleanup, and no permanent deadlock from a crashed worker.

The lease expiry is what makes this safe without a coordinator. Set it comfortably longer than a normal batch and shorter than the schedule interval, so a crashed run’s work is picked up by the next one rather than by a concurrent one — and no row is ever geocoded twice in the same window.

Scheduling with cron

Run the job on a fixed cadence with a lock so a slow run never overlaps the next tick. flock gives a simple host-level mutex; in a distributed scheduler, use an advisory lock in the database instead.

# /etc/cron.d/geocode-refresh
# Every 15 minutes, non-overlapping. Logs to a rotated file.
*/15 * * * * appuser flock -n /tmp/geocode-refresh.lock \
  /opt/pipeline/.venv/bin/python -m pipeline.refresh >> /var/log/geocode-refresh.log 2>&1

A 15-minute cadence with a 2,000-row batch drains routine change traffic quickly while keeping each run well inside provider quota. For the throughput of the geocoding step inside each run, the concurrency patterns in optimizing batch geocoding throughput apply directly.

Scheduling That Survives Contact With Reality

Cron expressions encode assumptions that break quietly: that the previous run finished, that the machine is in the timezone you think, and that a missed window will be retried. Four settings cover the gap between a schedule that looks right and one that behaves.

Four scheduling settings and the failures they prevent Four rows. Scheduling in UTC prevents the run from moving or repeating at daylight-saving transitions. Overlap prevention stops a slow run being joined by the next one. An explicit catch-up policy decides whether a missed window is skipped or run late. A jittered start prevents every service on the fleet firing at exactly the same minute. Setting Failure it prevents schedule in UTC a daylight-saving shift skipping or doubling a run prevent overlap a slow run joined by the next one, doubling provider load explicit catch-up policy a maintenance window silently losing a night of enrichment jittered start minute every job on the fleet hitting the provider at 02:00:00

The overlap case is the one that produces the most memorable incidents. A run that normally takes forty minutes but occasionally takes ninety will, on those nights, be running when the next one starts — and two concurrent runs against the same quota guarantee both are throttled, which makes them slower, which makes the overlap worse.

Edge cases

Clock skew

If the job host clock runs ahead of the database, a watermark written from the host can skip rows updated in the gap. Derive every timestamp from the database (now() and the watermark column) rather than the host clock, so all comparisons happen against a single monotonic source.

Backfill storm

A bulk re-normalization flips millions of addr_hash values at once, and every one becomes eligible. The LIMIT and oldest-first ordering turn that into a controlled drain across many runs instead of one quota-exhausting surge. If you must clear the backlog faster, temporarily raise batch_size while watching spend against a ceiling, as covered in API quota tracking and cost management.

Integration note

This scheduled refresh is the sync half of automating CI/CD sync for spatial enrichment: the CI gate keeps the pipeline correct on merge, while this cron job keeps the enriched dataset current between merges. The freshly refreshed rows should still pass confidence thresholds — validate them with the scoring described in validating geocoding accuracy and confidence scoring before they reach consumers. For a zero-downtime cutover on large refreshes, build the delta into a shadow table and swap it as described in the parent guide.