feat(api): build/rebuild the resource-state read model tables

.github/workflows/python-test.yml

@@ -5,6 +5,26 @@ jobs:
   test:
     name: "tests (pytest)"
     runs-on: ubuntu-latest
+    # Throwaway Postgres for the API tests SQLite can't exercise (row locks under
+    # concurrency, JSONB): see deployments/api/tests/db/read_model/postgres.py.
+    # Each test creates and drops its own schema.
+    services:
+      postgres:
+        image: postgres:17-alpine
+        env:
+          POSTGRES_PASSWORD: postgres
+          POSTGRES_DB: stitch_test
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd "pg_isready -U postgres -d stitch_test"
+          --health-interval 5s
+          --health-timeout 5s
+          --health-retries 10
+    env:
+      STITCH_TEST_POSTGRES_URL: postgresql+psycopg://postgres:postgres@localhost:5432/stitch_test
+      # Fail rather than skip if the URL above ever goes missing.
+      STITCH_REQUIRE_POSTGRES_TESTS: "1"
     steps:
       - name: Checkout code
         uses: actions/checkout@v7

deployments/api/READ_MODEL.md

@@ -9,8 +9,8 @@ The table is a cache, not a source of truth. Everything in it can be rebuilt fro
 the authoritative tables at any time, and when it can't be trusted the API simply
 answers the slower way ("live"), with the same results.
 
-So far only the tables exist: they start empty and not ready, and nothing writes
-or reads them yet, so answers are unchanged.
+So far the tables and the rebuild command exist. Nothing reads the table yet, and
+API writes don't keep it current yet, so answers are unchanged.
 
 ## How it works
 
@@ -26,10 +26,104 @@ or reads them yet, so answers are unchanged.
   may be used only when it is ready **and** was built by the same commit the API
   is running.
 
+## Running a rebuild
+
+```bash
+# Full rebuild (all resources). Safe while the API is serving traffic.
+python -m stitch.api.db.read_model.rebuild
+
+# Full rebuild only if it isn't already built by this code (the post-deploy step).
+python -m stitch.api.db.read_model.rebuild --if-stale
+
+# Recompute specific resources only (e.g. after fixing data by hand).
+python -m stitch.api.db.read_model.rebuild --ids 123 456
+```
+
+- **Where to run it:** with the *same code* the API is running, because the
+  rebuild records its commit and the read model only counts as ready for the
+  commit that built it. For a deployed lane, run the deployed `api` image (it
+  carries the commit as `GIT_SHA`) the same way CI runs migrations, with the
+  lane's `POSTGRES_*` variables, as `stitch_app` or `stitch_migrator`. A rebuild
+  run from a laptop records `unknown`, which matches no deployed commit.
+
+  ```bash
+  docker run --rm -e POSTGRES_HOST -e POSTGRES_PORT -e POSTGRES_DB -e POSTGRES_USER -e POSTGRES_PASSWORD -e PGSSLMODE <api-image> python -m stitch.api.db.read_model.rebuild
+  ```
+
+  Locally, run it inside the `api` container if the API runs in Docker, or on
+  your machine if the API does:
+
+  ```bash
+  POSTGRES_HOST=127.0.0.1 POSTGRES_USER=stitch_app uv run --env-file .env -- python -m stitch.api.db.read_model.rebuild
+  ```
+
+- **What happens:** readiness is cleared first, so the table isn't used while
+  it's rebuilt. Resources are rebuilt in committed batches (`--batch-size`,
+  default 500), each batch replacing its resources' rows. Then it checks for
+  resources whose rows were since rewritten by other code (an older API
+  revision still serving writes, or an `--ids` repair run from another image)
+  and redoes those, repeating until none are left. The last of these checks
+  happens in the same transaction that sets readiness, so no repair can slip in
+  between: rows for resources that are no longer listable are removed, the
+  result is checked (every listable resource has all four rows, all computed by
+  this code, and nothing else has any), and only then is readiness set. The
+  table is never emptied along the way, so a reader that checked readiness just
+  before the rebuild started still finds a row for every resource. At ~50K
+  resources it takes about 1.5 minutes.
+- **Full rebuild vs `--if-stale`:** a plain full rebuild recomputes every
+  resource. `--if-stale` only recomputes resources not already computed by this
+  code, so re-running it after a failed or cancelled run picks up where that run
+  stopped.
+- **If it fails:** readiness stays cleared; just run it again. Whatever rows a
+  failed run leaves are never used, and the next run replaces or reuses them. If
+  other code keeps rewriting rows, it stops after 5 passes with an error asking
+  whether an older API version is still serving writes.
+- **Two at once:** a full rebuild started by hand while another is running
+  exits with "Another resource-state rebuild is already running." `--if-stale`
+  (the deploy step) instead waits up to 10 minutes for it to finish, then checks
+  again and does nothing if that rebuild already made the read model current.
+- **A rebuild that dies mid-run** (e.g. a cancelled CI job) releases its lock
+  within about a minute: the lock's connection asks Postgres to drop vanished
+  clients quickly. If a lock still seems stuck, find the session holding it and,
+  once you're sure no rebuild is running, end it:
+
+  ```sql
+  SELECT l.pid, a.state, a.backend_start, a.client_addr
+  FROM pg_locks l JOIN pg_stat_activity a USING (pid)
+  WHERE l.locktype = 'advisory' AND l.objid = 766001 AND l.granted;
+
+  SELECT pg_terminate_backend(<pid>);
+  ```
+- `--ids` never changes readiness, and is safe while the API serves reads. It
+  reports which ids it recomputed and which aren't listed (merged away, or no
+  active sources — any stored rows for those are removed). It changes nothing
+  and exits with an error if any id doesn't exist (usually a typo), or if the
+  table is ready but was built by different code: run it from the same image
+  as the API, so the table never mixes two versions. During a full rebuild the
+  table isn't ready, so a repair from any image runs; the rebuild waits for it
+  and redoes its rows before finishing. A repair that starts just as a rebuild
+  finishes waits for it, then refuses if that rebuild was by other code.
+- Expected failures (another rebuild running, a failed integrity check, rows
+  still rewritten by other code after 5 passes, unknown ids, a table built by
+  other code) print as a single `ERROR` line; unexpected ones (e.g. the
+  database is unreachable) show a traceback. Either way it exits with status 1.
+  Invalid arguments print the usage and exit with status 2.
+
 ## Changing the code safely
 
 - **New field** on `OilGasFieldBase`: add the column to `OGFieldResourceState`
-  (an import-time check fails until you do) plus a migration.
+  (an import-time check fails until you do) plus a migration, then rebuild.
 - **New source key:** decide whether it's public or restricted in
   `db/read_model/permissions.py`. A test fails until you do; until then, anyone
   holding the new key would get the live path.
+
+## Tests
+
+The normal suite runs on SQLite. A few tests need real Postgres (e.g. the rebuild
+lock). CI runs them on every PR against a throwaway Postgres; locally they're
+skipped unless pointed at a disposable database. They create and drop their own
+schema:
+
+```bash
+STITCH_TEST_POSTGRES_URL=postgresql+psycopg://postgres:pg@127.0.0.1:55432/stitch uv run --package stitch-api pytest deployments/api/tests/db/read_model -q
+```

deployments/api/src/stitch/api/db/queries.py

@@ -388,11 +388,14 @@ def coalesced_candidate_rows(
     )
 
 
-def _resource_universe() -> Select[tuple[int]]:
+def resource_universe() -> Select[tuple[int]]:
     """Resources eligible to appear in a list: any non-repointed resource with an
     active membership. Membership-derived and ungated by licensing/source, so a
     resource whose licensed values are all absent still appears as a null-shell on
-    an unfiltered list (and drops out once a field is filtered)."""
+    an unfiltered list (and drops out once a field is filtered).
+
+    Also decides which resources get rows in the resource-state read model
+    (``read_model.state``), so cached and live listability can't diverge."""
     m = MembershipModel
     r = ResourceModel
     return (
@@ -409,7 +412,7 @@ def base_resource_query(
     licensed_sources: Collection[OGSISrcKey] | None = None,
 ) -> Select[tuple[int]]:
     involved = _participating_columns(params)
-    universe = _resource_universe().cte("resource_universe")
+    universe = resource_universe().cte("resource_universe")
 
     if not involved:
         # No value field filtered or sorted -> the universe alone (every active

deployments/api/src/stitch/api/db/read_model/rebuild.py

@@ -0,0 +1,185 @@
+"""Rebuild or repair the resource-state read model.
+
+    python -m stitch.api.db.read_model.rebuild             # full rebuild
+    python -m stitch.api.db.read_model.rebuild --if-stale  # only if not current
+    python -m stitch.api.db.read_model.rebuild --ids 12 34 # repair these resources
+
+The read model counts as ready only once a full rebuild by the same commit has
+completed, so a full rebuild is needed after every deploy of new code (and after
+a data restore). ``--if-stale`` is the deploy step: it rebuilds only when the
+read model isn't already current for this code. ``--ids`` recomputes only the
+given resources and never changes readiness.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import logging
+from collections.abc import Sequence
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from ..config import dispose_engine, get_engine, get_session_factory
+from ..model import OGFieldResourceStateStatus, ResourceModel
+from ..model.og_field_resource_state import RESOURCE_STATE_STATUS_KEY
+from .state import (
+    DEFAULT_BATCH_SIZE,
+    ResourceStateRebuildError,
+    current_code_version,
+    rebuild_all_resource_state,
+    rebuild_resource_state_if_stale,
+    refresh_resource_states,
+)
+
+logger = logging.getLogger(__name__)
+
+
+def _positive_int(raw: str) -> int:
+    value = int(raw)
+    if value < 1:
+        raise argparse.ArgumentTypeError(f"must be at least 1, got {value}")
+    return value
+
+
+def _parse_args(argv: Sequence[str] | None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Rebuild the resource-state read model, or repair specific "
+        "resources."
+    )
+    scope = parser.add_mutually_exclusive_group()
+    scope.add_argument(
+        "--ids",
+        nargs="+",
+        type=int,
+        metavar="RESOURCE_ID",
+        help="Only recompute these resource ids (safe while serving reads).",
+    )
+    scope.add_argument(
+        "--if-stale",
+        action="store_true",
+        help="Do a full rebuild only if the read model isn't already built by "
+        "this code (the post-deploy step). Waits for a rebuild already running, "
+        "then checks again.",
+    )
+    parser.add_argument(
+        "--batch-size",
+        type=_positive_int,
+        default=DEFAULT_BATCH_SIZE,
+        help=f"Resources per committed batch in a full rebuild "
+        f"(default: {DEFAULT_BATCH_SIZE}).",
+    )
+    return parser.parse_args(argv)
+
+
+async def _refuse_if_served_by_other_code(session: AsyncSession) -> None:
+    """Refuse to change a table that is being served as built by other code.
+
+    The API trusts a ready table only as built entirely by its own commit, so
+    rows recomputed here by different code would be served under the wrong
+    version. A table that isn't ready isn't served, and the next full rebuild
+    replaces its rows. Holds a share lock on the status row until this
+    transaction ends, so readiness can't change meanwhile: a full rebuild
+    finishing now waits for the repair to commit, then sees its rows and redoes
+    them before marking the table ready.
+    """
+    status = (
+        await session.execute(
+            select(
+                OGFieldResourceStateStatus.ready,
+                OGFieldResourceStateStatus.code_version,
+            )
+            .where(OGFieldResourceStateStatus.key == RESOURCE_STATE_STATUS_KEY)
+            .with_for_update(read=True)
+        )
+    ).one_or_none()
+    if (
+        status is not None
+        and status.ready
+        and status.code_version != current_code_version()
+    ):
+        raise ResourceStateRebuildError(
+            f"The read model is being served as built by code "
+            f"{status.code_version}, but this is code {current_code_version()}. "
+            "Run the repair from that code (or a full rebuild from this code); "
+            "nothing was changed."
+        )
+
+
+async def _repair(
+    session_factory: async_sessionmaker[AsyncSession], ids: list[int]
+) -> None:
+    """Recompute the given resources, refusing unknown ids (likely typos) and
+    tables being served as built by other code."""
+    requested = sorted(set(ids))
+    async with session_factory.begin() as session:
+        await _refuse_if_served_by_other_code(session)
+        existing = set(
+            (
+                await session.scalars(
+                    select(ResourceModel.id).where(ResourceModel.id.in_(requested))
+                )
+            ).all()
+        )
+        if missing := [rid for rid in requested if rid not in existing]:
+            raise ResourceStateRebuildError(
+                f"No resource with id(s) {missing}; nothing was changed."
+            )
+        recomputed = await refresh_resource_states(session, requested)
+    if recomputed:
+        logger.info("Recomputed resource state for ids %s.", recomputed)
+    if unlisted := [rid for rid in requested if rid not in recomputed]:
+        logger.info(
+            "Ids %s are not listed (merged away, or no active sources), so they "
+            "have no stored state; any they had was removed.",
+            unlisted,
+        )
+
+
+async def _run(ids: list[int] | None, batch_size: int, if_stale: bool = False) -> None:
+    # Outside prod the app engine echoes every statement and its parameters; for
+    # a batch job over every resource that floods the logs. Progress is logged
+    # per batch instead.
+    get_engine().echo = False
+    session_factory = get_session_factory()
+    try:
+        if ids:
+            await _repair(session_factory, ids)
+            return
+        if if_stale:
+            count = await rebuild_resource_state_if_stale(
+                session_factory, batch_size=batch_size
+            )
+        else:
+            count = await rebuild_all_resource_state(
+                session_factory, batch_size=batch_size
+            )
+        if count is None:
+            logger.info(
+                "The read model is already built by this code (%s); nothing to do.",
+                current_code_version(),
+            )
+        else:
+            logger.info(
+                "Rebuilt resource state for %d resources with code %s; the read "
+                "model is ready.",
+                count,
+                current_code_version(),
+            )
+    finally:
+        await dispose_engine()
+
+
+def main(argv: Sequence[str] | None = None) -> None:
+    args = _parse_args(argv)
+    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
+    try:
+        asyncio.run(_run(args.ids, args.batch_size, args.if_stale))
+    except ResourceStateRebuildError as exc:
+        logger.error("%s", exc)
+        raise SystemExit(1) from exc
+
+
+if __name__ == "__main__":
+    main()