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.

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"]]

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.

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.