From 96a791791bdfe89dc758c4c22ee0b965058da2ad Mon Sep 17 00:00:00 2001 From: Abhimanyu Saharan Date: Sun, 4 Oct 2026 05:28:35 +0530 Subject: [PATCH] Reduce recurring query spills and database work --- CHANGELOG.md | 6 + apps/admin-api/src/automation.py | 34 ++- packages/core/OBSERVABILITY.md | 26 ++ packages/core/src/search_index.py | 26 +- tests/test_recurring_query_budgets.py | 370 ++++++++++++++++++++++++++ 5 files changed, 441 insertions(+), 21 deletions(-) create mode 100644 tests/test_recurring_query_budgets.py diff --git a/CHANGELOG.md b/CHANGELOG.md index af88dcc49..2497b1ac0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,12 @@ Application releases and Alembic schema revisions are separate identifiers. +## Unreleased + +### Fixes + +- Reduce database work and temporary disk writes during search-index reconciliation and admin publication reporting while retaining exact visibility and automation counts. + ## 0.0.50 — 2026-10-03 ### Enhancements diff --git a/apps/admin-api/src/automation.py b/apps/admin-api/src/automation.py index 4191fdb22..1429df738 100644 --- a/apps/admin-api/src/automation.py +++ b/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, diff --git a/packages/core/OBSERVABILITY.md b/packages/core/OBSERVABILITY.md index fa635b0ff..8b920dcf1 100644 --- a/packages/core/OBSERVABILITY.md +++ b/packages/core/OBSERVABILITY.md @@ -166,6 +166,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 diff --git a/packages/core/src/search_index.py b/packages/core/src/search_index.py index e25676786..83eaf3817 100644 --- a/packages/core/src/search_index.py +++ b/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()} diff --git a/tests/test_recurring_query_budgets.py b/tests/test_recurring_query_budgets.py new file mode 100644 index 000000000..9979aa3cb --- /dev/null +++ b/tests/test_recurring_query_budgets.py @@ -0,0 +1,370 @@ +"""Exact recurring counts and bounded plans on disposable, synthetic PostgreSQL data.""" + +import json +import os +import statistics +import time +from datetime import UTC, datetime, timedelta +from pathlib import Path + +import pytest +from devfeed_admin_api.automation import _publication_automation_query, publication_automation +from devfeed_core.db import get_engine +from devfeed_core.models import ( + Article, + ArticleOrigin, + ArticleReview, + ArticleTag, + ArticleTopic, + Source, + Tag, + Topic, +) +from devfeed_core.search_index import _visible_counts, _visible_tag_count_query +from sqlalchemy import event, text + +pytestmark = pytest.mark.integration + +NOW = datetime(2026, 10, 3, 12, tzinfo=UTC) + +# Previous production queries are reference workloads, not an implementation of +# the new paths. Keep their wide joins to detect reintroduced spills at scale. +PREVIOUS_TAG_COUNT = text(""" + SELECT count(*) FROM tags t WHERE EXISTS ( + SELECT 1 FROM article_tags at JOIN articles a ON a.id=at.article_id + WHERE at.tag_id=t.id AND a.publication_status='published' AND a.review_status='approved' + AND EXISTS (SELECT 1 FROM article_origins o JOIN sources s ON s.id=o.source_id + WHERE o.article_id=a.id AND s.approval_status='approved')) +""") +PREVIOUS_PUBLICATION_REPORT = text(""" + WITH publications AS ( + SELECT id,published_to_feed_at,discovered_at FROM articles + WHERE published_to_feed_at>=:start AND published_to_feed_at<=:now) + SELECT count(*), count(*) FILTER (WHERE r.automatic AND NOT r.manual), + percentile_cont(0.5) WITHIN GROUP ( + ORDER BY EXTRACT(epoch FROM p.published_to_feed_at-p.discovered_at)) + FILTER (WHERE p.published_to_feed_at>=p.discovered_at) + FROM publications p LEFT JOIN ( + SELECT article_id, bool_or(action='publish' AND automation!='{}'::jsonb) automatic, + bool_or(automation='{}'::jsonb) manual + FROM article_reviews r JOIN publications p ON p.id=r.article_id + WHERE r.created_at<=p.published_to_feed_at GROUP BY article_id + ) r ON r.article_id=p.id +""") + + +def test_visible_counts_preserve_visibility_and_distinct_destinations(database): + with database.begin() as session: + enabled = Source( + name="Enabled", + feed_url="https://enabled.test/rss", + source_type="publisher", + approval_status="approved", + ) + disabled = Source( + name="Disabled", + feed_url="https://disabled.test/rss", + source_type="publisher", + approval_status="approved", + enabled=False, + ) + rejected = Source(name="Rejected", source_type="publisher", approval_status="rejected") + primary = Topic(name="Primary", slug="primary", kind="technology", status="active") + supporting = Topic(name="Supporting", slug="supporting", kind="technology", status="active") + incidental = Topic(name="Incidental", slug="incidental", kind="technology", status="active") + proposed = Topic(name="Proposed", slug="proposed", kind="technology", status="proposed") + shared = Tag(name="Shared", slug="shared") + disabled_tag = Tag(name="Disabled source", slug="disabled-source") + private = Tag(name="Private", slug="private") + session.add_all( + [ + enabled, + disabled, + rejected, + primary, + supporting, + incidental, + proposed, + shared, + disabled_tag, + private, + Tag(name="Unlinked", slug="unlinked"), + ] + ) + session.flush() + for i, (publisher, publication, review, tag) in enumerate( + [ + (enabled, "published", "approved", shared), + (enabled, "published", "approved", shared), + (disabled, "published", "approved", disabled_tag), + (rejected, "published", "approved", private), + (enabled, "unpublished", "approved", private), + (enabled, "unpublished", "pending", private), + (enabled, "unpublished", "rejected", private), + (None, "published", "approved", private), + ] + ): + article = Article( + title=f"Article {i}", + canonical_url=f"https://article.test/{i}", + url_hash=str(i), + publication_status=publication, + review_status=review, + ) + session.add(article) + session.flush() + session.add(ArticleTag(article_id=article.id, tag_id=tag.id)) + if publisher: + session.add( + ArticleOrigin( + article_id=article.id, + source_id=publisher.id, + entry_key=str(i), + original_url=article.canonical_url, + ) + ) + if i < 2: + # Duplicate approved origins must not multiply any count. + session.add( + ArticleOrigin( + article_id=article.id, + source_id=enabled.id, + entry_key=f"duplicate-{i}", + original_url=article.canonical_url, + ) + ) + for topic, role in ( + (primary, "primary"), + (supporting, "supporting"), + (incidental, "incidental"), + (proposed, "primary"), + ): + session.add( + ArticleTopic( + article_id=article.id, + topic_id=topic.id, + role=role, + relevance=1.0, + evidence="Synthetic evidence", + ) + ) + session.flush() + assert _visible_counts(session) == dict(articles=3, topics=2, sources=1, tags=2) + enabled.approval_status = "rejected" + session.flush() + # Disabled approved sources still make articles visible, but aren't indexed + # as discoverable sources themselves. Visibility changes must count now. + assert _visible_counts(session) == dict(articles=1, topics=0, sources=0, tags=1) + disabled.approval_status = "rejected" + session.flush() + assert _visible_counts(session) == dict(articles=0, topics=0, sources=0, tags=0) + + +@pytest.mark.parametrize( + ("reviews", "autonomous"), + [ + ([], 0), + ([("approve", {"policy": "test"}, -1)], 0), + ([("publish", {}, -1)], 0), + ([("publish", {"policy": "test"}, 1)], 0), + ([("publish", {"policy": "test"}, 0)], 1), + ([("publish", {"policy": "test"}, -1), ("approve", {}, -2)], 0), + ([("publish", {"policy": "test"}, -1), ("approve", {}, 0)], 0), + ([("publish", {"policy": "test"}, -1), ("approve", {}, 1)], 1), + ([("publish", {"policy": "test"}, -2), ("publish", {"policy": "test"}, -1)], 1), + ], +) +def test_publication_review_flags_preserve_intervention_and_cutoff(database, reviews, autonomous): + with database.begin() as session: + article = Article( + title="Synthetic publication", + canonical_url="https://article.test/1", + url_hash="1", + discovered_at=NOW - timedelta(minutes=10), + published_to_feed_at=NOW, + ) + session.add(article) + session.flush() + session.add_all( + [ + ArticleReview( + article_id=article.id, + action=action, + revision=i, + automation=automation, + created_at=NOW + timedelta(seconds=offset), + ) + for i, (action, automation, offset) in enumerate(reviews) + ] + ) + session.flush() + assert publication_automation(session, NOW - timedelta(days=1), NOW) == dict( + published_in_window=1, + published_without_intervention=autonomous, + automatic_publication_percent=100.0 if autonomous else 0.0, + median_ingestion_to_publication_seconds=600.0, + ) + + +def test_publication_window_boundaries_empty_windows_and_nonnegative_median(database): + with database.begin() as session: + start = NOW - timedelta(days=1) + for i, (published, delay) in enumerate( + [ + (None, 0), + (start - timedelta(microseconds=1), 600), + (start, 120), + (NOW, 480), + (NOW + timedelta(microseconds=1), 600), + (NOW, -10), + ] + ): + session.add( + Article( + title=f"Publication {i}", + canonical_url=f"https://article.test/{i}", + url_hash=str(i), + published_to_feed_at=published, + discovered_at=(published or NOW) - timedelta(seconds=delay), + ) + ) + session.flush() + assert publication_automation(session, start, NOW) == dict( + published_in_window=3, + published_without_intervention=0, + automatic_publication_percent=0.0, + median_ingestion_to_publication_seconds=300.0, + ) + assert publication_automation( + session, NOW + timedelta(days=1), NOW + timedelta(days=2) + ) == dict( + published_in_window=0, + published_without_intervention=0, + automatic_publication_percent=None, + median_ingestion_to_publication_seconds=None, + ) + + +@pytest.fixture +def recurring_query_data(database): + # Match the production cardinalities and the report's wide inline JSON. Using + # raw INSERT SELECT keeps this reproducible without external data or credentials. + with get_engine().begin() as connection: + # Bulk fixture creation exercises foreign keys and outbox triggers. Allow + # it longer than the application timeout only in this disposable database. + connection.execute(text("SET LOCAL statement_timeout='5min'")) + for sql in ( + """INSERT INTO sources(id,name,feed_url,source_type,approval_status,enabled, + submission_channel,updated_at,poll_interval_seconds,next_fetch_at, + consecutive_failures,created_at,slug) + SELECT md5('s'||i)::uuid,'Publisher '||i,'https://publisher.test/'||i,'publisher', + CASE WHEN i<=1196 THEN 'approved' ELSE 'rejected' END,i<=1190, + 'cli',now(),1800,now(),0,now(),'publisher-'||i + FROM generate_series(1,1261) i""", + """INSERT INTO tags(id,name,slug,aliases) + SELECT md5('t'||i)::uuid,'Tag '||i,'tag-'||i,ARRAY[]::varchar[] + FROM generate_series(1,34074) i""", + """INSERT INTO articles(id,canonical_url,url_hash,title,summary,review_status, + publication_status,feed_at,published_to_feed_at,discovered_at,content_type,slug) + SELECT md5('a'||i)::uuid,'https://article.test/'||i,md5('a'||i),'Article '||i, + repeat('Synthetic article body ',30), + CASE WHEN i<=55730 THEN 'approved' ELSE 'rejected' END, + CASE WHEN i<=55730 THEN 'published' ELSE 'unpublished' END, + :now - (i%30)*interval '1 day', + CASE WHEN i<=55730 THEN :now - (i%30)*interval '1 day' END, + :now - interval '10 minutes' - (i%30)*interval '1 day','article','article-'||i + FROM generate_series(1,60340) i""", + """INSERT INTO article_origins(id,article_id,source_id,entry_key,original_url, + source_metadata) + SELECT gen_random_uuid(),md5('a'||i)::uuid,md5('s'||(1+i%1261))::uuid,i::text, + 'https://article.test/'||i,'{}'::jsonb FROM generate_series(1,60340) i""", + """INSERT INTO article_tags(article_id,tag_id) + SELECT md5('a'||i)::uuid,md5('t'||(1+(i*10+j)%34074))::uuid + FROM generate_series(1,60340) i CROSS JOIN generate_series(1,10) j""", + """INSERT INTO article_reviews(id,article_id,action,revision,automation,created_at) + SELECT gen_random_uuid(),a.id,CASE WHEN j=1 THEN 'publish' ELSE 'approve' END,j, + CASE WHEN i%31=0 AND j=2 THEN '{}'::jsonb + ELSE jsonb_build_object('policy','synthetic','context',repeat(md5(i::text),9)) + END,coalesce(a.published_to_feed_at,a.discovered_at)-interval '1 minute' + FROM generate_series(1,60340) i JOIN articles a ON a.id=md5('a'||i)::uuid + CROSS JOIN generate_series(1,3) j WHERE j<=2 OR i<=33000""", + ): + connection.execute(text(sql), {"now": NOW}) + for table in ( + "articles", + "sources", + "tags", + "article_origins", + "article_tags", + "article_reviews", + ): + connection.execute(text(f"ANALYZE {table}")) + + +def explain(connection, statement, parameters): + # Capture driver-ready values (including SQLAlchemy's JSONB bind processors), + # so EXPLAIN measures the exact application SQL, not a rewritten approximation. + captured = [] + + def capture(conn, cursor, sql, params, context, executemany): + captured.append((sql, params)) + + event.listen(connection, "before_cursor_execute", capture) + try: + result = tuple(connection.execute(statement, parameters).one()) + finally: + event.remove(connection, "before_cursor_execute", capture) + assert len(captured) == 1 + sql, params = captured[0] + plan = connection.exec_driver_sql("EXPLAIN (ANALYZE,BUFFERS,FORMAT JSON) " + sql, params) + return result, plan.scalar_one()[0] + + +def profile_pair(connection, before, after, parameters): + timings = {"before": [], "after": []} + for _ in range(3): + for name, statement in (("before", before), ("after", after)): + started = time.perf_counter() + connection.execute(statement, parameters).one() + timings[name].append(1000 * (time.perf_counter() - started)) + before_result, before_plan = explain(connection, before, parameters) + after_result, after_plan = explain(connection, after, parameters) + assert before_result == after_result + return { + "result": list(after_result), + "before_ms": statistics.median(timings["before"]), + "after_ms": statistics.median(timings["after"]), + "before_plan": before_plan, + "after_plan": after_plan, + } + + +def test_recurring_query_plans_bound_spills_at_production_scale(recurring_query_data): + reports = {} + with get_engine().connect() as connection: + # Compare both paths at production's 4 MB limit; no global DB tuning. + connection.execute(text("SET LOCAL work_mem='4MB'")) + connection.execute(text("SET LOCAL statement_timeout='15s'")) + reports["tags"] = profile_pair( + connection, PREVIOUS_TAG_COUNT, _visible_tag_count_query(), {} + ) + assert reports["tags"]["result"] == [34074] + assert reports["tags"]["after_plan"]["Plan"]["Temp Written Blocks"] == 0 + for days in (1, 7, 30, 90): + parameters = {"start": NOW - timedelta(days=days), "now": NOW} + report = profile_pair( + connection, + PREVIOUS_PUBLICATION_REPORT, + _publication_automation_query(parameters["start"], NOW), + parameters, + ) + reports[f"publication_{days}d"] = report + before_temp = report["before_plan"]["Plan"]["Temp Written Blocks"] + after_temp = report["after_plan"]["Plan"]["Temp Written Blocks"] + assert after_temp <= 1024 # 8 MB, including the shared publication CTE. + if days >= 30 and before_temp: + assert after_temp < before_temp * 0.15 + if days >= 30: + assert report["result"] == [55730, 53933, 600.0] + if target := os.environ.get("DEVFEED_RECURRING_PROFILE_REPORT"): + Path(target).write_text(json.dumps(reports, indent=2) + "\n")