Invalidating tile caches when a rezone lands

def audit_invalidation(change, purged_urls, pyramid_manifest):
    """What SHOULD have been purged against what was. Run this after an incident."""
    expected = set()
    for geom in (change.geometry_before, change.geometry_after):
        if geom is None:
            continue
        expected |= tiles_for_bounds(*geom.bounds,
                                     zooms=pyramid_manifest["layers"]["parcels"])
    expected_urls = {f"/tiles/{t.key()}.pbf" for t in expected}
    return {"expected": len(expected_urls),
            "purged": len(purged_urls),
            "missed": sorted(expected_urls - set(purged_urls)),
            "extra": sorted(set(purged_urls) - expected_urls)}

import math
from dataclasses import dataclass


@dataclass(frozen=True)
class Tile:
    z: int
    x: int
    y: int

    def key(self) -> str:
        return f"{self.z}/{self.x}/{self.y}"


def tiles_for_bounds(minx, miny, maxx, maxy, zooms) -> set[Tile]:
    """Web-Mercator coverage for a lon/lat bounding box, INCLUSIVE on both ends.

    The inclusive range is the whole point: a parcel whose edge lands exactly on a
    tile boundary is clipped into both neighbours, and a truncating calculation
    returns only one. That is the stale-seam bug, and it survives every test because
    it needs one parcel, at one zoom, on one edge.
    """
    out = set()
    for z in zooms:
        n = 2 ** z
        x0 = int(math.floor((minx + 180.0) / 360.0 * n))
        x1 = int(math.floor((maxx + 180.0) / 360.0 * n))

        def ytile(lat):
            r = math.radians(max(min(lat, 85.05112878), -85.05112878))
            return int(math.floor((1.0 - math.asinh(math.tan(r)) / math.pi) / 2.0 * n))

        y0, y1 = ytile(maxy), ytile(miny)          # note: y is inverted
        for x in range(max(0, x0), min(n - 1, x1) + 1):
            for y in range(max(0, y0), min(n - 1, y1) + 1):
                out.add(Tile(z, x, y))
    return out

GEOMETRY_CLASSES = {"boundary_adjustment", "split", "merge"}
ATTRIBUTE_CLASSES = {"rezone", "overlay_only", "attribute_only"}


def plan_invalidation(changes, manifest) -> dict:
    parcel_zooms = range(manifest["layers"]["parcels"][0],
                         manifest["layers"]["parcels"][1] + 1)
    state_ids, rebuild = set(), set()

    for ch in changes:
        if ch.change_class in ATTRIBUTE_CLASSES:
            state_ids.add(ch.feature_id)           # a file write, not a tile rebuild
        elif ch.change_class in GEOMETRY_CLASSES:
            for geom in (ch.geometry_before, ch.geometry_after):
                if geom is not None:
                    rebuild |= tiles_for_bounds(*geom.bounds, zooms=parcel_zooms)
            state_ids.add(ch.feature_id)
        else:
            # A new change class must not fall through to "attribute only", which
            # would leave a geometry change uninvalidated and the map wrong.
            raise ValueError(
                f"unclassified change {ch.change_class!r}: refusing to guess whether "
                f"it touches geometry")

    return {
        "state_file_ids": sorted(state_ids),
        "tiles_to_rebuild": sorted(t.key() for t in rebuild),
        "purge_urls": [f"/tiles/{k}.pbf" for k in sorted(t.key() for t in rebuild)],
    }

import logging

log = logging.getLogger(__name__)


def purge_and_confirm(cdn, urls, *, sample=0.1) -> dict:
    """Purge APIs are eventually consistent and occasionally silent. An unconfirmed
    purge is the mechanism by which a correct pipeline publishes a wrong map."""
    accepted, rejected = cdn.purge(urls)
    if rejected:
        log.error("purge rejected %d of %d URLs", len(rejected), len(urls))

    # Verify a sample by requesting with a cache-revealing header and checking the age.
    import random
    checks = random.sample(accepted, max(1, int(len(accepted) * sample)))
    stale = []
    for url in checks:
        resp = cdn.fetch(url, headers={"Cache-Control": "no-cache"})
        if resp.headers.get("age") and int(resp.headers["age"]) > 60:
            stale.append(url)

    return {"requested": len(urls), "accepted": len(accepted),
            "rejected": rejected, "sampled": len(checks), "still_stale": stale,
            "confirmed": not rejected and not stale}

def test_before_and_after_geometry_are_both_invalidated():
    ch = change(cls="boundary_adjustment",
                before=box(-105.10, 40.50, -105.09, 40.51),
                after=box(-105.12, 40.50, -105.11, 40.51))     # moved west
    plan = plan_invalidation([ch], MANIFEST)
    after_only = tiles_for_bounds(*ch.geometry_after.bounds, zooms=range(13, 17))
    assert len(plan["tiles_to_rebuild"]) > len(after_only), \
        "the old extent was not invalidated — a ghost will remain"


def test_a_parcel_on_a_tile_boundary_hits_both_tiles():
    """Place the parcel's edge exactly on a z14 tile boundary."""
    edge_lon = tile_west_edge_lon(z=14, x=3423)
    ch = change(cls="boundary_adjustment",
                before=None,
                after=box(edge_lon - 0.0002, 40.50, edge_lon + 0.0002, 40.501))
    tiles = {t for t in tiles_for_bounds(*ch.geometry_after.bounds, zooms=[14])}
    assert len({t.x for t in tiles}) == 2, "the seam tile was missed"


def test_a_rezone_rebuilds_no_tiles():
    plan = plan_invalidation([change(cls="rezone")], MANIFEST)
    assert plan["tiles_to_rebuild"] == []
    assert plan["state_file_ids"]


def test_an_unknown_change_class_raises():
    with pytest.raises(ValueError, match="refusing to guess"):
        plan_invalidation([change(cls="something_new")], MANIFEST)


def test_zoom_range_comes_from_the_manifest():
    manifest = dict(MANIFEST, layers={"parcels": [13, 17]})     # a new zoom added
    plan = plan_invalidation([change(cls="split")], manifest)
    assert any(k.startswith("17/") for k in plan["tiles_to_rebuild"])


def test_unconfirmed_purge_is_reported_not_swallowed():
    result = purge_and_confirm(FlakyCDN(reject=["/tiles/14/3423/6192.pbf"]), URLS)
    assert result["confirmed"] is False and result["rejected"]