feat: access runs by (proposal, run) identity and render the table from a normalized cache

api/README.md

@@ -43,13 +43,23 @@ podman compose up
 
 ```gql
 query TableMetadataQuery {
-  metadata(database: {proposal: "<PROPOSAL_NUMBER>"})
+  metadata(database: {proposal: "<PROPOSAL_NUMBER>"}) {
+    runs {
+      proposal
+      run
+    }
+    variables
+    tags
+    timestamp
+  }
 }
 ```
 
-This returns a JSON snapshot for the proposal:
+This returns a `TableMeta` snapshot for the proposal:
 
-- `runs` - sorted list of run numbers in the proposal
+- `runs` - server-ordered list of `{proposal, run}` pairs. Run numbers repeat
+  across the proposals sharing one database, so a run is only identified by
+  the pair
 - `variables` - map of variable name to its title and tags (includes the
   known variables listed below alongside any user-defined ones)
 - `tags` - map of tag name to its id and the variables it groups
@@ -63,6 +73,9 @@ of their cells:
 ```gql
 query TableDataQuery($per_page: Int = 10) {
   runs(database: {proposal: "2956"}, per_page: $per_page) {
+    database
+    proposal
+    run
     cells {
       name
       value
@@ -72,8 +85,12 @@ query TableDataQuery($per_page: Int = 10) {
 }
 ```
 
-Each run is returned as a flat list of `Cell` entries (`name`,
-`value`, `dtype`). One can pass a list of `names` to select variables:
+Every run carries the identity trio `database`, `proposal` and `run`. Select
+all three in every document: it is what the client normalizes each run by, and
+what keeps two proposals' run 5 apart.
+
+The cells are a flat list of `Cell` entries (`name`, `value`, `dtype`). One can
+pass a list of `names` to select variables:
 
 ```gql
 query TableDataQuery($per_page: Int = 10) {
@@ -98,17 +115,43 @@ ones from the proposal's context file:
 Pagination is controlled with `page` (1-indexed, defaults to `1`) and
 `per_page` (defaults to `10`).
 
-### 3. Subscribe to latest data
+### 3. Subscribe to run updates
 
 ```gql
-subscription LatestDataSubscription {
-  latest_data(database: {proposal: "<PROPOSAL_NUMBER>"}, timestamp: <TIMESTAMP>)
+subscription RunUpdatesSubscription {
+  run_updates(database: {proposal: "<PROPOSAL_NUMBER>"}, since: <TIMESTAMP>) {
+    runs {
+      database
+      proposal
+      run
+      cells {
+        name
+        value
+        dtype
+      }
+    }
+    metadata {
+      runs {
+        proposal
+        run
+      }
+      variables
+      tags
+      timestamp
+    }
+    timestamp
+  }
 }
 ```
 
-This returns the following:
+Each push is a `RunUpdates`:
 
-- list of (new) runs
-- updated metadata
+- `runs` - the runs whose cells changed since `since`
+- `metadata` - the full `TableMeta`, but only on a push where it materially
+  changed. It is `null` otherwise, so a tag edit or a new variable arrives
+  even on a tick that brings no runs
+- `timestamp` - the cursor to send as the next `since`
 
-Note that the `timestamp` is in milliseconds since Unix epoch.
+Note that both `since` and `timestamp` are in milliseconds since the Unix
+epoch. Passing `since: 0` is the cold start: it delivers the next tick's
+changed runs, not the proposal's history.

api/src/damnit_api/graphql/metadata.py

@@ -1,4 +1,6 @@
 import asyncio
+import hashlib
+import json
 
 from async_lru import alru_cache
 
@@ -11,15 +13,16 @@
 async def fetch_metadata(proposal=db.DEFAULT_PROPOSAL):
     """Fetch the per-proposal metadata snapshot from SQLite.
 
-    Returns a dict with `runs`, `variables`, `tags`, and `timestamp`. Result
-    is TTL-cached; the `latest_data` subscription invalidates this cache when
-    it observes new data so subsequent reads stay fresh.
+    Returns a dict with `runs`, `variables`, `tags`, and `timestamp`. `runs`
+    is a server-ordered list of (proposal, run) pairs (active block first).
+    Result is TTL-cached; the `run_updates` subscription invalidates this
+    cache when it observes new data so subsequent reads stay fresh.
     """
     tags, variables, variable_tags, runs, max_timestamp = await asyncio.gather(
         db.async_all_tags(proposal),
         db.async_variables(proposal),
         db.async_variable_tags(proposal),
-        db.async_column(proposal, table="run_info", name="run"),
+        db.async_run_identifiers(proposal),
         db.async_max(proposal, table="run_variables", column="timestamp"),
     )
 
@@ -39,9 +42,26 @@ async def fetch_metadata(proposal=db.DEFAULT_PROPOSAL):
     }
     tags = create_map([untagged, *tags.values()], key="name")
 
-    return {
-        "runs": sorted(runs or []),
+    snapshot = {
+        "runs": runs,
         "variables": variables,
         "tags": tags,
         "timestamp": max_timestamp or 0,
     }
+    snapshot["signature"] = _signature(snapshot)
+    return snapshot
+
+
+def _signature(snapshot) -> str:
+    """Hash of everything a subscriber would be pushed.
+
+    Computed here so it costs one pass per actual read rather than one per
+    subscription tick. `timestamp` is left out: it moves whenever any value
+    changes, which would make every tick look like a metadata change.
+    """
+    payload = json.dumps(
+        {key: snapshot[key] for key in ("runs", "variables", "tags")},
+        sort_keys=True,
+        default=str,
+    )
+    return hashlib.sha256(payload.encode()).hexdigest()

api/src/damnit_api/graphql/queries.py

@@ -1,5 +1,5 @@
 import strawberry
-from sqlalchemy import and_, func, select
+from sqlalchemy import and_, func, select, tuple_
 from strawberry.scalars import JSON
 from strawberry.types import Info
 from strawberry.types.nodes import SelectedField
@@ -8,8 +8,13 @@
 from ..auth.permissions import PROPOSAL_PERMISSIONS
 from ..metadata.services import _get_proposal_meta, _update_proposal_meta
 from ..runs.preview import get_preview_data
-from ..runs.sqlite import async_table, get_session
-from ..runs.types import KNOWN_DTYPES, DamnitRun
+from ..runs.sqlite import (
+    async_active_proposal,
+    async_table,
+    get_session,
+    order_by_active,
+)
+from ..runs.types import KNOWN_DTYPES, DamnitRun, TableMeta
 from .metadata import fetch_metadata
 from .utils import DatabaseInput, fetch_info
 
@@ -82,27 +87,32 @@ async def fetch_cells(proposal, *, limit, offset, names=None):
     if table is None:
         return []
 
+    active = await async_active_proposal(proposal)
     runs_subquery = (
         select(table.c.proposal, table.c.run)
         .distinct()
-        .order_by(table.c.run)
+        .order_by(*order_by_active(table, active))
         .limit(limit)
         .offset(offset)
         .subquery()
     )
 
+    page_pairs = select(runs_subquery.c.proposal, runs_subquery.c.run)
     latest_timestamp_subquery = select(
         table.c.proposal,
         table.c.run,
         table.c.name,
         func.max(table.c.timestamp).label("latest_timestamp"),
-    ).where(table.c.run.in_(select(runs_subquery.c.run)))
+    ).where(tuple_(table.c.proposal, table.c.run).in_(page_pairs))
     if names is not None:
         latest_timestamp_subquery = latest_timestamp_subquery.where(
             table.c.name.in_(names)
         )
+    # Group by (proposal, run, name): grouping by run alone would let one
+    # run's max timestamp mix in a colliding run from another proposal, and
+    # SQLite's bare-column group-by would then pick an arbitrary proposal.
     latest_timestamp_subquery = latest_timestamp_subquery.group_by(
-        table.c.run, table.c.name
+        table.c.proposal, table.c.run, table.c.name
     ).subquery()
 
     # Outer-join from `runs_subquery` so a `names` filter that excludes every
@@ -119,7 +129,10 @@ async def fetch_cells(proposal, *, limit, offset, names=None):
         .select_from(runs_subquery)
         .outerjoin(
             latest_timestamp_subquery,
-            runs_subquery.c.run == latest_timestamp_subquery.c.run,
+            and_(
+                runs_subquery.c.proposal == latest_timestamp_subquery.c.proposal,
+                runs_subquery.c.run == latest_timestamp_subquery.c.run,
+            ),
         )
         .outerjoin(
             table,
@@ -130,20 +143,22 @@ async def fetch_cells(proposal, *, limit, offset, names=None):
                 table.c.timestamp == latest_timestamp_subquery.c.latest_timestamp,
             ),
         )
-        .order_by(runs_subquery.c.run)
+        .order_by(*order_by_active(runs_subquery, active))
     )
 
     async with get_session(proposal) as session:
         result = await session.execute(query)
-        if not result:
-            raise ValueError  # TODO: Better error handling
-
         return group_by_run(result.mappings().all())  # type: ignore[assignment]
 
 
 def _selected_cell_names(info: Info) -> list[str] | None:
     """Union the `names` arguments across every `cells` sub-selection.
-    Returns None if any selection omits the argument (forces a full fetch).
+
+    Returns one of three things:
+    - ``[]`` if no `cells` field is selected (an identity-only query): fetch no
+      cells and skip the run_info fetch, since there is nothing to serialize.
+    - ``None`` if a `cells` selection omits `names`: fetch every cell.
+    - the sorted union of the requested names otherwise.
     """
     union = set()
     found = False
@@ -156,7 +171,10 @@ def _selected_cell_names(info: Info) -> list[str] | None:
             if arg is None:
                 return None
             union.update(arg)
-    return sorted(union) if found else None
+    if not found:
+        # No `cells` selected: caller wants run identities only.
+        return []
+    return sorted(union)
 
 
 def _wants_run_info(names: list[str] | None) -> bool:
@@ -199,23 +217,25 @@ async def runs(
         if not len(cells):
             return []
 
-        if _wants_run_info(names):
-            info_rows = await fetch_info(
-                proposal, runs=[c["run"]["value"] for c in cells]
-            )
-        else:
-            info_rows = [{} for _ in cells]
+        pairs = [(c["proposal"]["value"], c["run"]["value"]) for c in cells]
+        run_info = (
+            await fetch_info(proposal, runs=pairs) if _wants_run_info(names) else {}
+        )
 
         return [
-            DamnitRun.from_db({**c, **i}) for c, i in zip(cells, info_rows, strict=True)
+            DamnitRun.from_db(
+                {**c, **run_info.get(pair, {})},
+                database=database.proposal,
+            )
+            for c, pair in zip(cells, pairs, strict=True)
         ]
 
     @strawberry.field(permission_classes=PROPOSAL_PERMISSIONS)
     async def metadata(
         self,
         info: Info,
         database: DatabaseInput,
-    ) -> JSON:  # FIX: # pyright: ignore[reportInvalidTypeForm]
+    ) -> TableMeta:
         proposal = database.proposal
         if not proposal:
             msg = "Proposal number is required."
@@ -225,10 +245,7 @@ async def metadata(
         await _ensure_damnit_path(info, proposal)
 
         snapshot = await fetch_metadata(proposal)
-        return {
-            **snapshot,
-            "timestamp": snapshot["timestamp"] * 1000,  # ms for JS
-        }  # pyright: ignore[reportReturnType]
+        return TableMeta.from_snapshot(snapshot)
 
     # Nullable, because a preview asks for many runs in one request, aliasing
     # this field once per run. A non-null field that raises propagates the null