DIE-364: Rebuild pygeoapi after pipeline runs

SPEC.md

@@ -554,11 +554,16 @@ Cloud Run Service (`orchestration/pygeoapi/cloudbuild.yaml`):
 Baked products (§6.3) are only as fresh as the image: a pipeline run that publishes new
 data (a combine asset with `skipped_unchanged: false` or `parquet_status: written`) is
 not served until pygeoapi is rebuilt and redeployed with
-`orchestration/pygeoapi/cloudbuild.yaml`. **For now this rebuild is manual**: run the
-build after the scheduled product runs finish, and not while any are still running, so
-the image doesn't copy a half-updated set of products. Automating it (a Cloud Build
-trigger started from Dagster after product runs publish new data) is planned; see
-`docs/pygeoapi-performance-plan.md` T5.
+`orchestration/pygeoapi/cloudbuild.yaml`.
+
+The `pygeoapi_rebuild` sensor (`orchestration/pygeoapi_rebuild.py`) does this. Every
+10 minutes it checks for product jobs (cohort and standalone) that succeeded since its
+last check. Once none are queued or running, it starts the `die-pygeoapi-rebuild` Cloud
+Build trigger (`us-central1`) once. Waiting until no runs are active means the image never
+bakes a half-updated set of products. Sensor ticks use no Dagster+ credits, and the
+sensor launches no Dagster run. It is stopped by default and is turned on only in the
+prod deployment. It authenticates with `GCP_SERVICE_ACCOUNT_KEY`, which needs
+`roles/cloudbuild.builds.editor`.
 
 ---
 

docs/pygeoapi-performance-plan.md

@@ -1,7 +1,7 @@
 # Plan: pygeoapi performance and correctness (post-standup)
 
-Status: T1–T4 done on DIE-363 and deployed (2026-09-24); T5 deferred (rebuild is manual
-for now); T6 partly (format version); T7 partly (per-feature bbox). See "Progress" below.
+Status: T1–T4 done on DIE-363 and deployed (2026-09-24); T5 done on DIE-364
+(`pygeoapi_rebuild` sensor); T6 partly (format version); T7 partly (per-feature bbox). See "Progress" below.
 Picks up from the pygeoapi standup (PR #137, branch
 `DIE-359-deploy-pygeo-api-on-cloud-run`) and the per-product GeoParquet work (branch
 `DIE-363-update-the-layer-generation-code-to-land-in-pygeo-api-instead-of-geo-server`,
@@ -18,7 +18,7 @@ Measured locally: patched image, files on local disk, 4 workers, 2 GiB.
 | T2 pin + patch | done | pages are exactly 5,000 rows, no overlap, 48,809 distinct ids; date-only `datetime` returns 200 |
 | T3 sort | done, changed | sorted by **`id`, `source`**, then `datetime` (not `source` first: with source first the `id` ranges of row groups overlap and well 9033 hit 4 of 9 groups; id first, 1–2). Whole series 1.3–1.4 s vs 1.6–2.5 s time-sorted |
 | T4 bake | done | `--bake-to`; 31/31 collections pass items/bbox/by-id from baked files; 88 MB baked; `limit=10` in 13–35 ms |
-| T5 rebuild | deferred | Not automated yet. Rebuild manually after product runs publish new data (SPEC §6.6) |
+| T5 rebuild | done (DIE-364) | `pygeoapi_rebuild` sensor starts the Cloud Build trigger once product runs finish (SPEC §6.6) |
 | T6 | partial | `PARQUET_FORMAT_VERSION` done (now `"2"`, so every existing `latest.parquet` is rewritten on the next run). Others open |
 | T7 | partial | per-feature `bbox` removed (with the empty-result fix below). The `bbox` covering column is still a property |
 
@@ -36,10 +36,10 @@ memory stayed at 26–29% of 2 GiB, and the latency breakdown is almost all user
   total, so clients can fetch pages in parallel.
 - Not fixed here (separate ticket): WQP `source` labels differ between products. See T6.
 
-Until T5 is done, **each pipeline run that changes data needs a manual pygeoapi build**
-(`orchestration/pygeoapi/cloudbuild.yaml`), started after the product runs finish.
-Whatever account runs it needs `storage.objectViewer` on the products bucket for the
-bake, as well as what builds need today.
+The rebuild (`orchestration/pygeoapi/cloudbuild.yaml`) is started by the
+`pygeoapi_rebuild` sensor after product runs finish (T5). The build's account needs
+`storage.objectViewer` on the products bucket for the bake, as well as what builds need
+today.
 
 New finding — **memory at 2 GiB with 4 workers.** Under sustained 10,000-row page load the
 local container's cgroup peak reached its 2 GiB cap (anon 1.88 GiB; workers 470–650 MB
@@ -208,29 +208,21 @@ Do them in order. Each is independently shippable. Re-run the measurement protoc
   - The build log lists each product's baked file and the reason.
   - All 31 collections return 200.
 
-### T5 — Rebuild after a pipeline run (deferred; manual for now)
-
-Not being done yet. Until it is, rebuild by hand after product runs publish new data
-(SPEC §6.6). Design notes for when it's picked up:
-
-- Create a Cloud Build **trigger** on the GitHub repo, `main` branch, manual invocation,
-  using `orchestration/pygeoapi/cloudbuild.yaml`. Triggers fill `COMMIT_SHA`
-  automatically.
-- Start it when a pipeline run has published new data. Recommended: a Dagster
-  job/sensor that runs after the scheduled product runs complete. If any product's
-  materialization has `parquet_status: written` or `skipped_unchanged: false`, it calls
-  the Cloud Build `projects.triggers.run` API.
-  - One rebuild per run, not per product. Debounce, and never build mid-run, so the image
-    never bakes a half-updated set.
-  - The Dagster+ credentials (`GCP_SERVICE_ACCOUNT_KEY`) need permission to run the
-    trigger (`roles/cloudbuild.builds.editor`, or a narrower custom role).
-  - Simpler fallback: Cloud Scheduler runs the trigger on a fixed cadence after the
-    pipeline window.
-- Schedules today are **monthly and staggered** (`products.yaml`: days 1, 15, 22, 23, 25
-  at 06:00 UTC). Moving to weekly means updating those `schedule:` values. With the
-  current staggering, the trigger fires up to 5 times a month.
-- Accept: a pipeline run that changes a product results in exactly one build, and the new
-  revision serves the new data with no manual step.
+### T5 — Rebuild after a pipeline run (done, DIE-364)
+
+The `pygeoapi_rebuild` sensor (`orchestration/pygeoapi_rebuild.py`, SPEC §6.6) polls
+every 10 minutes. After a product job succeeds, and once no product job is queued or
+running, it calls `projects.locations.triggers.run` on the `die-pygeoapi-rebuild` Cloud
+Build trigger. Several runs finishing on the same day give one build.
+
+- It launches no Dagster run. Sensor ticks cost no Dagster+ credits.
+- It rebuilds after every successful product run, even if the data didn't change.
+  Skipping unchanged runs (`parquet_status`) was left out for simplicity; builds fall
+  within the Cloud Build free tier.
+- Schedules are **monthly and staggered** (`products.yaml`: days 1, 15, 22, 23, 25 at
+  06:00 America/Denver), so expect up to 5 builds a month.
+- It is stopped by default (so local and branch deployments never rebuild prod). Turn it
+  on in the prod deployment.
 
 ### T6 — Pipeline correctness fixes
 

orchestration/AGENTS.md

@@ -55,7 +55,8 @@ uv run python -c "import orchestration.definitions; print('ok')"
   `assets/products.py:build_product_assets` — into a per-source asset graph:
   `sources/<key>` → combine (`<product_id>`) → `<product_id>/geoserver`.
 - `definitions.py` wires assets, one asset job per product (`<product_id>_job`),
-  schedules, and resources (`die_config`, `gcs`, `geoserver`, `io_manager`).
+  schedules, the `pygeoapi_rebuild` sensor (`pygeoapi_rebuild.py`), and resources
+  (`die_config`, `gcs`, `geoserver`, `io_manager`).
 - Resources live in `resources/`. The backend DIE engine is the `nmuwd` package
   (repo root `backend/`); orchestration only drives it.
 - Per-source and geoserver assets soft-fail: errors surface as red asset checks
@@ -67,7 +68,8 @@ Dagster+ serverless builds from the repo root per `dagster_cloud.yaml`
 (`module_name: orchestration.definitions`). Runtime deps come from the root
 `requirements.txt`. Secrets are set as Dagster+ env vars, not in code:
 
-- `GCP_SERVICE_ACCOUNT_KEY` — GCS upload/IO-manager auth (JSON key).
+- `GCP_SERVICE_ACCOUNT_KEY` — JSON key for GCS upload/IO-manager and the
+  `pygeoapi_rebuild` Cloud Build trigger.
 - `GEOSERVER_URL` / `GEOSERVER_USER` / `GEOSERVER_PASSWORD` / `GEOSERVER_WORKSPACE`.
 - `USGS_API_KEY` — USGS/NWIS API key. Without it the USGS water data API is
   heavily rate-limited. Resolved via `dg.EnvVar` into `DIEConfigResource`, which

orchestration/definitions.py

@@ -5,6 +5,7 @@
 import dagster as dg
 import yaml
 
+from orchestration.pygeoapi_rebuild import build_rebuild_sensor
 from orchestration.resources.die_config import DIEConfigResource
 from orchestration.resources.gcs import GCSResource, AuthedGCSResource
 from orchestration.resources.geoserver import GeoServerResource
@@ -244,6 +245,7 @@ def _build_standalone_jobs_and_schedules(
     assets=_assets,
     jobs=list(_cohort_jobs.values()),
     schedules=_schedules,
+    sensors=[build_rebuild_sensor(list(_cohort_jobs))],
     resources={
         # USGS_API_KEY is a Dagster+ secret; EnvVar resolves it at run time and
         # the resource exports it for the NWIS connector. Resolves to None when

orchestration/pygeoapi_rebuild.py

@@ -0,0 +1,77 @@
+"""Rebuild pygeoapi once product runs finish, so the image bakes the new data (SPEC §6.6)."""
+import json
+import os
+from datetime import datetime, timezone
+
+import dagster as dg
+
+TRIGGER = "die-pygeoapi-rebuild"
+REGION = "us-central1"
+
+_IN_PROGRESS = [
+    dg.DagsterRunStatus.QUEUED,
+    dg.DagsterRunStatus.NOT_STARTED,
+    dg.DagsterRunStatus.STARTING,
+    dg.DagsterRunStatus.STARTED,
+]
+
+
+def run_trigger() -> None:
+    """Start the Cloud Build trigger, using the same credentials as GCS."""
+    import google.auth
+    from google.auth.transport.requests import AuthorizedSession
+    from google.oauth2 import service_account
+
+    scopes = ["https://www.googleapis.com/auth/cloud-platform"]
+    key = os.environ.get("GCP_SERVICE_ACCOUNT_KEY")
+    if key:
+        info = json.loads(key)
+        creds = service_account.Credentials.from_service_account_info(info, scopes=scopes)
+        project = info["project_id"]
+    else:
+        creds, project = google.auth.default(scopes=scopes)
+    url = (
+        f"https://cloudbuild.googleapis.com/v1/projects/{project}"
+        f"/locations/{REGION}/triggers/{TRIGGER}:run"
+    )
+    AuthorizedSession(creds).post(url, json={}).raise_for_status()
+
+
+def build_rebuild_sensor(job_names: list[str]) -> dg.SensorDefinition:
+    """Sensor that starts one rebuild after a product job succeeds, once no
+    product job is queued or running. The cursor is the last check time."""
+    jobs = set(job_names)
+
+    # Stopped by default so local and branch deployments never rebuild prod;
+    # turn it on in the prod deployment.
+    @dg.sensor(
+        name="pygeoapi_rebuild",
+        minimum_interval_seconds=600,
+        default_status=dg.DefaultSensorStatus.STOPPED,
+    )
+    def pygeoapi_rebuild(context: dg.SensorEvaluationContext):
+        now = datetime.now(timezone.utc)
+        if context.cursor is None:
+            context.update_cursor(now.isoformat())
+            return dg.SkipReason("First tick; watching for product runs from now.")
+
+        def product_runs(**filters):
+            runs = context.instance.get_runs(filters=dg.RunsFilter(**filters))
+            return [r for r in runs if r.job_name in jobs]
+
+        if product_runs(statuses=_IN_PROGRESS):
+            return dg.SkipReason("Product runs still in progress.")
+        succeeded = product_runs(
+            statuses=[dg.DagsterRunStatus.SUCCESS],
+            updated_after=datetime.fromisoformat(context.cursor),
+        )
+        if not succeeded:
+            context.update_cursor(now.isoformat())
+            return dg.SkipReason("No product runs succeeded since the last check.")
+
+        run_trigger()
+        context.update_cursor(now.isoformat())
+        context.log.info(f"Started {TRIGGER} after {len(succeeded)} product run(s).")
+        return dg.SkipReason(f"Started Cloud Build trigger {TRIGGER}.")
+
+    return pygeoapi_rebuild

tests/test_pygeoapi_rebuild.py

@@ -0,0 +1,65 @@
+from datetime import datetime, timedelta, timezone
+from unittest import mock
+
+import dagster as dg
+import pytest
+
+from orchestration import pygeoapi_rebuild
+from orchestration.pygeoapi_rebuild import build_rebuild_sensor
+
+SUCCESS = dg.DagsterRunStatus.SUCCESS
+STARTED = dg.DagsterRunStatus.STARTED
+
+
+@pytest.fixture
+def trigger():
+    with mock.patch.object(pygeoapi_rebuild, "run_trigger") as run:
+        yield run
+
+
+def _evaluate(runs, cursor):
+    """Run the sensor against fake (job_name, status) runs."""
+    def get_runs(filters):
+        return [
+            mock.Mock(job_name=job)
+            for job, status in runs
+            if status in filters.statuses
+        ]
+
+    instance = mock.Mock(spec=dg.DagsterInstance)
+    instance.get_runs.side_effect = get_runs
+    context = dg.build_sensor_context(instance=instance, cursor=cursor)
+    result = build_rebuild_sensor(["wl_job", "chem_job"])(context)
+    return result, context.cursor
+
+
+def _earlier():
+    return (datetime.now(timezone.utc) - timedelta(hours=1)).isoformat()
+
+
+def test_first_tick_sets_cursor(trigger):
+    _, cursor = _evaluate([("wl_job", SUCCESS)], cursor=None)
+    assert cursor is not None
+    trigger.assert_not_called()
+
+
+def test_rebuilds_once_product_runs_finish(trigger):
+    _evaluate([("wl_job", SUCCESS), ("chem_job", SUCCESS)], cursor=_earlier())
+    trigger.assert_called_once()
+
+
+def test_waits_while_a_product_run_is_in_progress(trigger):
+    earlier = _earlier()
+    _, cursor = _evaluate([("wl_job", SUCCESS), ("chem_job", STARTED)], cursor=earlier)
+    trigger.assert_not_called()
+    assert cursor == earlier  # keep the success for the next tick
+
+
+def test_ignores_other_jobs(trigger):
+    _evaluate([("other_job", SUCCESS)], cursor=_earlier())
+    trigger.assert_not_called()
+
+
+def test_no_rebuild_without_new_success(trigger):
+    _evaluate([], cursor=_earlier())
+    trigger.assert_not_called()