Reuse scheduler catalog ranking before publication locks

CHANGELOG.md

@@ -6,6 +6,7 @@ Application releases and Alembic schema revisions are separate identifiers.
 
 ### Fixes
 
+- Reduce delays in automated article review as the topic and tag catalog grows.
 - Keep sign-in codes, state values, and authentication credentials out of request logs and browser diagnostics.
 
 ## 0.0.50 — 2026-10-03

migrations/versions/0022_catalog_ranking_revision.py

@@ -0,0 +1,41 @@
+"""Include topic description vocabulary in the authoritative ranking revision."""
+
+from alembic import op
+
+revision = "0022"
+down_revision = "0021"
+branch_labels = None
+depends_on = None
+
+IDENTITY_COLUMNS = "id, name, slug, aliases, kind, keywords, status"
+
+
+def replace_revision_function(columns):
+    op.execute(f"""
+        CREATE OR REPLACE FUNCTION refresh_topics_revision() RETURNS trigger
+        LANGUAGE plpgsql AS $fn$
+        BEGIN
+          IF TG_OP = 'INSERT' THEN
+            IF NOT EXISTS (SELECT 1 FROM new_rows) THEN RETURN NULL; END IF;
+          ELSIF TG_OP = 'DELETE' THEN
+            IF NOT EXISTS (SELECT 1 FROM old_rows) THEN RETURN NULL; END IF;
+          ELSIF TG_OP = 'UPDATE' THEN
+            IF NOT EXISTS (
+              SELECT {columns} FROM new_rows EXCEPT SELECT {columns} FROM old_rows
+            ) THEN RETURN NULL; END IF;
+          END IF;
+          INSERT INTO catalog_revisions(name) VALUES (TG_TABLE_NAME)
+          ON CONFLICT (name) DO UPDATE SET revision = gen_random_uuid();
+          RETURN NULL;
+        END $fn$
+    """)
+    # Evict snapshots cached before the trigger covered all ranking inputs.
+    op.execute("UPDATE catalog_revisions SET revision=gen_random_uuid() WHERE name='topics'")
+
+
+def upgrade():
+    replace_revision_function(IDENTITY_COLUMNS + ", description, ai_description")
+
+
+def downgrade():
+    replace_revision_function(IDENTITY_COLUMNS)

packages/core/OBSERVABILITY.md

@@ -180,6 +180,23 @@ waiting on `ClientRead`. Candidate selection reuses immutable normalized identit
 and description vocabulary; it still recomputes eligibility against the current
 catalog. Catalog edits and ordering changes cannot reuse stale retrieval indexes.
 
+Article automation reuses one catalog snapshot and its retrieval indexes during a
+batch. PostgreSQL revisions are checked before every reuse and again under the
+publication locks. Ranking runs after the preparation transaction closes; the
+locked checks reuse at most eight article candidate results. Content or catalog
+changes between preparation and application leave the article due for the next
+tick. Source eligibility, editorial revisions and job state are checked under
+their existing locks. Publication evaluations are reused only within that same
+locked transaction.
+
+Migration `0022` makes topic description and AI description edits advance the
+catalog revision, because both supply ranking vocabulary. Apply it before running
+the updated services. The migration and its downgrade invalidate existing topic
+snapshots; downgrading restores the previous description-invalidation behavior.
+Compare `devfeed_background_duration_seconds` for `scheduler.tick`, scheduler CPU,
+and topic-lock wait latency over matching windows after rollout. Local catalog
+benchmarks do not establish production tick latency.
+
 The discovery profile covers feed facets with language, content-type and source
 filters. Broad browsing uses early-exit probes; selective topic, tag and text
 searches resolve their matching articles once. It checks two SQL queries and bounded article visits on a larger fixture:

packages/core/src/analysis.py

@@ -6,6 +6,9 @@
 import re
 import unicodedata
 import uuid
+from collections import OrderedDict
+from contextlib import contextmanager
+from contextvars import ContextVar
 from datetime import timedelta
 from functools import lru_cache
 from typing import Literal
@@ -274,18 +277,23 @@ def candidate_description_index(descriptions: tuple[tuple[str | None, str | None
     return document_frequency, postings
 
 
-def candidate_scores(items: list[dict], snapshot: dict) -> list[float]:
-    """Rank exact catalog matches and distinctive words in topic descriptions."""
+def _candidate_indexes(items):
     terms = tuple(
         candidate_terms(
             (item["name"], item["slug"], *item.get("aliases", [])),
             tuple(item.get("keywords", [])),
         )
         for item in items
     )
-    root = candidate_index(terms)
-    scores = [0.0] * len(items)
-    for text, weight in candidate_evidence(snapshot):
+    descriptions = tuple((item.get("description"), item.get("ai_description")) for item in items)
+    frequency, postings = candidate_description_index(descriptions)
+    return candidate_index(terms), frequency, postings
+
+
+def _candidate_scores(root, document_frequency, postings, size, snapshot):
+    scores = [0.0] * size
+    evidence = candidate_evidence(snapshot)
+    for text, weight in evidence:
         matches = set(root.get(None, ())) if "  " in text else set()
         words = text.strip().split()
         for start in range(len(words)):
@@ -303,20 +311,68 @@ def candidate_scores(items: list[dict], snapshot: dict) -> list[float]:
     # security practice, or product strategy. Topic descriptions give retrieval
     # additional vocabulary while inverse document frequency discounts generic
     # words shared by most of the catalog.
-    descriptions = tuple((item.get("description"), item.get("ai_description")) for item in items)
-    document_frequency, postings = candidate_description_index(descriptions)
-    catalog_size = len(items)
-    for text, weight in candidate_evidence(snapshot):
+    for text, weight in evidence:
         for word in set(re.findall(r"[\w+#]{3,}", text)):
             frequency = document_frequency.get(word, 0)
             if not frequency:
                 continue
-            idf = math.log1p((catalog_size + 1) / frequency)
+            idf = math.log1p((size + 1) / frequency)
             for index in postings[word]:
                 scores[index] += weight * idf
     return scores
 
 
+def candidate_scores(items: list[dict], snapshot: dict) -> list[float]:
+    """Rank exact catalog matches and distinctive words in topic descriptions."""
+    root, frequency, postings = _candidate_indexes(items)
+    return _candidate_scores(root, frequency, postings, len(items), snapshot)
+
+
+class _PreparedCatalog:
+    """Read-only ranking inputs and bounded results for one tick's current snapshot."""
+
+    def __init__(self, taxonomy):
+        self.taxonomy = taxonomy
+        self.entries = [(field, item) for field in ("topics", "tags") for item in taxonomy[field]]
+        self.sort_keys = [
+            (item["name"].casefold(), item["id"], field) for field, item in self.entries
+        ]
+        self.indexes = _candidate_indexes([item for _, item in self.entries])
+        self.prompt_items: dict[int, tuple[dict, int]] = {}
+        self.identities: dict[str, dict] = {}
+        self.results: OrderedDict[tuple, dict] = OrderedDict()
+
+    def current(self, field):
+        if field not in self.identities:
+            self.identities[field] = {item["id"]: item for item in self.taxonomy[field]}
+        return self.identities[field]
+
+
+_ranking: ContextVar[list[_PreparedCatalog] | None] = ContextVar("catalog_ranking", default=None)
+
+
+@contextmanager
+def reuse_candidates():
+    """Reuse ranking only within an explicit scope, never a cached publication decision."""
+    token = _ranking.set([])
+    try:
+        yield
+    finally:
+        _ranking.reset(token)
+
+
+def _prepared_catalog(taxonomy):
+    values = _ranking.get()
+    if values and values[0].taxonomy is taxonomy:
+        return values[0]
+    prepared = _PreparedCatalog(taxonomy)
+    if values is not None:
+        for field in ("topics", "tags"):
+            prepared.current(field)
+        values[:] = [prepared]
+    return prepared
+
+
 def analysis_candidates(taxonomy: dict, snapshot: dict) -> dict:
     """Rank catalog entries against article evidence and bound the inference input.
 
@@ -325,28 +381,48 @@ def analysis_candidates(taxonomy: dict, snapshot: dict) -> dict:
     """
 
     settings = get_settings()
-    entries = [(field, item) for field in ("topics", "tags") for item in taxonomy[field]]
-    scores = candidate_scores([item for _, item in entries], snapshot)
-    ranked = [
-        (score, item["name"].casefold(), item["id"], field, item)
-        for (field, item), score in zip(entries, scores, strict=True)
-    ]
+    prepared = _prepared_catalog(taxonomy)
+    key = (
+        snapshot_hash(snapshot),
+        settings.analysis_max_candidates,
+        settings.analysis_fallback_candidates,
+    )
+    if key in prepared.results:
+        return prepared.results[key]
+    root, frequency, postings = prepared.indexes
+    scores = _candidate_scores(root, frequency, postings, len(prepared.entries), snapshot)
     result: dict[str, list] = {"topics": [], "tags": []}
     # Codex allows 250 KB. Reserve space for the article, instructions and JSON
     # separators rather than assuming a count alone bounds long alias lists.
     remaining = 240_000 - len(analysis_prompt(snapshot, result).encode())
     fallback = {"topics": 0, "tags": 0}
-    for score, _, _, field, item in sorted(ranked, key=lambda row: (-row[0], *row[1:4])):
+    for index in sorted(range(len(scores)), key=lambda i: (-scores[i], *prepared.sort_keys[i])):
+        score = scores[index]
+        if score == 0 and all(
+            len(result[field]) >= settings.analysis_max_candidates
+            or fallback[field] >= settings.analysis_fallback_candidates
+            for field in result
+        ):
+            break
+        field, item = prepared.entries[index]
         if len(result[field]) >= settings.analysis_max_candidates:
             continue
         if score == 0 and fallback[field] >= settings.analysis_fallback_candidates:
             continue
-        prompt_item = _prompt_catalog_item(item)
-        size = len(json.dumps(prompt_item, ensure_ascii=False).encode()) + 2
+        if index not in prepared.prompt_items:
+            prompt_item = _prompt_catalog_item(item)
+            prepared.prompt_items[index] = (
+                prompt_item,
+                len(json.dumps(prompt_item, ensure_ascii=False).encode()) + 2,
+            )
+        prompt_item, size = prepared.prompt_items[index]
         if size <= remaining:
             result[field].append(prompt_item)
             remaining -= size
             fallback[field] += score == 0
+    prepared.results[key] = result
+    if len(prepared.results) > 8:
+        prepared.results.popitem(last=False)
     return result
 
 
@@ -388,7 +464,11 @@ def analysis_catalog_current(job, taxonomy: dict, snapshot: dict) -> bool:
         }
         if relevant_old != relevant_new:
             return False
-        current = {item["id"]: item for item in taxonomy[field]}
+        current = (
+            _prepared_catalog(taxonomy).current(field)
+            if _ranking.get() is not None
+            else {item["id"]: item for item in taxonomy[field]}
+        )
         for selection in (job.result or {}).get(field, []):
             key = selection.get(identifier)
             if key not in old or key not in current:

packages/core/src/article_automation.py

@@ -4,6 +4,7 @@
 from datetime import timedelta
 
 from sqlalchemy import func, select
+from sqlalchemy.orm import lazyload
 
 from devfeed_core.ai_content import eligible_article, eligible_articles
 from devfeed_core.analysis import (
@@ -14,6 +15,7 @@
     candidate_score,
     catalog,
     request_analysis,
+    reuse_candidates,
     snapshot_hash,
     source_snapshot,
 )
@@ -22,6 +24,7 @@
     approved_sources,
     request_article_enrichment,
 )
+from devfeed_core.catalog_cache import reuse_snapshots, snapshot_current
 from devfeed_core.catalog_cache import snapshot as catalog_snapshot
 from devfeed_core.config import get_settings
 from devfeed_core.editorial import EditorialDecision, decide_article, meaningful_text
@@ -210,6 +213,45 @@ def reject_known_paywalled_articles(factory) -> int:
 
 
 def schedule_article_automation(factory) -> dict[str, int]:
+    with reuse_snapshots(), reuse_candidates():
+        return _schedule_article_automation(factory)
+
+
+def _prepare_article_candidates(factory, identifier):
+    from devfeed_core.analysis import analysis_candidates
+
+    with factory() as session:
+        article = session.scalar(
+            select(Article).where(Article.id == identifier).options(lazyload("*"))
+        )
+        if article is None:
+            return None
+        job = session.scalar(
+            select(ArticleAnalysisJob)
+            .where(ArticleAnalysisJob.article_id == identifier)
+            .order_by(ArticleAnalysisJob.created_at.desc(), ArticleAnalysisJob.id.desc())
+            .limit(1)
+        )
+        enrichment = session.scalar(
+            select(ArticleEnrichmentJob)
+            .where(ArticleEnrichmentJob.article_id == identifier)
+            .order_by(ArticleEnrichmentJob.created_at.desc(), ArticleEnrichmentJob.id.desc())
+            .limit(1)
+        )
+        if (
+            (job is not None and job.status in {"queued", "running"})
+            or (enrichment is not None and enrichment.status in {"queued", "running"})
+            or (job is None and enrichment is None)
+        ):
+            return None
+        snapshot = source_snapshot(article, session.get(ArticleContent, identifier))
+        taxonomy = catalog(session)
+    # No row/catalog locks or open transaction during full-catalog ranking.
+    analysis_candidates(taxonomy, snapshot)
+    return taxonomy, snapshot
+
+
+def _schedule_article_automation(factory) -> dict[str, int]:
     counts = {
         "articles_checked": 0,
         "articles_published": 0,
@@ -244,6 +286,7 @@ def schedule_article_automation(factory) -> dict[str, int]:
             .limit(get_settings().automation_batch_size)
         ).all()
     for identifier in identifiers:
+        prepared = _prepare_article_candidates(factory, identifier)
         with factory.begin() as session:
             # Workers acquire job -> source -> article -> taxonomy. Use the same order.
             job = session.scalar(
@@ -321,8 +364,18 @@ def schedule_article_automation(factory) -> dict[str, int]:
             if not negative:
                 counts["source_topics_proposed"] += propose_source_topics(session, article)
             lock_topics(session, read=True)
-            taxonomy = catalog(session)
             snapshot = source_snapshot(article, session.get(ArticleContent, identifier))
+            if (
+                prepared is None
+                or snapshot != prepared[1]
+                or not snapshot_current(session, ("topics", "tags"), prepared[0])
+            ):
+                # Catalog/input changes must be ranked again outside these locks.
+                # Leave the article due for the next tick rather than using stale data.
+                article.automation_next_check_at = now
+                continue
+            taxonomy = prepared[0]
+            evaluation = None
             if negative:
                 decision = apply_publication_policy(session, article, job, taxonomy=taxonomy)
                 if decision["status"] == "rejected":
@@ -381,9 +434,12 @@ def schedule_article_automation(factory) -> dict[str, int]:
             elif job.result.get("topic_match_status") == "no_primary_topic":
                 reasons = ["no_supported_primary_topic"]
             else:
-                decision = evaluate_publication(session, article, job, taxonomy=taxonomy)
+                evaluation = evaluate_publication(session, article, job, taxonomy=taxonomy)
+                decision = evaluation
                 if decision["status"] == "would_publish":
-                    apply_publication_policy(session, article, job, taxonomy=taxonomy)
+                    apply_publication_policy(
+                        session, article, job, taxonomy=taxonomy, evaluation=evaluation
+                    )
                     counts["articles_published"] += 1
                     continue
                 reasons = decision["reasons"]
@@ -405,6 +461,7 @@ def schedule_article_automation(factory) -> dict[str, int]:
                 article,
                 job,
                 taxonomy=taxonomy,
+                evaluation=evaluation,
             )
             if decision["status"] == "rejected":
                 counts["articles_rejected"] += 1