Reduce recurring query spills and database work

CHANGELOG.md

@@ -6,6 +6,7 @@ Application releases and Alembic schema revisions are separate identifiers.
 
 ### Fixes
 
+- Reduce database work and temporary disk writes during search-index reconciliation and admin publication reporting while retaining exact visibility and automation counts.
 - Reduce delays in automated article review as the topic and tag catalog grows.
 - Keep topic edits responsive while completed article analyses prepare publication checks.
 - Keep sign-in codes, state values, and authentication credentials out of request logs and browser diagnostics.

apps/admin-api/src/automation.py

@@ -564,7 +564,7 @@ def automation_blockers(session):
     return blockers
 
 
-def publication_automation(session, start, now):
+def _publication_automation_query(start, now):
     publications = (
         select(Article.id, Article.published_to_feed_at, Article.discovered_at)
         .where(
@@ -573,20 +573,32 @@ def publication_automation(session, start, now):
         )
         .cte("publications")
     )
-    reviews = (
+    # OFFSET 0 keeps this boolean projection below the grouping operation without
+    # materializing it. Otherwise PostgreSQL can sort full automation evidence for
+    # every review, producing large temporary writes.
+    review_flags = (
         select(
             ArticleReview.article_id,
-            func.bool_or(
-                (ArticleReview.action == "publish") & (ArticleReview.automation != {})
-            ).label("automatic"),
-            func.bool_or(ArticleReview.automation == {}).label("manual"),
+            ((ArticleReview.action == "publish") & (ArticleReview.automation != {})).label(
+                "automatic"
+            ),
+            (ArticleReview.automation == {}).label("manual"),
         )
         .join(publications, publications.c.id == ArticleReview.article_id)
         .where(ArticleReview.created_at <= publications.c.published_to_feed_at)
-        .group_by(ArticleReview.article_id)
+        .offset(0)
+        .subquery("publication_review_flags")
+    )
+    reviews = (
+        select(
+            review_flags.c.article_id,
+            func.bool_or(review_flags.c.automatic).label("automatic"),
+            func.bool_or(review_flags.c.manual).label("manual"),
+        )
+        .group_by(review_flags.c.article_id)
         .subquery()
     )
-    published, autonomous, median = session.execute(
+    return (
         select(
             func.count(),
             func.count().filter(reviews.c.automatic, ~reviews.c.manual),
@@ -600,7 +612,11 @@ def publication_automation(session, start, now):
         )
         .select_from(publications)
         .outerjoin(reviews, reviews.c.article_id == publications.c.id)
-    ).one()
+    )
+
+
+def publication_automation(session, start, now):
+    published, autonomous, median = session.execute(_publication_automation_query(start, now)).one()
     return dict(
         published_in_window=published,
         published_without_intervention=autonomous,

packages/core/OBSERVABILITY.md

@@ -174,6 +174,32 @@ and use bounded read-only execution plans before changing a query. Search-index
 reconciliation intentionally counts visible articles and tags exactly; those
 recurring aggregate queries are distinct from reader request latency.
 
+Tag reconciliation counts distinct tag UUIDs from visible article assignments.
+It avoids building a semi-join hash table over every assignment, while preserving
+publication, review and approved-origin checks. Publication reporting projects
+review UUIDs and automatic/manual boolean flags before grouping. Keep its `OFFSET 0`
+projection boundary: removing it can let PostgreSQL sort full automation JSON.
+Reviews after publication remain excluded, and any manual review before publication
+still excludes the article from the autonomous count.
+
+The recurring-query regression uses approximately production cardinalities: 60,340
+articles, 34,074 tags, 603,400 assignments and 153,680 reviews. It compares exact
+application SQL with the previous queries at `work_mem=4MB`, including 1-, 7-, 30-
+and 90-day reporting windows. Plan budgets enforce zero tag-count temporary writes
+and substantially smaller reporting spills; elapsed times are reported rather than
+asserted because they depend on the machine and cache state. Run it with disposable
+test URLs configured as described in `tests/conftest.py`:
+
+```sh
+DEVFEED_RECURRING_PROFILE_REPORT=/tmp/devfeed-recurring-queries.json \
+  uv run --locked pytest -q tests/test_recurring_query_budgets.py
+```
+
+Compare post-deployment deltas in `pg_stat_statements` calls, execution time and
+temporary blocks per call, plus the DevFeed database's 24-hour `temp_bytes` increase.
+The daily database total includes other workloads; improvements in these queries
+alone do not establish how much of that total has been removed.
+
 Topic-catalog lock waits can come from application work in the transaction holding
 the lock. Inspect `pg_locks` together with `pg_stat_activity`, including transactions
 waiting on `ClientRead`. Candidate selection reuses immutable normalized identities

packages/core/src/search_index.py

@@ -14,7 +14,6 @@
     ArticleTopic,
     SearchEvent,
     Source,
-    Tag,
     Topic,
 )
 from devfeed_core.publication import visible_article
@@ -34,6 +33,19 @@ def _metric_instruments():
     return runtime.instruments if runtime else None
 
 
+def _visible_tag_count_query():
+    # Deduplicate destination UUIDs before counting. A tag EXISTS semi-join can
+    # hash every matching assignment and spill hundreds of thousands of rows.
+    visible_tags = (
+        select(ArticleTag.tag_id)
+        .join(Article, Article.id == ArticleTag.article_id)
+        .where(visible_article())
+        .distinct()
+        .subquery()
+    )
+    return select(func.count()).select_from(visible_tags)
+
+
 def _visible_counts(session):
     counts = {
         "articles": session.scalar(
@@ -73,17 +85,7 @@ def _visible_counts(session):
                 .exists(),
             )
         ),
-        "tags": session.scalar(
-            select(func.count())
-            .select_from(Tag)
-            .where(
-                select(1)
-                .select_from(ArticleTag)
-                .join(Article, Article.id == ArticleTag.article_id)
-                .where(ArticleTag.tag_id == Tag.id, visible_article())
-                .exists()
-            )
-        ),
+        "tags": session.scalar(_visible_tag_count_query()),
     }
     return {kind: int(value or 0) for kind, value in counts.items()}