diff --git a/CHANGELOG.md b/CHANGELOG.md index 03000021..2b763293 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/migrations/versions/0022_catalog_ranking_revision.py b/migrations/versions/0022_catalog_ranking_revision.py new file mode 100644 index 00000000..4f5a65e9 --- /dev/null +++ b/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) diff --git a/packages/core/OBSERVABILITY.md b/packages/core/OBSERVABILITY.md index 452daad3..b0aaf2d9 100644 --- a/packages/core/OBSERVABILITY.md +++ b/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: diff --git a/packages/core/src/analysis.py b/packages/core/src/analysis.py index 9fa621d7..c7d214db 100644 --- a/packages/core/src/analysis.py +++ b/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,8 +277,7 @@ 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", [])), @@ -283,9 +285,15 @@ def candidate_scores(items: list[dict], snapshot: dict) -> list[float]: ) 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: diff --git a/packages/core/src/article_automation.py b/packages/core/src/article_automation.py index 1c4fbf98..277d1c57 100644 --- a/packages/core/src/article_automation.py +++ b/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 diff --git a/packages/core/src/catalog_cache.py b/packages/core/src/catalog_cache.py index e5e5d5d9..d39a0b91 100644 --- a/packages/core/src/catalog_cache.py +++ b/packages/core/src/catalog_cache.py @@ -1,7 +1,8 @@ """Immutable snapshots keyed by transactional PostgreSQL revisions, never a TTL decision.""" import json -from contextlib import suppress +from contextlib import contextmanager, suppress +from contextvars import ContextVar from sqlalchemy import select @@ -9,6 +10,39 @@ from devfeed_core.config import get_settings from devfeed_core.models import CatalogRevision +_snapshots: ContextVar[dict | None] = ContextVar("catalog_snapshots", default=None) + + +@contextmanager +def reuse_snapshots(): + """Reuse one snapshot per projection during a tick, always checking PostgreSQL.""" + token = _snapshots.set({}) + try: + yield + finally: + _snapshots.reset(token) + + +def _revisions(session, names): + return tuple( + (name, str(revision)) + for name, revision in session.execute( + select(CatalogRevision.name, CatalogRevision.revision) + .where(CatalogRevision.name.in_(names)) + .order_by(CatalogRevision.name) + ) + ) + + +def snapshot_current(session, names, value) -> bool: + """Validate prepared data under the caller's locks without loading new catalog rows.""" + values = _snapshots.get() + if values is None: + return False + session.flush() + entry = values.get((session.get_bind(), tuple(sorted(names)))) + return entry is not None and entry[1] is value and _revisions(session, names) == entry[0] + def snapshot(session, names, loader): """A miss/failure reads PostgreSQL; a concurrent write prevents cache publication. @@ -18,24 +52,25 @@ def snapshot(session, names, loader): Read before and after loading so READ COMMITTED cannot cache mixed revisions. Existing publication locks still protect the final check/application boundary. """ - if not get_settings().cache_enabled: + values = _snapshots.get() + enabled = get_settings().cache_enabled + if not enabled and values is None: record_cache_read("catalog", "bypass", "disabled") return loader() session.flush() - - def revisions(): - return tuple( - (name, str(revision)) - for name, revision in session.execute( - select(CatalogRevision.name, CatalogRevision.revision) - .where(CatalogRevision.name.in_(names)) - .order_by(CatalogRevision.name) - ) - ) - - before = revisions() + before = _revisions(session, names) if len(before) != len(names): return loader() + slot = (session.get_bind(), tuple(sorted(names))) + if values is not None and slot in values and values[slot][0] == before: + record_cache_read("catalog", "hit") + return values[slot][1] + if not enabled: + record_cache_read("catalog", "bypass", "disabled") + value = loader() + if values is not None and _revisions(session, names) == before: + values[slot] = (before, value) + return value cache = get_cache() # Reuse the cache's bounded Redis failure policy, independently of public # response invalidation. Revisions are authoritative; expiry only bounds storage. @@ -50,6 +85,8 @@ def revisions(): entry = json.loads(body) if entry["revision"] == [list(item) for item in before]: record_cache_read("catalog", "hit") + if values is not None: + values[slot] = (before, entry["value"]) return entry["value"] except CacheUnavailable: outcome, reason = "bypass", "cache_unavailable" @@ -57,7 +94,9 @@ def revisions(): reason = "invalid_entry" record_cache_read("catalog", outcome, reason) value = loader() - if revisions() == before: + if _revisions(session, names) == before: + if values is not None: + values[slot] = (before, value) with suppress(CacheUnavailable): cache._run( lambda: cache.redis.set( diff --git a/packages/core/src/publication_policy.py b/packages/core/src/publication_policy.py index 4e931e7b..ed9a8c25 100644 --- a/packages/core/src/publication_policy.py +++ b/packages/core/src/publication_policy.py @@ -106,9 +106,16 @@ def evaluate_publication(session, article, job, *, taxonomy=None) -> dict: } -def apply_publication_policy(session, article, job, *, taxonomy=None) -> dict: - """Caller owns source, article and taxonomy locks, in that order.""" - decision = evaluate_publication(session, article, job, taxonomy=taxonomy) +def apply_publication_policy(session, article, job, *, taxonomy=None, evaluation=None) -> dict: + """Caller owns source, article and taxonomy locks, in that order. + + An evaluation may be reused only from this same locked transaction. + """ + decision = ( + evaluate_publication(session, article, job, taxonomy=taxonomy) + if evaluation is None + else evaluation + ) if ( "paywalled_content" in decision["reasons"] and article.review_status == "pending" diff --git a/packages/core/src/version.py b/packages/core/src/version.py index 1465d887..6fab3748 100644 --- a/packages/core/src/version.py +++ b/packages/core/src/version.py @@ -6,7 +6,7 @@ # Schema revisions are not application versions. Advance this when adding a # migration required by this application; never change an existing migration. -SCHEMA_REVISION = "0021" +SCHEMA_REVISION = "0022" # Scheduled like-based recommendations require migration 0015. BACKWARD_COMPATIBLE_SCHEMA_REVISIONS: frozenset[str] = frozenset() diff --git a/tests/test_analysis_catalog.py b/tests/test_analysis_catalog.py index e5b0be5b..59faca30 100644 --- a/tests/test_analysis_catalog.py +++ b/tests/test_analysis_catalog.py @@ -283,3 +283,53 @@ def test_production_sized_catalog_reuses_normalized_identities(): misses = analysis.candidate_terms.cache_info().misses assert analysis.candidate_scores(items, snapshot) == before assert analysis.candidate_terms.cache_info().misses == misses + + +def test_tick_ranking_reuses_inputs_and_invalidates_changed_content_catalog_and_limits(monkeypatch): + from devfeed_core.config import get_settings + + taxonomy = {"topics": [entry("Angular"), entry("React"), entry("Other")], "tags": []} + snapshot = {"title": "Angular routing"} + expected = analysis.analysis_candidates(taxonomy, snapshot) + scores = analysis._candidate_scores + calls = [] + + def count(*args): + calls.append(True) + return scores(*args) + + monkeypatch.setattr(analysis, "_candidate_scores", count) + with analysis.reuse_candidates(): + assert analysis.analysis_candidates(taxonomy, snapshot) == expected + assert analysis.analysis_candidates(taxonomy, dict(snapshot)) == expected + assert len(calls) == 1 + changed = {"title": "React routing"} + assert analysis.analysis_candidates(taxonomy, changed)["topics"][0]["name"] == "React" + monkeypatch.setattr(get_settings(), "analysis_fallback_candidates", 0) + assert len(analysis.analysis_candidates(taxonomy, changed)["topics"]) == 1 + updated = {"topics": [entry("React", aliases=["Angular"])], "tags": []} + assert analysis.analysis_candidates(updated, snapshot)["topics"] == updated["topics"] + assert len(calls) == 4 + analysis.analysis_candidates(taxonomy, snapshot) + assert len(calls) == 5 # Tick results do not leak into another scope. + + +def test_tick_ranking_releases_old_article_results(monkeypatch): + taxonomy = {"topics": [entry("Angular")], "tags": []} + scores = analysis._candidate_scores + calls = [] + + def count(*args): + calls.append(True) + return scores(*args) + + monkeypatch.setattr(analysis, "_candidate_scores", count) + snapshots = [{"title": f"Angular routing {index}"} for index in range(10)] + with analysis.reuse_candidates(): + original = analysis.analysis_candidates(taxonomy, snapshots[0]) + for snapshot in snapshots[1:]: + analysis.analysis_candidates(taxonomy, snapshot) + analysis.analysis_candidates(taxonomy, snapshots[-1]) + assert len(calls) == 10 # Recent results stay available to locked checks. + assert analysis.analysis_candidates(taxonomy, snapshots[0]) == original + assert len(calls) == 11 # Older results are discarded without changing selection. diff --git a/tests/test_catalog_snapshots.py b/tests/test_catalog_snapshots.py index 8579d9c9..f289c680 100644 --- a/tests/test_catalog_snapshots.py +++ b/tests/test_catalog_snapshots.py @@ -196,3 +196,55 @@ def test_rapid_edits_keep_one_bounded_cache_slot(cached): assert analysis.catalog(session)["tags"][0]["name"] == f"Edit {i}" cache = get_cache() assert len(list(cache.redis.scan_iter(f"{cache.namespace}:catalog:*"))) == 1 + + +@pytest.mark.parametrize("redis_enabled", [True, False]) +def test_tick_snapshot_reuse_rechecks_revisions_and_handles_rollback( + database, monkeypatch, redis_enabled +): + from devfeed_core.catalog_cache import reuse_snapshots, snapshot_current + + monkeypatch.setattr(get_settings(), "cache_enabled", redis_enabled) + with database.begin() as session: + session.add(Tag(name="Original", slug="original")) + with reuse_snapshots(): + with database() as session: + original = analysis.catalog(session) + with database() as session: + assert analysis.catalog(session) is original + assert snapshot_current(session, ("topics", "tags"), original) + with database() as editor: + editor.execute(text("UPDATE tags SET name='Uncommitted'")) + assert analysis.catalog(editor)["tags"][0]["name"] == "Uncommitted" + with database() as reader: + assert analysis.catalog(reader)["tags"][0]["name"] == "Original" + editor.rollback() + with database.begin() as editor: + editor.execute(text("UPDATE tags SET name='Committed'")) + with database() as session: + assert not snapshot_current(session, ("topics", "tags"), original) + current = analysis.catalog(session) + assert current["tags"][0]["name"] == "Committed" + assert analysis.catalog(session) is current + + +@pytest.mark.parametrize("column", ["description", "ai_description"]) +def test_description_vocabulary_edits_invalidate_cached_catalog(cached, monkeypatch, column): + from devfeed_core.catalog_cache import reuse_snapshots, snapshot_current + from devfeed_core.models import Topic + + monkeypatch.setattr(get_settings(), "analysis_fallback_candidates", 0) + with cached.begin() as session: + session.add(Topic(name="Career", slug="career", kind="technology", status="active")) + with reuse_snapshots(), analysis.reuse_candidates(): + with cached() as session: + before = analysis.catalog(session) + evidence = {"title": "Interviewing developers"} + assert analysis.analysis_candidates(before, evidence)["topics"] == [] + with cached.begin() as editor: + editor.execute(text(f"UPDATE topics SET {column}='Interviewing software developers'")) + with cached() as session: + assert not snapshot_current(session, ("topics", "tags"), before) + after = analysis.catalog(session) + assert after["topics"][0][column] == "Interviewing software developers" + assert analysis.analysis_candidates(after, evidence)["topics"][0]["name"] == "Career" diff --git a/tests/test_scheduler_catalog_reuse.py b/tests/test_scheduler_catalog_reuse.py new file mode 100644 index 00000000..76fdaca0 --- /dev/null +++ b/tests/test_scheduler_catalog_reuse.py @@ -0,0 +1,122 @@ +"""Scheduler publication keeps authoritative guards without ranking while holding locks.""" + +import pytest +from devfeed_core import analysis, article_automation, publication_policy +from devfeed_core.config import get_settings +from devfeed_core.db import get_engine +from devfeed_core.models import Article, ArticleAnalysisJob, utcnow +from sqlalchemy import event, select, text +from sqlalchemy.orm import Session +from test_automation_integration import ready, seed +from test_full_app_automation import full + +pytestmark = pytest.mark.integration + + +def test_publication_batch_ranks_once_per_article_before_locks(database, monkeypatch): + full(monkeypatch, AUTOMATION_BATCH_SIZE=5) + monkeypatch.setattr(get_settings(), "cache_enabled", False) + with database.begin() as session: + articles = [seed(session, name=f"Framework{index}") for index in range(5)] + for _, article, topic in articles: + ready(session, article, topic) + locked = [False] + connections = [0] + ranks, evaluations, loads = [], [], [] + rank = analysis._candidate_scores + evaluate = publication_policy.evaluate_publication + load = analysis._read_catalog + + def capture(conn, cursor, statement, parameters, context, executemany): + if "FOR UPDATE" in statement or "LOCK TABLE topics" in statement: + locked[0] = True + + def released(session, transaction): + if transaction.parent is None: + locked[0] = False + + def checkout(*args): + connections[0] += 1 + + def checkin(*args): + connections[0] -= 1 + + def ranking(*args): + assert not locked[0], "Full-catalog ranking must precede row and taxonomy locks" + assert connections[0] == 0, "Ranking must release the read transaction" + ranks.append(True) + return rank(*args) + + def evaluation(*args, **kwargs): + evaluations.append(True) + return evaluate(*args, **kwargs) + + def loading(*args): + loads.append(True) + return load(*args) + + monkeypatch.setattr(analysis, "_candidate_scores", ranking) + monkeypatch.setattr(analysis, "_read_catalog", loading) + monkeypatch.setattr(publication_policy, "evaluate_publication", evaluation) + monkeypatch.setattr(article_automation, "evaluate_publication", evaluation) + event.listen(get_engine(), "before_cursor_execute", capture) + event.listen(get_engine(), "checkout", checkout) + event.listen(get_engine(), "checkin", checkin) + event.listen(Session, "after_transaction_end", released) + try: + counts = article_automation.schedule_article_automation(database) + finally: + event.remove(get_engine(), "before_cursor_execute", capture) + event.remove(get_engine(), "checkout", checkout) + event.remove(get_engine(), "checkin", checkin) + event.remove(Session, "after_transaction_end", released) + assert counts["articles_published"] == 5 + assert len(ranks) == len(evaluations) == 5 + assert len(loads) == 1 + + +@pytest.mark.parametrize("changed", ["catalog", "content", "editorial", "source"]) +def test_change_between_preparation_and_locking_never_applies_stale_result( + database, monkeypatch, changed +): + full(monkeypatch) + with database.begin() as session: + source, article, topic = seed(session) + ready(session, article, topic) + identifier, source_id, topic_id = article.id, source.id, topic.id + prepare = article_automation._prepare_article_candidates + + def racing_prepare(*args): + result = prepare(*args) + with database.begin() as editor: + if changed == "catalog": + editor.execute( + text("UPDATE topics SET name='Changed' WHERE id=:id"), {"id": topic_id} + ) + elif changed == "content": + editor.execute( + text("UPDATE articles SET summary='Changed article text' WHERE id=:id"), + {"id": identifier}, + ) + elif changed == "editorial": + editor.execute( + text( + "UPDATE articles SET editorial_revision=editorial_revision+1 WHERE id=:id" + ), + {"id": identifier}, + ) + else: + editor.execute( + text("UPDATE sources SET enabled=false WHERE id=:id"), {"id": source_id} + ) + return result + + monkeypatch.setattr(article_automation, "_prepare_article_candidates", racing_prepare) + counts = article_automation.schedule_article_automation(database) + assert counts["articles_published"] == counts["articles_rejected"] == 0 + with database() as session: + article = session.get(Article, identifier) + assert article.review_status == "pending" and article.publication_status == "unpublished" + if changed in {"catalog", "content"}: + assert article.automation_next_check_at <= utcnow() + assert len(session.scalars(select(ArticleAnalysisJob)).all()) == 1