perf: the /data read surface in milliseconds (#219)

.gitignore

@@ -245,3 +245,6 @@ server/osa.yaml
 
 # MCP widget bundles staged into the server package (built from packages/osa-widgets)
 server/osa/application/api/mcp/bundles/
+
+# graphify knowledge-graph outputs (generated caches, any directory)
+graphify-out/

server/migrations/versions/a8c4e6f19b02_table_statistics.py

@@ -0,0 +1,79 @@
+"""``table_statistics`` — lockstep counts, backfilled once from the data.
+
+Creates the per-(schema version, table) count state and populates it from
+COUNT(*) / COUNT(DISTINCT record_srn) group-bys over the existing tables.
+This backfill and the admin verifier are the only sanctioned whole-table
+counting after #219; from here on the writing adapters maintain the counts
+transactionally (osa/infrastructure/persistence/statistics_upsert.py).
+
+Feature-table names come from the ``feature_tables`` catalog and are
+re-validated against the strict identifier pattern before interpolation —
+never string-built from user input.
+
+Revision ID: a8c4e6f19b02
+Revises: f3a1c9d27e54
+Create Date: 2026-08-15
+"""
+
+import re
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "a8c4e6f19b02"
+down_revision = "f3a1c9d27e54"
+branch_labels = None
+depends_on = None
+
+_SAFE_IDENT = re.compile(r"^[a-z][a-z0-9_]{0,62}$")
+
+
+def upgrade() -> None:
+    op.create_table(
+        "table_statistics",
+        sa.Column("schema_id", sa.Text(), primary_key=True),
+        sa.Column("schema_version", sa.Text(), primary_key=True),
+        sa.Column("table_name", sa.Text(), primary_key=True),
+        sa.Column("row_count", sa.BigInteger(), nullable=False, server_default="0"),
+        sa.Column("records_covered", sa.BigInteger(), nullable=True),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+    )
+
+    conn = op.get_bind()
+    # Records: one row per schema version present in the data.
+    conn.execute(
+        sa.text(
+            """
+            INSERT INTO table_statistics
+                (schema_id, schema_version, table_name, row_count, records_covered, updated_at)
+            SELECT schema_id, schema_version, 'records', count(*), NULL, now()
+            FROM records
+            GROUP BY schema_id, schema_version
+            """
+        )
+    )
+    # Features: every registered feature table, scoped per schema via the records join.
+    tables = conn.execute(sa.text("SELECT hook_name, pg_table FROM feature_tables")).fetchall()
+    for hook_name, pg_table in tables:
+        if not _SAFE_IDENT.match(pg_table):
+            continue
+        conn.execute(
+            sa.text(
+                f"""
+                INSERT INTO table_statistics
+                    (schema_id, schema_version, table_name, row_count,
+                     records_covered, updated_at)
+                SELECT r.schema_id, r.schema_version, :hook, count(ft.id),
+                       count(DISTINCT ft.record_srn), now()
+                FROM features."{pg_table}" ft
+                JOIN records r ON r.srn = ft.record_srn
+                GROUP BY r.schema_id, r.schema_version
+                """
+            ),
+            {"hook": hook_name},
+        )
+
+
+def downgrade() -> None:
+    op.drop_table("table_statistics")

server/migrations/versions/f3a1c9d27e54_records_read_index.py

@@ -0,0 +1,43 @@
+"""Composite read index on records; drop the subsumed schema_id index.
+
+``records (schema_id, schema_version, published_at, srn)`` serves the default
+table read — schema equality prefix + (published_at, srn) ordering — as one
+(backward) index range scan, including the row-value keyset predicate (#219
+phase 3). ``idx_records_schema_id`` is its left prefix and is dropped;
+``idx_records_published_at`` stays (used by count_this_month's month filter).
+
+Built ``CONCURRENTLY``: live archives run this migration on a records table
+serving traffic, so the build must not take a table lock. CONCURRENTLY cannot
+run inside a transaction block, hence the autocommit block.
+
+Revision ID: f3a1c9d27e54
+Revises: b47f9c2e8a31
+Create Date: 2026-08-15
+"""
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "f3a1c9d27e54"
+down_revision = "b47f9c2e8a31"
+branch_labels = None
+depends_on = None
+
+_INDEX = "idx_records_schema_version_published"
+
+
+def upgrade() -> None:
+    with op.get_context().autocommit_block():
+        op.create_index(
+            _INDEX,
+            "records",
+            ["schema_id", "schema_version", "published_at", "srn"],
+            postgresql_concurrently=True,
+        )
+    op.drop_index("idx_records_schema_id", table_name="records")
+
+
+def downgrade() -> None:
+    op.create_index("idx_records_schema_id", "records", ["schema_id"])
+    with op.get_context().autocommit_block():
+        op.drop_index(_INDEX, table_name="records", postgresql_concurrently=True)

server/osa/application/api/v1/routes/data/features_table.py

@@ -16,7 +16,11 @@
 from osa.application.api.v1.routes.data._streaming import build_table_response
 from osa.application.api.v1.routes.data.formats import DataResponseFormat
 from osa.application.api.v1.routes.data.tables import format_key, register_table_routes
-from osa.domain.data.query.read_table import ReadFeatureTable, ReadFeatureTableHandler
+from osa.domain.data.query.read_table import (
+    ReadFeatureTable,
+    ReadFeatureTableHandler,
+    ReadMode,
+)
 from osa.domain.shared.model.ids import FeatureName
 
 
@@ -36,6 +40,7 @@ async def endpoint(
                 cursor=cursor,
                 limit=limit,
                 sort=parse_sort(sort),
+                mode=ReadMode.PAGE if fmt.paginated else ReadMode.STREAM,
                 timeout=fmt.timeout,
             )
         )
@@ -60,6 +65,7 @@ async def endpoint(
                 cursor=body.cursor,
                 limit=body.limit,
                 sort=parse_sort(body.sort),
+                mode=ReadMode.PAGE if fmt.paginated else ReadMode.STREAM,
                 timeout=fmt.timeout,
             )
         )

server/osa/application/api/v1/routes/data/records_table.py

@@ -19,7 +19,11 @@
 from osa.application.api.v1.routes.data._streaming import build_table_response
 from osa.application.api.v1.routes.data.formats import DataResponseFormat
 from osa.application.api.v1.routes.data.tables import format_key, register_table_routes
-from osa.domain.data.query.read_table import ReadRecordsTable, ReadRecordsTableHandler
+from osa.domain.data.query.read_table import (
+    ReadMode,
+    ReadRecordsTable,
+    ReadRecordsTableHandler,
+)
 
 
 def _make_get_endpoint(fmt: DataResponseFormat):
@@ -36,6 +40,7 @@ async def endpoint(
                 cursor=cursor,
                 limit=limit,
                 sort=parse_sort(sort),
+                mode=ReadMode.PAGE if fmt.paginated else ReadMode.STREAM,
                 timeout=fmt.timeout,
             )
         )
@@ -58,6 +63,7 @@ async def endpoint(
                 cursor=body.cursor,
                 limit=body.limit,
                 sort=parse_sort(body.sort),
+                mode=ReadMode.PAGE if fmt.paginated else ReadMode.STREAM,
                 timeout=fmt.timeout,
             )
         )

server/osa/application/api/v1/routes/stats.py

@@ -6,7 +6,12 @@
 from fastapi import APIRouter
 from pydantic import BaseModel
 
-from osa.domain.record.query.get_stats import GetStats, GetStatsHandler
+from osa.domain.data.command.verify_statistics import (
+    StatisticsDriftReport,
+    VerifyTableStatistics,
+    VerifyTableStatisticsHandler,
+)
+from osa.domain.data.query.get_stats import GetStats, GetStatsHandler
 
 router = APIRouter(
     prefix="/stats",
@@ -36,6 +41,19 @@ class StatsResponse(BaseModel):
     data_url: str = "/api/v1/data"
 
 
+@router.post("/verify")
+async def verify_statistics(
+    handler: FromDishka[VerifyTableStatisticsHandler],
+    repair: bool = False,
+) -> StatisticsDriftReport:
+    """Recompute table_statistics truth; report drift; repair on request.
+
+    ADMIN-gated (handler ``__auth__``). The only sanctioned whole-table
+    counting after deploy — defence in depth for the lockstep counts (#219).
+    """
+    return await handler.run(VerifyTableStatistics(repair=repair))
+
+
 @router.get("")
 async def get_stats(
     handler: FromDishka[GetStatsHandler],

server/osa/application/workflow/process_batch.py

@@ -606,7 +606,9 @@ async def _publish(
             ingest_run_id=event.ingest_run_id,
         )
 
-        # Checkpoint C: records durable before feature inserts (separate engine + FK).
+        # Checkpoint C: a redo boundary — records durable so a crash during the
+        # feature stage replays from here. (Feature DML shares the UoW session
+        # since #219 phase 4; the old separate-engine FK ordering is gone.)
         await self.uow.commit()
         return mapping
 

server/osa/domain/data/command/verify_statistics.py

@@ -0,0 +1,41 @@
+"""Admin verifier for ``table_statistics`` (#219 phase 6).
+
+The lockstep counts are load-bearing for every discovery surface; this command
+is the trust-but-verify safety net that lets exact maintenance replace live
+counting. It recomputes truth with the backfill migration's query, reports any
+drift, and overwrites only when explicitly asked. Defence in depth — never
+load-bearing, never scheduled.
+"""
+
+from osa.domain.auth.model.principal import Principal
+from osa.domain.auth.model.role import Role
+from osa.domain.data.model.statistics import StatisticsDrift
+from osa.domain.data.port.statistics_store import StatisticsStore
+from osa.domain.shared.authorization.gate import at_least
+from osa.domain.shared.command import Command, CommandHandler, Result
+
+
+class VerifyTableStatistics(Command):
+    repair: bool = False
+
+
+class StatisticsDriftReport(Result):
+    drift: list[StatisticsDrift]
+    repaired: bool
+
+
+class VerifyTableStatisticsHandler(CommandHandler[VerifyTableStatistics, StatisticsDriftReport]):
+    """Recompute → diff → report; mutate only on ``repair=True``."""
+
+    __auth__ = at_least(Role.ADMIN)
+
+    principal: Principal
+    stats_store: StatisticsStore
+
+    async def run(self, cmd: VerifyTableStatistics) -> StatisticsDriftReport:
+        drift = await self.stats_store.table_statistics_drift()
+        repaired = False
+        if cmd.repair and drift:
+            await self.stats_store.repair_table_statistics()
+            repaired = True
+        return StatisticsDriftReport(drift=drift, repaired=repaired)

server/osa/domain/data/model/query_plan.py

@@ -18,7 +18,7 @@
 from collections.abc import AsyncIterator, Mapping
 from dataclasses import dataclass
 from enum import StrEnum
-from typing import Any
+from typing import Annotated, Any, Literal
 
 from pydantic import BaseModel, Field, model_validator
 
@@ -53,7 +53,14 @@ def __str__(self) -> str:
         return self.value
 
 
-class PaginationParams(BaseModel):
+class BoundedPage(BaseModel):
+    """A page read: cursor + limit, compiled to ``LIMIT limit+1`` in SQL.
+
+    The only pagination interactive paths can construct — reading without a
+    bound requires naming :class:`FullStream` explicitly (#219 phase 2).
+    """
+
+    mode: Literal["page"] = "page"
     cursor: PaginationCursor | None = None
     limit: int = Field(default=50, ge=1)
 
@@ -64,8 +71,8 @@ def clamped(
         cursor: PaginationCursor | None = None,
         limit: int,
         max_limit: int,
-    ) -> "PaginationParams":
-        """Build params with ``limit`` clamped into ``[1, max_limit]``.
+    ) -> "BoundedPage":
+        """Build a page with ``limit`` clamped into ``[1, max_limit]``.
 
         Clamp, don't reject: a consumer asking for "everything" with a big
         number gets the max page, not a 422. The ceiling is operator
@@ -75,6 +82,19 @@ def clamped(
         return cls(cursor=cursor, limit=max(1, min(limit, max_limit)))
 
 
+class FullStream(BaseModel):
+    """An explicitly unbounded read — CSV / gzipped-CSV dumps only.
+
+    Carries no limit and no cursor by construction; the store serves it
+    through a server-side cursor so memory stays bounded.
+    """
+
+    mode: Literal["stream"] = "stream"
+
+
+Pagination = Annotated[BoundedPage | FullStream, Field(discriminator="mode")]
+
+
 class Keyset(BaseModel):
     """The keyset-pagination contract for a plan — the single source of truth
     for which column is the effective primary sort and which breaks ties.
@@ -106,7 +126,7 @@ def cursor_from_row(self, row: Mapping[str, Any]) -> str:
     TableKind.FEATURE: "id",
 }
 
-# Default sort keys per table kind (data-model.md §PaginationParams).
+# Default sort keys per table kind.
 _DEFAULT_SORTS: dict[TableKind, list[SortSpec]] = {
     TableKind.RECORDS: [
         SortSpec(column="created_at", direction=SortDirection.DESC),
@@ -121,7 +141,7 @@ class QueryPlan(BaseModel):
     table_kind: TableKind
     feature_name: FeatureName | None = None
     filter: FilterExpr | None = None
-    pagination: PaginationParams = Field(default_factory=PaginationParams)
+    pagination: Pagination = Field(default_factory=BoundedPage)
     sort: list[SortSpec] = Field(default_factory=list)
 
     @model_validator(mode="after")
@@ -145,6 +165,8 @@ async def take_page(self, rows: AsyncIterator[Mapping[str, Any]]) -> PageSlice:
         the REST paginated-JSON path and the view queries build on this, so
         the encode side of pagination cannot fork.
         """
+        if not isinstance(self.pagination, BoundedPage):
+            raise ValueError("take_page requires a BoundedPage plan; dumps never paginate")
         limit = self.pagination.limit
         page: list[Mapping[str, Any]] = []
         truncated = False