perf: bound Dashboard backfill priority plan

backend/app/post_content_queue.py

@@ -381,28 +381,21 @@ async def ensure_post_content_job(
     )
 
 
-async def enqueue_post_content_backfill(
-    pool: asyncpg.Pool,
-    client: redis.Redis | None,
-    *,
-    limit: int,
-    require_embedding: bool,
-    require_structure: bool,
-) -> dict[str, int]:
-    """Durably enqueue one bounded page of eligible incomplete source posts.
-
-    PostgreSQL is committed before Valkey is touched.  A missing wake-up is
-    therefore recoverable by :func:`republish_queued_post_content_jobs` rather
-    than turning an operator request into lost work.  Active and terminal jobs
-    are excluded so repeated requests neither duplicate work nor reset the
-    explicit retry boundary.
-    """
-    if not 1 <= limit <= 200:
-        raise ValueError("limit must be between 1 and 200")
-    query = f"""
+POST_CONTENT_BACKFILL_CANDIDATE_SQL = f"""
         select post.post_id, post.post_body
           from source_post post
           left join post_content_ingestion_job job on job.post_id = post.post_id
+          left join operations_case_analysis analysis
+            on analysis.post_id = post.post_id
+           and analysis.source_body_sha256 = job.source_body_sha256
+          left join post_product_analysis product_analysis
+            on product_analysis.post_id = post.post_id
+           and product_analysis.source_body_sha256 = job.source_body_sha256
+          left join (
+              select distinct project.post_id
+                from post_project_mention project
+               where nullif(btrim(project.ontology_iri), '') is not null
+          ) ontology_project on ontology_project.post_id = post.post_id
          where {SOURCE_POST_ELIGIBILITY_SQL.format(alias='post')}
            and (job.post_id is null or job.status_code = $1)
            and (
@@ -443,53 +436,74 @@ async def enqueue_post_content_backfill(
                           or structure.decision_source_code = 'unresolved'
                       )
                ))
-               or ($3::boolean and not exists (
-                   select 1
-                     from operations_case_analysis analysis
-                    where analysis.post_id = post.post_id
-                      and analysis.source_body_sha256 = job.source_body_sha256
-               ))
-               or ($3::boolean and not exists (
-                   select 1
-                     from post_product_analysis analysis
-                    where analysis.post_id = post.post_id
-                      and analysis.source_body_sha256 = job.source_body_sha256
-               ))
+               or ($3::boolean and analysis.post_id is null)
+               or ($3::boolean and product_analysis.post_id is null)
            )
-         order by case
-                      when $3::boolean
-                       and exists (
-                           select 1
-                             from post_project_mention project
-                            where project.post_id = post.post_id
-                              and nullif(btrim(project.ontology_iri), '') is not null
-                       )
-                       and job.source_body_sha256 is not null
-                       and not exists (
-                           select 1
-                             from operations_case_analysis analysis
-                            where analysis.post_id = post.post_id
-                              and analysis.source_body_sha256 = job.source_body_sha256
-                       )
-                      then 0 else 1
-                  end,
-                  coalesce(post.event_occurred_at, post.created_at),
+           and ($5::boolean = (
+                   $3::boolean
+                   and ontology_project.post_id is not null
+                   and job.source_body_sha256 is not null
+                   and analysis.post_id is null
+               ))
+         order by coalesce(post.event_occurred_at, post.created_at),
                   post.created_at,
                   post.post_id
          limit $4
          for update of post skip locked
     """
+
+
+async def enqueue_post_content_backfill(
+    pool: asyncpg.Pool,
+    client: redis.Redis | None,
+    *,
+    limit: int,
+    require_embedding: bool,
+    require_structure: bool,
+) -> dict[str, int]:
+    """Durably enqueue one bounded page of eligible incomplete source posts.
+
+    PostgreSQL is committed before Valkey is touched.  A missing wake-up is
+    therefore recoverable by :func:`republish_queued_post_content_jobs` rather
+    than turning an operator request into lost work.  Active and terminal jobs
+    are excluded so repeated requests neither duplicate work nor reset the
+    explicit retry boundary.
+    """
+    if not 1 <= limit <= 200:
+        raise ValueError("limit must be between 1 and 200")
     requests: list[PostContentJobRequest] = []
     async with pool.acquire() as conn:
         async with conn.transaction():
-            # Safe SQL: the eligibility predicate is an immutable schema fragment; values are bound.
-            rows = await conn.fetch(  # nosemgrep: python.lang.security.audit.sqli.asyncpg-sqli.asyncpg-sqli
-                query,
-                SUCCEEDED,
-                require_embedding,
-                require_structure,
-                limit,
-            )
+            rows = []
+            if require_structure:
+                # Safe SQL: the eligibility predicate is an immutable schema fragment; values are bound.
+                rows = await conn.fetch(  # nosemgrep: python.lang.security.audit.sqli.asyncpg-sqli.asyncpg-sqli
+                    POST_CONTENT_BACKFILL_CANDIDATE_SQL,
+                    SUCCEEDED,
+                    require_embedding,
+                    require_structure,
+                    limit,
+                    True,
+                )
+            if len(rows) < limit:
+                # Safe SQL: the same immutable candidate statement is reused with bound tier values.
+                rows += await conn.fetch(  # nosemgrep: python.lang.security.audit.sqli.asyncpg-sqli.asyncpg-sqli
+                    POST_CONTENT_BACKFILL_CANDIDATE_SQL,
+                    SUCCEEDED,
+                    require_embedding,
+                    require_structure,
+                    limit - len(rows),
+                    False,
+                )
+            unique_rows = []
+            seen_post_ids: set[str] = set()
+            for row in rows:
+                post_id = str(row["post_id"])
+                if post_id in seen_post_ids:
+                    continue
+                seen_post_ids.add(post_id)
+                unique_rows.append(row)
+            rows = unique_rows
             for row in rows:
                 post_id = str(row["post_id"])
                 body = str(row["post_body"] or "")

docs/adr/0206-evidence-operations-dashboard.md

@@ -221,6 +221,18 @@ treated as a negative case.
   browser output, and k6 evidence
   outside the repository. An empty synthetic case list remains a valid UI/API
   shape check; it is not evidence that grounded production cases exist.
+- `scripts/explain_post_content_backfill.py` executes the exact bounded
+  candidate SQL with `EXPLAIN (ANALYZE, BUFFERS, WAL, FORMAT JSON)` inside a
+  rolled-back transaction. It reports only aggregate timing, buffer, temporary
+  block, node-kind, and relation-scan counts, so priority-sort, correlated
+  subquery, index, spill, and lock-path evidence is reproducible without
+  emitting source rows.
+- Backfill admission reads the ontology-backed priority tier first and reads
+  the remaining eligible tier only when fewer than the requested bounded page
+  are locked. This preserves the documented total order while avoiding a
+  corpus-wide priority `CASE` sort and its per-row correlated probes.
+  Candidate post identifiers are de-duplicated before mutation because the two
+  `READ COMMITTED` statements may observe a target moving between tiers.
 
 ## References
 

docs/product-technical-gap-baseline.md

@@ -89,9 +89,32 @@ Security/operability: every aggregation applies `post_read` plus row-level
 corporate-entity visibility before counting; source-body digests invalidate
 stale inference; provider errors persist no positive/negative result; PII
 remains authorized at the UI boundary and is excluded from telemetry. The
-tables use composite keys and bounded kind-first indexes; production hot-path
-acceptance still requires `EXPLAIN (ANALYZE, BUFFERS)` on an anonymized runtime
-snapshot.
+tables use composite keys and bounded kind-first indexes. Production hot-path
+acceptance uses `scripts/explain_post_content_backfill.py` on an anonymized
+runtime snapshot; the exact candidate SQL runs in a rolled-back transaction
+and emits aggregate plan/buffer metrics only. A deployment-specific
+capacity/SLO remains separate from this query-shape evidence.
+
+On an isolated exact-schema synthetic snapshot based on #716 `c01de078`
+(20,000 eligible posts and jobs, 9,927 ontology-backed project mentions, and
+4,951 current operations analyses), a consecutive rolled-back comparison
+returned the same 200-row priority page in 2,056.629 ms before and 1,327.868 ms
+after the change. Root shared-hit blocks fell from 275,642 to 100,514; both
+plans recorded zero shared reads and zero temporary reads/writes. The former
+plan made 20,000 correlated project probes and 9,927 correlated
+operations-analysis probes, while the semantics-equivalent two-tier query
+removes the corpus-wide priority `CASE` and its extra correlated priority
+subplans. The reproducible summary includes both relation plan-node counts and
+actual scan-loop totals so a single nested-loop node cannot be mislabeled as a
+single execution.
+The plan remains `Limit -> LockRows -> Sort`; `SKIP LOCKED` and the transaction
+boundary therefore remain intact, and the remaining tier runs only when the
+priority tier cannot fill the requested page. A separate remaining-tier
+observation returned 200 rows in 12,649.654 ms with 241,317 root shared-hit
+blocks and no reads or temporary spill; it is retained as the next
+distribution-specific optimization target, not hidden by the priority-path
+improvement. These observations establish query shape only, not a deployment
+capacity or latency SLO.
 
 ### Historical UI audit evidence
 

scripts/explain_post_content_backfill.py

@@ -0,0 +1,108 @@
+#!/usr/bin/env python3
+"""Measure the exact backfill candidate query without exposing source rows."""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import os
+from collections import Counter
+from collections.abc import Iterator, Mapping
+from typing import Any
+
+import asyncpg
+
+from backend.app.post_content_queue import POST_CONTENT_BACKFILL_CANDIDATE_SQL, SUCCEEDED
+
+
+def _nodes(plan: Mapping[str, Any]) -> Iterator[Mapping[str, Any]]:
+    """Yield every PostgreSQL plan node without retaining result rows."""
+    yield plan
+    for child in plan.get("Plans", ()):
+        yield from _nodes(child)
+
+
+def summarize_plan(document: list[Mapping[str, Any]]) -> dict[str, Any]:
+    """Project EXPLAIN JSON into non-identifying aggregate plan evidence."""
+    root = document[0]
+    nodes = tuple(_nodes(root["Plan"]))
+    node_counts = Counter(str(node["Node Type"]) for node in nodes)
+    relation_scans = Counter(
+        str(node["Relation Name"]) for node in nodes if "Relation Name" in node
+    )
+    relation_scan_loops = Counter()
+    for node in nodes:
+        if "Relation Name" in node:
+            relation_scan_loops[str(node["Relation Name"])] += int(
+                node.get("Actual Loops", 0)
+            )
+    return {
+        "planning_time_ms": root.get("Planning Time"),
+        "execution_time_ms": root.get("Execution Time"),
+        "actual_rows": root["Plan"].get("Actual Rows"),
+        "shared_hit_blocks": int(root["Plan"].get("Shared Hit Blocks", 0)),
+        "shared_read_blocks": int(root["Plan"].get("Shared Read Blocks", 0)),
+        "temp_read_blocks": int(root["Plan"].get("Temp Read Blocks", 0)),
+        "temp_written_blocks": int(root["Plan"].get("Temp Written Blocks", 0)),
+        "node_counts": dict(sorted(node_counts.items())),
+        "relation_scans": dict(sorted(relation_scans.items())),
+        "relation_scan_loops": dict(sorted(relation_scan_loops.items())),
+    }
+
+
+async def _measure(
+    dsn: str,
+    *,
+    limit: int,
+    embeddings: bool,
+    structure: bool,
+    priority: bool,
+) -> dict[str, Any]:
+    """Run EXPLAIN inside a rolled-back transaction and return its summary."""
+    conn = await asyncpg.connect(dsn)
+    transaction = conn.transaction()
+    await transaction.start()
+    try:
+        value = await conn.fetchval(
+            "EXPLAIN (ANALYZE, BUFFERS, WAL, FORMAT JSON) "
+            + POST_CONTENT_BACKFILL_CANDIDATE_SQL,
+            SUCCEEDED,
+            embeddings,
+            structure,
+            limit,
+            priority,
+        )
+        document = json.loads(value) if isinstance(value, str) else value
+        return summarize_plan(document)
+    finally:
+        await transaction.rollback()
+        await conn.close()
+
+
+def main() -> None:
+    """Parse bounded operator inputs and print aggregate JSON only."""
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--dsn", default=os.environ.get("DATABASE_URL"))
+    parser.add_argument("--limit", type=int, default=200, choices=range(1, 201))
+    parser.add_argument("--embeddings", action="store_true")
+    parser.add_argument("--structure", action="store_true")
+    parser.add_argument("--tier", choices=("priority", "remaining"), default="priority")
+    args = parser.parse_args()
+    if not args.dsn:
+        parser.error("--dsn or DATABASE_URL is required")
+    result = asyncio.run(
+        _measure(
+            args.dsn,
+            limit=args.limit,
+            embeddings=args.embeddings,
+            structure=args.structure,
+            priority=args.tier == "priority",
+        )
+    )
+    result["candidate_tier"] = args.tier
+    print(json.dumps(result, indent=2, sort_keys=True))
+
+
+if __name__ == "__main__":
+    main()

tests/test_explain_post_content_backfill.py

@@ -0,0 +1,21 @@
+"""Tests for non-identifying backfill plan evidence."""
+
+from scripts.explain_post_content_backfill import summarize_plan
+
+
+def test_summarize_plan_reports_aggregate_buffers_and_relations_only() -> None:
+    """The evidence summary contains plan metrics but no source-row values."""
+    result = summarize_plan([{"Planning Time": 1.25, "Execution Time": 2.5, "Plan": {"Node Type": "Limit", "Actual Rows": 12, "Shared Hit Blocks": 2, "Plans": [{"Node Type": "Index Scan", "Relation Name": "source_post", "Actual Loops": 4, "Shared Hit Blocks": 3, "Shared Read Blocks": 1}]}}])
+
+    assert result == {
+        "planning_time_ms": 1.25,
+        "execution_time_ms": 2.5,
+        "actual_rows": 12,
+        "shared_hit_blocks": 2,
+        "shared_read_blocks": 0,
+        "temp_read_blocks": 0,
+        "temp_written_blocks": 0,
+        "node_counts": {"Index Scan": 1, "Limit": 1},
+        "relation_scans": {"source_post": 1},
+        "relation_scan_loops": {"source_post": 4},
+    }