Release catalog locks before worker ranking

CHANGELOG.md

@@ -7,6 +7,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 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.
 
 ## 0.0.50 — 2026-10-03

apps/aggregator/src/analysis_tasks.py

@@ -17,12 +17,14 @@
     fail_analysis,
     refresh_superseded_analysis,
     require_primary_topic,
+    reuse_candidates,
     snapshot_hash,
     source_snapshot,
     validate_evidence,
 )
 from devfeed_core.analysis_wire import compact_request, restore_identities
 from devfeed_core.article_jobs import approved_sources
+from devfeed_core.catalog_cache import reuse_snapshots, snapshot_current
 from devfeed_core.config import get_settings
 from devfeed_core.db import session_factory
 from devfeed_core.inference_validation import feedback_prompt, validation_feedback
@@ -33,6 +35,7 @@
 from devfeed_core.models import Article, ArticleAnalysisJob, ArticleContent, utcnow
 from pydantic import ValidationError
 from sqlalchemy import select
+from sqlalchemy.orm import lazyload
 
 from devfeed_aggregator.analysis_telemetry import record_attempt
 from devfeed_aggregator.codex_client import AnalysisError, CodexClient
@@ -41,6 +44,47 @@
 logger = logging.getLogger(__name__)
 
 
+class _ApplicationInputsChanged(Exception):
+    """Release publication locks before preparing changed catalog/content inputs."""
+
+
+def _prepare_application(factory, article_id, snapshot):
+    with factory() as session:
+        current_catalog = catalog(session)
+        article = session.scalar(
+            select(Article).where(Article.id == article_id).options(lazyload("*"))
+        )
+        current = (
+            source_snapshot(article, session.get(ArticleContent, article_id))
+            if article is not None
+            else None
+        )
+    # Warm both validity checks and any replacement request with no open transaction.
+    analysis_candidates(current_catalog, snapshot)
+    if current is not None and current != snapshot:
+        analysis_candidates(current_catalog, current)
+    return current_catalog, current
+
+
+def _apply_prepared(factory, article_id, snapshot, operation):
+    from devfeed_core.transaction_retry import run_transaction
+
+    for attempt in range(3):
+        # Do not retain full retrieval indexes during the external inference wait.
+        with reuse_snapshots(), reuse_candidates():
+            taxonomy, current = _prepare_application(factory, article_id, snapshot)
+            try:
+                return run_transaction(
+                    factory,
+                    lambda session, taxonomy=taxonomy, current=current: operation(
+                        session, taxonomy, current
+                    ),
+                )
+            except _ApplicationInputsChanged:
+                if attempt == 2:
+                    raise  # Existing durable dependency retries handle continued churn.
+
+
 def analyze_article(job_id: str) -> None:
     with job_log_context("analysis", job_id):
         try:
@@ -108,7 +152,9 @@ def _analyze_claimed(settings, factory, identifier, token, snapshot, article_id)
     try:
         # No transaction remains open while waiting for Codex.
         with factory() as session:
-            taxonomy = analysis_candidates(catalog(session), snapshot)
+            inference_catalog = catalog(session)
+        taxonomy = analysis_candidates(inference_catalog, snapshot)
+        del inference_catalog
         with factory.begin() as session:
             job = owned_job(session, ArticleAnalysisJob, identifier, token)
             if job is None:
@@ -167,7 +213,7 @@ def _analyze_claimed(settings, factory, identifier, token, snapshot, article_id)
                 if article is not None and source_ids and article.review_status != "rejected":
                     propose_source_topics(session, article)
 
-        def apply_result(session):
+        def apply_result(session, current_catalog, prepared_snapshot):
             job = owned_job(session, ArticleAnalysisJob, identifier, token)
             if job is None:
                 logger.warning("article_analysis_lease_lost")
@@ -193,7 +239,12 @@ def apply_result(session):
                     from devfeed_core.topics import lock_topics
 
                     lock_topics(session, read=True)
-                    current_catalog = catalog(session)
+                    if source_snapshot(
+                        article, session.get(ArticleContent, article_id)
+                    ) != prepared_snapshot or not snapshot_current(
+                        session, ("topics", "tags"), current_catalog
+                    ):
+                        raise _ApplicationInputsChanged()
                     if not analysis_catalog_current(job, current_catalog, snapshot):
                         finish_job(job, "superseded", utcnow())
                     else:
@@ -205,12 +256,10 @@ def apply_result(session):
                             apply_publication_policy(
                                 session, article, job, taxonomy=current_catalog
                             )
-                    refresh_superseded_analysis(session, article, job)
+                    refresh_superseded_analysis(session, article, job, taxonomy=current_catalog)
             return job.outcome
 
-        from devfeed_core.transaction_retry import run_transaction
-
-        outcome = run_transaction(factory, apply_result)
+        outcome = _apply_prepared(factory, article_id, snapshot, apply_result)
         logger.info("article_analysis_completed", extra={"outcome": outcome})
     except Exception as exc:
         feedback = validation_feedback(exc)

packages/core/OBSERVABILITY.md

@@ -197,6 +197,16 @@ Compare `devfeed_background_duration_seconds` for `scheduler.tick`, scheduler CP
 and topic-lock wait latency over matching windows after rollout. Local catalog
 benchmarks do not establish production tick latency.
 
+Analysis workers also prepare current catalog candidates and identity membership
+before acquiring source, article and topic locks. A revision or source-content
+change after preparation rolls back the application transaction and prepares the
+new inputs outside the locks. Up to three preparations can reuse the same
+inference result; continued churn uses the existing durable dependency retry
+policy. Reanalysis of superseded content receives that validated catalog rather
+than loading and ranking the entire catalog inside the application transaction.
+Candidate scopes begin after inference, so they do not retain a full catalog
+through the external model wait. Evidence and publication guards remain required.
+
 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

@@ -568,7 +568,11 @@ def validate_evidence(result: Classifications, snapshot: dict, taxonomy: dict) -
         ("topics", result.topics),
         ("tags", result.tags),
     ):
-        valid = {item["id"] for item in taxonomy[field]}
+        valid = (
+            _prepared_catalog(taxonomy).current(field)
+            if _ranking.get() is not None
+            else {item["id"] for item in taxonomy[field]}
+        )
         for item in selections:
             identifier = item.topic_id if isinstance(item, TopicSelection) else item.id
             if str(identifier) not in valid:
@@ -716,7 +720,9 @@ def request_analysis(
     return job
 
 
-def refresh_superseded_analysis(session: Session, article: Article, job: ArticleAnalysisJob):
+def refresh_superseded_analysis(
+    session: Session, article: Article, job: ArticleAnalysisJob, *, taxonomy=None
+):
     """Coalescing an active delivery must not strand newer source evidence."""
     if job.outcome != "superseded" or article.review_status != "pending":
         return None
@@ -726,10 +732,11 @@ def refresh_superseded_analysis(session: Session, article: Article, job: Article
         # not snapshot the candidate hash, so catalog changes cannot be inferred.
         if job.catalog_hash is None or article.editorial_revision != job.editorial_revision:
             return None
-        if analysis_catalog_current(job, catalog(session), current):
+        current_catalog = catalog(session) if taxonomy is None else taxonomy
+        if analysis_catalog_current(job, current_catalog, current):
             return None
     session.flush()  # Release the active-job uniqueness slot before enqueueing.
-    return request_analysis(session, article.id, automatic=True)
+    return request_analysis(session, article.id, automatic=True, taxonomy=taxonomy)
 
 
 def backfill_analyses(

tests/test_analysis.py

@@ -128,6 +128,16 @@ def test_evidence_must_be_from_input_and_ids_from_dynamic_catalog():
         analysis.validate_evidence(invalid, SNAPSHOT, CATALOG)
 
 
+def test_prepared_catalog_membership_rejects_a_removed_identity():
+    with analysis.reuse_candidates():
+        analysis.analysis_candidates(CATALOG, SNAPSHOT)
+        analysis.validate_evidence(result(), SNAPSHOT, CATALOG)
+        changed = {"topics": [], "tags": []}
+        analysis.analysis_candidates(changed, SNAPSHOT)
+        with pytest.raises(ValueError, match="unknown"):
+            analysis.validate_evidence(result(), SNAPSHOT, changed)
+
+
 @pytest.mark.parametrize(
     "values",
     [
@@ -382,7 +392,7 @@ def test_new_content_coalesced_into_active_analysis_gets_a_followup(monkeypatch)
     monkeypatch.setattr(analysis, "request_analysis", lambda *a, **kw: calls.append((a, kw)))
     finish_job(job, "superseded", utcnow())
     analysis.refresh_superseded_analysis(db, current, job)
-    assert len(calls) == 1 and calls[0][1] == {"automatic": True}
+    assert len(calls) == 1 and calls[0][1] == {"automatic": True, "taxonomy": None}
 
 
 @pytest.mark.parametrize("status", ["approved", "rejected", "pending"])