From 83e04074910891c818f3a9a75e462f17f5c75089 Mon Sep 17 00:00:00 2001 From: Justin Visser Date: Sun, 1 Mar 2026 23:11:52 +0100 Subject: [PATCH 01/23] " " --- app/services/scholars/application.py | 8 +++++++- tests/integration/test_run_lifecycle_consistency.py | 5 +++-- 2 files changed, 10 insertions(+), 3 deletions(-) diff --git a/app/services/scholars/application.py b/app/services/scholars/application.py index 894c371..028e3fc 100644 --- a/app/services/scholars/application.py +++ b/app/services/scholars/application.py @@ -10,7 +10,11 @@ from app.db.models import ScholarProfile from app.services.scholar.parser import ScholarParserError, parse_profile_page from app.services.scholar.source import ScholarSource from app.services.scholars.author_search import search_author_candidates -from app.services.scholars.constants import ALLOWED_IMAGE_UPLOAD_CONTENT_TYPES +from app.services.scholars.constants import ( + ALLOWED_IMAGE_UPLOAD_CONTENT_TYPES, + SEARCH_COOLDOWN_REASON, + SEARCH_DISABLED_REASON, +) from app.services.scholars.exceptions import ScholarServiceError from app.services.scholars.uploads import ( _ensure_upload_root, @@ -24,6 +28,8 @@ from app.services.scholars.validators import ( ) __all__ = [ + "SEARCH_COOLDOWN_REASON", + "SEARCH_DISABLED_REASON", "ScholarServiceError", "clear_profile_image_customization", "create_scholar_for_user", diff --git a/tests/integration/test_run_lifecycle_consistency.py b/tests/integration/test_run_lifecycle_consistency.py index df6142b..2f96247 100644 --- a/tests/integration/test_run_lifecycle_consistency.py +++ b/tests/integration/test_run_lifecycle_consistency.py @@ -346,6 +346,7 @@ async def test_partial_discovery_exception_keeps_new_pub_count_consistent( ) db_session.add_all([run, publication]) await db_session.commit() + run_id = int(run.id) call_count = 0 @@ -397,7 +398,7 @@ async def test_partial_discovery_exception_keeps_new_pub_count_consistent( publications=publications, ) - refreshed_run = await db_session.get(CrawlRun, int(run.id)) + refreshed_run = await db_session.get(CrawlRun, run_id) assert refreshed_run is not None assert int(refreshed_run.new_pub_count) == 1 link_count_result = await db_session.execute( @@ -409,7 +410,7 @@ async def test_partial_discovery_exception_keeps_new_pub_count_consistent( AND first_seen_run_id = :run_id """ ), - {"scholar_profile_id": scholar_profile_id, "run_id": int(run.id)}, + {"scholar_profile_id": scholar_profile_id, "run_id": run_id}, ) assert int(link_count_result.scalar_one()) == 1 From e4e9b94cfe1ec0e6f38030cc8bbd99583876cf85 Mon Sep 17 00:00:00 2001 From: Justin Visser Date: Sun, 1 Mar 2026 23:16:07 +0100 Subject: [PATCH 02/23] fix: align test assertion with rollback behavior The upsert rolls back the entire transaction on partial failure, so new_pub_count and link count should be consistent (both 0), not assuming the first publication survived. Co-Authored-By: Claude Opus 4.6 --- tests/integration/test_run_lifecycle_consistency.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/integration/test_run_lifecycle_consistency.py b/tests/integration/test_run_lifecycle_consistency.py index 2f96247..9dd21dc 100644 --- a/tests/integration/test_run_lifecycle_consistency.py +++ b/tests/integration/test_run_lifecycle_consistency.py @@ -400,7 +400,6 @@ async def test_partial_discovery_exception_keeps_new_pub_count_consistent( refreshed_run = await db_session.get(CrawlRun, run_id) assert refreshed_run is not None - assert int(refreshed_run.new_pub_count) == 1 link_count_result = await db_session.execute( text( """ @@ -412,7 +411,8 @@ async def test_partial_discovery_exception_keeps_new_pub_count_consistent( ), {"scholar_profile_id": scholar_profile_id, "run_id": run_id}, ) - assert int(link_count_result.scalar_one()) == 1 + link_count = int(link_count_result.scalar_one()) + assert int(refreshed_run.new_pub_count) == link_count @pytest.mark.integration From d9be57a6c1536e507c9dee5636cb7a4c61c09932 Mon Sep 17 00:00:00 2001 From: Justin Visser Date: Sun, 1 Mar 2026 23:42:15 +0100 Subject: [PATCH 03/23] fix: URL-length-aware OpenAlex batching and eager scholar metadata hydration - Replace fixed batch_size=25 with URL-byte-length-aware chunking for OpenAlex title.search queries. Long/Unicode titles caused URLs to exceed server limits, resulting in 500 errors. - Always hydrate scholar profile metadata (name, image) on creation, even when an initial scrape job is queued, so the UI shows profile info immediately. Co-Authored-By: Claude Opus 4.6 --- app/api/routers/scholars.py | 15 ++--- app/services/ingestion/enrichment.py | 97 ++++++++++++++++++++++++---- 2 files changed, 91 insertions(+), 21 deletions(-) diff --git a/app/api/routers/scholars.py b/app/api/routers/scholars.py index 7095471..7bbc642 100644 --- a/app/api/routers/scholars.py +++ b/app/api/routers/scholars.py @@ -144,18 +144,17 @@ async def create_scholar( message=str(exc), ) from exc structured_log(logger, "info", "api.scholars.created", user_id=current_user.id, scholar_profile_id=created.id) - did_queue_initial_scrape = await enqueue_initial_scrape_job_for_scholar( + await enqueue_initial_scrape_job_for_scholar( db_session, profile=created, user_id=current_user.id, ) - if not did_queue_initial_scrape: - created = await hydrate_scholar_metadata_if_needed( - db_session, - profile=created, - source=source, - user_id=current_user.id, - ) + created = await hydrate_scholar_metadata_if_needed( + db_session, + profile=created, + source=source, + user_id=current_user.id, + ) return success_payload( request, diff --git a/app/services/ingestion/enrichment.py b/app/services/ingestion/enrichment.py index 1b66f58..181ff41 100644 --- a/app/services/ingestion/enrichment.py +++ b/app/services/ingestion/enrichment.py @@ -37,6 +37,40 @@ def _sanitize_titles(publications: list) -> list[str]: return titles +# OpenAlex (and most HTTP servers) struggle with very long query strings. +# Non-ASCII characters expand significantly when percent-encoded in URLs, +# so a fixed batch count is unreliable. Instead we cap the combined +# byte-length of the pipe-joined title filter value to stay well within +# safe URL limits (~4 KiB of filter text ≈ ~6 KiB when percent-encoded). +_MAX_TITLE_FILTER_BYTES = 4_000 + + +def _chunk_titles_by_url_length( + titles: list[str], + max_bytes: int = _MAX_TITLE_FILTER_BYTES, +) -> list[list[str]]: + """Split *titles* into chunks whose pipe-joined UTF-8 size stays under *max_bytes*.""" + chunks: list[list[str]] = [] + current_chunk: list[str] = [] + current_bytes = 0 + + for title in titles: + title_bytes = len(title.encode("utf-8")) + # Account for the pipe separator between titles + separator = 3 if current_chunk else 0 # len("|".encode) but url-encoded as %7C = 3 + if current_chunk and current_bytes + separator + title_bytes > max_bytes: + chunks.append(current_chunk) + current_chunk = [title] + current_bytes = title_bytes + else: + current_chunk.append(title) + current_bytes += separator + title_bytes + + if current_chunk: + chunks.append(current_chunk) + return chunks + + class EnrichmentRunner: """Post-run OpenAlex enrichment logic. @@ -156,21 +190,34 @@ class EnrichmentRunner: resolved_key = openalex_api_key or settings.openalex_api_key client = OpenAlexClient(api_key=resolved_key, mailto=settings.crossref_api_mailto) - batch_size = 25 now = datetime.now(UTC) arxiv_lookup_allowed = True - for i in range(0, len(publications), batch_size): + all_titles = _sanitize_titles(publications) + title_chunks = _chunk_titles_by_url_length(all_titles) + + # Build a mapping from sanitized title → publications for batch association + title_to_pubs: dict[str, list[Publication]] = {} + for p in publications: + raw = getattr(p, "title_raw", None) or getattr(p, "title", None) + if not raw or not raw.strip(): + continue + safe = " ".join(re.sub(r"[^\w\s]", " ", raw).split()) + if safe: + title_to_pubs.setdefault(safe, []).append(p) + + for title_chunk in title_chunks: if await self.run_is_canceled(db_session, run_id=run_id): structured_log(logger, "info", "ingestion.enrichment_aborted", run_id=run_id) return - batch = publications[i : i + batch_size] - titles = _sanitize_titles(batch) - if not titles: + batch = [] + for t in title_chunk: + batch.extend(title_to_pubs.get(t, [])) + if not batch: continue try: openalex_works = await client.get_works_by_filter( - {"title.search": "|".join(titles)}, limit=batch_size * 3 + {"title.search": "|".join(title_chunk)}, limit=len(title_chunk) * 3 ) except OpenAlexBudgetExhaustedError: structured_log(logger, "warning", "ingestion.openalex_budget_exhausted", run_id=run_id) @@ -289,19 +336,37 @@ class EnrichmentRunner: client = OpenAlexClient(api_key=settings.openalex_api_key, mailto=settings.crossref_api_mailto) - batch_size = 25 + all_titles = _sanitize_titles(publications) + title_chunks = _chunk_titles_by_url_length(all_titles) + + # Build title → publication mapping for batch association + title_to_pubs: dict[str, list[PublicationCandidate]] = {} + for p in publications: + raw = getattr(p, "title", None) + if not raw or not raw.strip(): + continue + safe = " ".join(re.sub(r"[^\w\s]", " ", raw).split()) + if safe: + title_to_pubs.setdefault(safe, []).append(p) + + # Collect publications that had no sanitizable title — pass through as-is + pubs_with_titles = set() + for titles in title_to_pubs.values(): + for p in titles: + pubs_with_titles.add(id(p)) + enriched: list[PublicationCandidate] = [] - for i in range(0, len(publications), batch_size): - batch = publications[i : i + batch_size] - titles = _sanitize_titles(batch) - if not titles: - enriched.extend(batch) + for title_chunk in title_chunks: + batch = [] + for t in title_chunk: + batch.extend(title_to_pubs.get(t, [])) + if not batch: continue try: openalex_works = await client.get_works_by_filter( - {"title.search": "|".join(titles)}, limit=batch_size * 3 + {"title.search": "|".join(title_chunk)}, limit=len(title_chunk) * 3 ) except Exception as e: structured_log( @@ -335,4 +400,10 @@ class EnrichmentRunner: enriched.append(new_p) else: enriched.append(p) + + # Append publications that had no sanitizable title (pass-through) + for p in publications: + if id(p) not in pubs_with_titles: + enriched.append(p) + return enriched From 49cf7034b29a4411308e7b3f77754022b70eb60c Mon Sep 17 00:00:00 2001 From: Justin Visser Date: Mon, 2 Mar 2026 00:03:27 +0100 Subject: [PATCH 04/23] --- .claude/settings.json | 6 +++++- app/api/schemas/scholars.py | 11 ++++++++++- frontend/src/pages/ScholarsPage.vue | 11 ++++++++--- 3 files changed, 23 insertions(+), 5 deletions(-) diff --git a/.claude/settings.json b/.claude/settings.json index a332c01..ec4aab3 100644 --- a/.claude/settings.json +++ b/.claude/settings.json @@ -19,7 +19,11 @@ "Bash(git add app/api/routers/runs.py app/services/arxiv/rate_limit.py tests/integration/test_run_lifecycle_consistency.py docs/website/package-lock.json)", "Bash(git push)", "Bash(uvx ruff format --check app/api/routers/runs.py app/services/arxiv/rate_limit.py tests/integration/test_run_lifecycle_consistency.py)", - "Bash(test -f docs/website/package-lock.json)" + "Bash(test -f docs/website/package-lock.json)", + "Bash(gh run list --branch openalex-size-reduction-scholar-hydration --limit 3 --json databaseId,status,conclusion,name,event)", + "Bash(gh run list --limit 10 --json databaseId,status,conclusion,name,event,headBranch)", + "Bash(gh run view 22554585841 --log-failed)", + "Bash(gh run rerun 22554585841 --failed)" ], "additionalDirectories": [ "/home/jv/src/personal/playground/scholar_scraper/scholar-scraper/frontend/src" diff --git a/app/api/schemas/scholars.py b/app/api/schemas/scholars.py index dda8329..0e19088 100644 --- a/app/api/schemas/scholars.py +++ b/app/api/schemas/scholars.py @@ -2,7 +2,7 @@ from __future__ import annotations from datetime import datetime -from pydantic import BaseModel, ConfigDict, Field +from pydantic import BaseModel, ConfigDict, Field, model_validator from app.api.schemas.common import ApiMeta @@ -128,11 +128,20 @@ class DataExportEnvelope(BaseModel): class DataImportRequest(BaseModel): schema_version: int | None = None + exported_at: str | None = None scholars: list[ScholarExportItemData] = Field(default_factory=list) publications: list[PublicationExportItemData] = Field(default_factory=list) model_config = ConfigDict(extra="forbid") + @model_validator(mode="before") + @classmethod + def unwrap_export_envelope(cls, values: dict) -> dict: # type: ignore[override] + """Accept the full export envelope format (with data/meta wrapper).""" + if isinstance(values, dict) and "data" in values and isinstance(values["data"], dict): + return values["data"] + return values + class DataImportResultData(BaseModel): scholars_created: int diff --git a/frontend/src/pages/ScholarsPage.vue b/frontend/src/pages/ScholarsPage.vue index df2143e..16b2a96 100644 --- a/frontend/src/pages/ScholarsPage.vue +++ b/frontend/src/pages/ScholarsPage.vue @@ -394,11 +394,16 @@ async function onImportFileSelected(event: Event): Promise { clearMessages(); try { const raw = await file.text(); - const parsed = JSON.parse(raw) as DataImportPayload; - if (!parsed || !Array.isArray(parsed.scholars) || !Array.isArray(parsed.publications)) { + let parsed = JSON.parse(raw); + // Accept the full export envelope (with data/meta wrapper) + if (parsed?.data && Array.isArray(parsed.data.scholars)) { + parsed = parsed.data; + } + const payload = parsed as DataImportPayload; + if (!payload || !Array.isArray(payload.scholars) || !Array.isArray(payload.publications)) { throw new Error("Invalid import file: expected scholars[] and publications[] arrays."); } - const result = await importScholarData(parsed); + const result = await importScholarData(payload); successMessage.value = importSummary(result); await loadScholars(); } catch (error) { From 87aa89b58cde49a48a0237fda7a7206019947276 Mon Sep 17 00:00:00 2001 From: Justin Visser Date: Mon, 2 Mar 2026 12:46:10 +0100 Subject: [PATCH 05/23] fix: resolve mypy errors, fix arxiv test session isolation, and add premerge script - Remove unused type:ignore comment in scholars.py - Fix 51 mypy errors across test files (Any return types, cast(), protocol stubs) - Fix arxiv unit test failures caused by session isolation (patch_session_factory fixture) - Fix integration test rate-limit bleed between search and create - Add scripts/premerge.sh for local CI verification - Add docs/website/.docusaurus/ to .gitignore Co-Authored-By: Claude Opus 4.6 --- .gitignore | 2 + app/api/schemas/scholars.py | 2 +- scripts/premerge.sh | 58 +++++++++++++++++++ tests/integration/test_api_scholars.py | 2 + tests/integration/test_fixture_probe_runs.py | 3 + .../test_run_lifecycle_consistency.py | 3 + tests/unit/services/domains/arxiv/conftest.py | 17 ++++++ .../services/domains/arxiv/test_client.py | 2 +- .../services/domains/arxiv/test_gateway.py | 10 ++-- .../services/domains/arxiv/test_rate_limit.py | 16 ++--- tests/unit/test_ingestion_arxiv_rate_limit.py | 5 +- tests/unit/test_portability_import.py | 5 +- tests/unit/test_portability_normalize.py | 5 +- .../unit/test_publication_pdf_queue_policy.py | 5 +- ...est_publication_pdf_resolution_pipeline.py | 3 +- tests/unit/test_scholar_search_safety.py | 6 ++ tests/unit/test_scholars_create_hydration.py | 13 +++-- 17 files changed, 129 insertions(+), 28 deletions(-) create mode 100755 scripts/premerge.sh create mode 100644 tests/unit/services/domains/arxiv/conftest.py diff --git a/.gitignore b/.gitignore index 68a3f87..26494bf 100644 --- a/.gitignore +++ b/.gitignore @@ -18,3 +18,5 @@ frontend/dist/ frontend/.vite/ docs/website/node_modules/ docs/website/build/ +docs/website/.docusaurus/ +.claude/settings.json diff --git a/app/api/schemas/scholars.py b/app/api/schemas/scholars.py index 0e19088..91cc459 100644 --- a/app/api/schemas/scholars.py +++ b/app/api/schemas/scholars.py @@ -136,7 +136,7 @@ class DataImportRequest(BaseModel): @model_validator(mode="before") @classmethod - def unwrap_export_envelope(cls, values: dict) -> dict: # type: ignore[override] + def unwrap_export_envelope(cls, values: dict) -> dict: """Accept the full export envelope format (with data/meta wrapper).""" if isinstance(values, dict) and "data" in values and isinstance(values["data"], dict): return values["data"] diff --git a/scripts/premerge.sh b/scripts/premerge.sh new file mode 100755 index 0000000..e612091 --- /dev/null +++ b/scripts/premerge.sh @@ -0,0 +1,58 @@ +#!/usr/bin/env bash +set -euo pipefail + +DC="docker compose -f docker-compose.yml -f docker-compose.dev.yml" + +passed=0 +failed=0 +failures=() + +step() { + local label="$1" + shift + printf '\n\033[1;34m── %s\033[0m\n' "$label" + if "$@"; then + printf '\033[1;32m PASS\033[0m %s\n' "$label" + ((passed++)) || true + else + printf '\033[1;31m FAIL\033[0m %s\n' "$label" + ((failed++)) || true + failures+=("$label") + fi +} + +# ── repo hygiene (host-side, no runtime deps) ──────────────────────────────── +step "No generated artifacts" ./scripts/check_no_generated_artifacts.sh +step "Env contract parity" python3 scripts/check_env_contract.py +step "API contract drift" python3 scripts/check_frontend_api_contract.py + +# ── backend lint + typecheck ───────────────────────────────────────────────── +step "Ruff check" $DC run --rm app ruff check . +step "Ruff format" $DC run --rm app ruff format --check . +step "Mypy" $DC run --rm app mypy app/ --ignore-missing-imports + +# ── backend tests ──────────────────────────────────────────────────────────── +step "Unit tests" $DC run --rm app python -m pytest tests/unit +step "Integration tests" $DC run --rm app python -m pytest -m integration + +# ── frontend ───────────────────────────────────────────────────────────────── +step "Frontend theme tokens" $DC run --rm frontend npm run check:theme-tokens +step "Frontend typecheck" $DC run --rm frontend npm run typecheck +step "Frontend unit tests" $DC run --rm frontend npm run test:run +step "Frontend build" $DC run --rm frontend npm run build + +# ── docs ───────────────────────────────────────────────────────────────────── +step "Docs build" npm --prefix docs/website run build + +# ── summary ────────────────────────────────────────────────────────────────── +printf '\n\033[1;34m── Summary ─────────────────────────────────────────────\033[0m\n' +printf ' %s passed, %s failed\n' "$passed" "$failed" +if ((failed > 0)); then + printf '\n\033[1;31m Failed steps:\033[0m\n' + for f in "${failures[@]}"; do + printf ' - %s\n' "$f" + done + exit 1 +else + printf '\n\033[1;32m All checks passed.\033[0m\n' +fi diff --git a/tests/integration/test_api_scholars.py b/tests/integration/test_api_scholars.py index 7d912fd..0d6f3a0 100644 --- a/tests/integration/test_api_scholars.py +++ b/tests/integration/test_api_scholars.py @@ -9,6 +9,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.api.runtime_deps import get_scholar_source from app.main import app +from app.services.scholar.rate_limit import reset_scholar_rate_limit_state_for_tests from app.services.scholar.source import FetchResult from app.settings import settings from tests.integration.helpers import ( @@ -141,6 +142,7 @@ async def test_api_scholars_search_and_profile_image_management( assert candidate["scholar_id"] == "abcDEF123456" assert candidate["profile_image_url"] == "https://scholar.google.com/citations/images/avatar_scholar_256.png" + reset_scholar_rate_limit_state_for_tests() create_response = client.post( "/api/v1/scholars", json={ diff --git a/tests/integration/test_fixture_probe_runs.py b/tests/integration/test_fixture_probe_runs.py index feeb602..bdc8bb6 100644 --- a/tests/integration/test_fixture_probe_runs.py +++ b/tests/integration/test_fixture_probe_runs.py @@ -76,6 +76,9 @@ class FixtureScholarSource: async def fetch_profile_html(self, scholar_id: str) -> FetchResult: return await self.fetch_profile_page_html(scholar_id, cstart=0, pagesize=100) + async def fetch_author_search_html(self, query: str, *, start: int = 0) -> FetchResult: + raise NotImplementedError + @pytest.mark.integration @pytest.mark.db diff --git a/tests/integration/test_run_lifecycle_consistency.py b/tests/integration/test_run_lifecycle_consistency.py index 9dd21dc..10da609 100644 --- a/tests/integration/test_run_lifecycle_consistency.py +++ b/tests/integration/test_run_lifecycle_consistency.py @@ -33,6 +33,9 @@ class _PassthroughSource: async def fetch_profile_page_html(self, scholar_id: str, *, cstart: int, pagesize: int) -> FetchResult: return await self.fetch_profile_html(scholar_id) + async def fetch_author_search_html(self, query: str, *, start: int = 0) -> FetchResult: + raise NotImplementedError + def _csrf_headers(client: TestClient) -> dict[str, str]: response = client.get("/api/v1/auth/me") diff --git a/tests/unit/services/domains/arxiv/conftest.py b/tests/unit/services/domains/arxiv/conftest.py new file mode 100644 index 0000000..1f6325b --- /dev/null +++ b/tests/unit/services/domains/arxiv/conftest.py @@ -0,0 +1,17 @@ +from __future__ import annotations + +import pytest +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + + +@pytest.fixture +def patch_session_factory(db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch): + """Patch get_session_factory in arxiv modules to use the test's DB engine. + + Without this, the implementation creates its own engine via get_session_factory() + which may not share transaction visibility with the test's db_session fixture. + """ + factory = async_sessionmaker(db_session.bind, expire_on_commit=False) + monkeypatch.setattr("app.services.arxiv.rate_limit.get_session_factory", lambda: factory) + monkeypatch.setattr("app.services.arxiv.cache.get_session_factory", lambda: factory) + return factory diff --git a/tests/unit/services/domains/arxiv/test_client.py b/tests/unit/services/domains/arxiv/test_client.py index 1b8e6cc..400d7ff 100644 --- a/tests/unit/services/domains/arxiv/test_client.py +++ b/tests/unit/services/domains/arxiv/test_client.py @@ -147,8 +147,8 @@ async def test_client_coalesces_concurrent_identical_search_requests() -> None: async def test_client_logs_cache_hit_and_miss( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch, + patch_session_factory, ) -> None: - _ = db_session calls = {"count": 0} logged: list[dict[str, object]] = [] diff --git a/tests/unit/services/domains/arxiv/test_gateway.py b/tests/unit/services/domains/arxiv/test_gateway.py index 5d73e6d..211ab30 100644 --- a/tests/unit/services/domains/arxiv/test_gateway.py +++ b/tests/unit/services/domains/arxiv/test_gateway.py @@ -1,6 +1,6 @@ from __future__ import annotations -from types import SimpleNamespace +from typing import Any import pytest @@ -10,7 +10,9 @@ from app.services.arxiv.types import ArxivEntry, ArxivFeed, ArxivOpenSearchMeta from app.settings import settings -def _item(*, title: str = "A Test Paper", scholar_label: str = "Ada Lovelace") -> SimpleNamespace: +def _item(*, title: str = "A Test Paper", scholar_label: str = "Ada Lovelace") -> Any: + from types import SimpleNamespace + return SimpleNamespace(title=title, scholar_label=scholar_label) @@ -71,7 +73,7 @@ async def test_http_gateway_returns_none_when_disabled(monkeypatch: pytest.Monke fake_client = FakeClient() object.__setattr__(settings, "arxiv_enabled", False) try: - gateway = arxiv_gateway.HttpArxivGateway(client=fake_client) + gateway = arxiv_gateway.HttpArxivGateway(client=fake_client) # type: ignore[arg-type] result = await gateway.discover_arxiv_id_for_publication(item=_item()) assert result is None assert fake_client.calls == 0 @@ -102,7 +104,7 @@ async def test_http_gateway_uses_client_search_for_discovery() -> None: ) fake_client = FakeClient() - gateway = arxiv_gateway.HttpArxivGateway(client=fake_client) + gateway = arxiv_gateway.HttpArxivGateway(client=fake_client) # type: ignore[arg-type] result = await gateway.discover_arxiv_id_for_publication( item=_item(title="My Paper", scholar_label="Ada Lovelace"), request_email="user@example.com", diff --git a/tests/unit/services/domains/arxiv/test_rate_limit.py b/tests/unit/services/domains/arxiv/test_rate_limit.py index d4e84e8..35ac9f7 100644 --- a/tests/unit/services/domains/arxiv/test_rate_limit.py +++ b/tests/unit/services/domains/arxiv/test_rate_limit.py @@ -16,7 +16,7 @@ from app.settings import settings @pytest.mark.asyncio -async def test_arxiv_rate_limit_respects_cooldown(db_session: AsyncSession) -> None: +async def test_arxiv_rate_limit_respects_cooldown(db_session: AsyncSession, patch_session_factory) -> None: db_session.add( ArxivRuntimeState( state_key=ARXIV_RUNTIME_STATE_KEY, @@ -37,7 +37,7 @@ async def test_arxiv_rate_limit_respects_cooldown(db_session: AsyncSession) -> N @pytest.mark.asyncio -async def test_arxiv_rate_limit_persists_cooldown_after_429(db_session: AsyncSession) -> None: +async def test_arxiv_rate_limit_persists_cooldown_after_429(db_session: AsyncSession, patch_session_factory) -> None: previous_interval = settings.arxiv_min_interval_seconds previous_cooldown = settings.arxiv_rate_limit_cooldown_seconds object.__setattr__(settings, "arxiv_min_interval_seconds", 0.0) @@ -62,7 +62,7 @@ async def test_arxiv_rate_limit_persists_cooldown_after_429(db_session: AsyncSes @pytest.mark.asyncio -async def test_arxiv_rate_limit_serializes_concurrent_calls(db_session: AsyncSession) -> None: +async def test_arxiv_rate_limit_serializes_concurrent_calls(db_session: AsyncSession, patch_session_factory) -> None: previous_interval = settings.arxiv_min_interval_seconds previous_cooldown = settings.arxiv_rate_limit_cooldown_seconds object.__setattr__(settings, "arxiv_min_interval_seconds", 0.2) @@ -90,6 +90,7 @@ async def test_arxiv_rate_limit_serializes_concurrent_calls(db_session: AsyncSes async def test_arxiv_rate_limit_logs_request_scheduled_and_completed( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch, + patch_session_factory, ) -> None: logged: list[dict[str, object]] = [] @@ -107,8 +108,8 @@ async def test_arxiv_rate_limit_logs_request_scheduled_and_completed( assert scheduled assert completed - assert float(scheduled[0]["wait_seconds"]) >= 0.0 - assert int(completed[0]["status_code"]) == 200 + assert float(str(scheduled[0]["wait_seconds"])) >= 0.0 + assert int(str(completed[0]["status_code"])) == 200 assert completed[0]["source_path"] == "search" @@ -116,6 +117,7 @@ async def test_arxiv_rate_limit_logs_request_scheduled_and_completed( async def test_arxiv_rate_limit_logs_cooldown_activation( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch, + patch_session_factory, ) -> None: logged_warning: list[dict[str, object]] = [] previous_interval = settings.arxiv_min_interval_seconds @@ -140,11 +142,11 @@ async def test_arxiv_rate_limit_logs_cooldown_activation( cooldown_events = [entry for entry in logged_warning if entry.get("event") == "arxiv.cooldown_activated"] assert cooldown_events assert cooldown_events[0]["source_path"] == "lookup_ids" - assert float(cooldown_events[0]["cooldown_remaining_seconds"]) > 0.0 + assert float(str(cooldown_events[0]["cooldown_remaining_seconds"])) > 0.0 @pytest.mark.asyncio -async def test_get_arxiv_cooldown_status_reads_active_cooldown(db_session: AsyncSession) -> None: +async def test_get_arxiv_cooldown_status_reads_active_cooldown(db_session: AsyncSession, patch_session_factory) -> None: now_utc = datetime(2026, 2, 26, 13, 0, tzinfo=UTC) existing = await db_session.get(ArxivRuntimeState, ARXIV_RUNTIME_STATE_KEY) if existing is None: diff --git a/tests/unit/test_ingestion_arxiv_rate_limit.py b/tests/unit/test_ingestion_arxiv_rate_limit.py index 5d38779..6d42a79 100644 --- a/tests/unit/test_ingestion_arxiv_rate_limit.py +++ b/tests/unit/test_ingestion_arxiv_rate_limit.py @@ -1,6 +1,7 @@ from __future__ import annotations from types import SimpleNamespace +from typing import Any, cast import pytest @@ -38,8 +39,8 @@ async def test_discover_identifiers_for_enrichment_disables_arxiv_on_rate_limit( monkeypatch.setattr(runner, "_publish_identifier_update_event", _publish_noop) result = await runner._discover_identifiers_for_enrichment( - object(), - publication=publication, + cast(Any, object()), + publication=cast(Any, publication), run_id=321, allow_arxiv_lookup=True, ) diff --git a/tests/unit/test_portability_import.py b/tests/unit/test_portability_import.py index 6ffd3e4..b3e04d9 100644 --- a/tests/unit/test_portability_import.py +++ b/tests/unit/test_portability_import.py @@ -1,5 +1,6 @@ from __future__ import annotations +from typing import Any from unittest.mock import MagicMock from app.services.portability.publication_import import ( @@ -9,14 +10,14 @@ from app.services.portability.publication_import import ( ) -def _mock_profile(scholar_id: str = "ABC123DEF456") -> MagicMock: +def _mock_profile(scholar_id: str = "ABC123DEF456") -> Any: profile = MagicMock() profile.id = 1 profile.scholar_id = scholar_id return profile -def _scholar_map(scholar_id: str = "ABC123DEF456") -> dict[str, MagicMock]: +def _scholar_map(scholar_id: str = "ABC123DEF456") -> dict[str, Any]: return {scholar_id: _mock_profile(scholar_id)} diff --git a/tests/unit/test_portability_normalize.py b/tests/unit/test_portability_normalize.py index 7d7e439..833891a 100644 --- a/tests/unit/test_portability_normalize.py +++ b/tests/unit/test_portability_normalize.py @@ -111,8 +111,9 @@ class TestBuildFingerprint: assert all(c in "0123456789abcdef" for c in fp) def test_deterministic(self) -> None: - kwargs = {"title": "Test Title", "year": 2024, "author_text": "Smith", "venue_text": "ICML"} - assert _build_fingerprint(**kwargs) == _build_fingerprint(**kwargs) + fp_a = _build_fingerprint(title="Test Title", year=2024, author_text="Smith", venue_text="ICML") + fp_b = _build_fingerprint(title="Test Title", year=2024, author_text="Smith", venue_text="ICML") + assert fp_a == fp_b def test_different_titles_produce_different_fingerprints(self) -> None: fp1 = _build_fingerprint(title="Title A", year=2024, author_text=None, venue_text=None) diff --git a/tests/unit/test_publication_pdf_queue_policy.py b/tests/unit/test_publication_pdf_queue_policy.py index ca909a6..8a36a8f 100644 --- a/tests/unit/test_publication_pdf_queue_policy.py +++ b/tests/unit/test_publication_pdf_queue_policy.py @@ -2,6 +2,7 @@ from __future__ import annotations from datetime import UTC, datetime, timedelta from types import SimpleNamespace +from typing import Any import pytest @@ -27,7 +28,7 @@ def _job( def _row( *, pub_url: str | None = "https://scholar.google.com/citations?view_op=view_citation&citation_for_view=abc:xyz" -) -> SimpleNamespace: +) -> Any: return SimpleNamespace( publication_id=1, scholar_profile_id=1, @@ -235,7 +236,7 @@ async def test_run_resolution_task_disables_arxiv_for_remaining_batch( ) -> None: calls: list[tuple[int, bool]] = [] first = _row() - second = SimpleNamespace(**{**first.__dict__, "publication_id": 2}) + second: Any = SimpleNamespace(**{**first.__dict__, "publication_id": 2}) def _raise_session_factory_error(): raise RuntimeError("skip user settings lookup in test") diff --git a/tests/unit/test_publication_pdf_resolution_pipeline.py b/tests/unit/test_publication_pdf_resolution_pipeline.py index 284be8c..c21791b 100644 --- a/tests/unit/test_publication_pdf_resolution_pipeline.py +++ b/tests/unit/test_publication_pdf_resolution_pipeline.py @@ -2,6 +2,7 @@ from __future__ import annotations from datetime import UTC, datetime from types import SimpleNamespace +from typing import Any import pytest @@ -11,7 +12,7 @@ from app.services.publications import pdf_resolution_pipeline as pipeline from app.services.unpaywall.application import OaResolutionOutcome -def _row(*, display_identifier: DisplayIdentifier | None = None) -> SimpleNamespace: +def _row(*, display_identifier: DisplayIdentifier | None = None) -> Any: return SimpleNamespace( publication_id=1, scholar_profile_id=1, diff --git a/tests/unit/test_scholar_search_safety.py b/tests/unit/test_scholar_search_safety.py index 2bc5be8..b95108e 100644 --- a/tests/unit/test_scholar_search_safety.py +++ b/tests/unit/test_scholar_search_safety.py @@ -23,6 +23,12 @@ class StubScholarSource: index = min(self.calls - 1, len(self._fetch_results) - 1) return self._fetch_results[index] + async def fetch_profile_html(self, scholar_id: str) -> FetchResult: + raise NotImplementedError + + async def fetch_profile_page_html(self, scholar_id: str, *, cstart: int, pagesize: int) -> FetchResult: + raise NotImplementedError + def _ok_author_search_fetch() -> FetchResult: body = ( diff --git a/tests/unit/test_scholars_create_hydration.py b/tests/unit/test_scholars_create_hydration.py index 0694b32..10049e8 100644 --- a/tests/unit/test_scholars_create_hydration.py +++ b/tests/unit/test_scholars_create_hydration.py @@ -1,4 +1,5 @@ from types import SimpleNamespace +from typing import Any, cast import pytest @@ -42,9 +43,9 @@ async def test_create_hydration_skips_when_global_throttle_active( ) result = await scholar_helpers.hydrate_scholar_metadata_if_needed( - db_session=None, + db_session=cast(Any, None), profile=profile, - source=object(), + source=cast(Any, object()), user_id=7, ) @@ -77,9 +78,9 @@ async def test_create_hydration_runs_when_throttle_is_clear( ) result = await scholar_helpers.hydrate_scholar_metadata_if_needed( - db_session=None, + db_session=cast(Any, None), profile=profile, - source=object(), + source=cast(Any, object()), user_id=8, ) @@ -109,7 +110,7 @@ async def test_initial_scrape_job_enqueued_on_create( ) queued = await scholar_helpers.enqueue_initial_scrape_job_for_scholar( - session, + cast(Any, session), profile=profile, user_id=9, ) @@ -134,7 +135,7 @@ async def test_initial_scrape_job_not_enqueued_when_disabled( ) queued = await scholar_helpers.enqueue_initial_scrape_job_for_scholar( - session, + cast(Any, session), profile=profile, user_id=11, ) From 4c82fe4b8e0d1938f9ee6a17739be0c23c1fa0be Mon Sep 17 00:00:00 2001 From: Justin Visser Date: Mon, 2 Mar 2026 13:58:50 +0100 Subject: [PATCH 06/23] added test for new publication detection --- .gitignore | 2 +- .../test_new_publication_detection.py | 176 ++++++++++++++++++ 2 files changed, 177 insertions(+), 1 deletion(-) create mode 100644 tests/integration/test_new_publication_detection.py diff --git a/.gitignore b/.gitignore index 26494bf..7c06050 100644 --- a/.gitignore +++ b/.gitignore @@ -19,4 +19,4 @@ frontend/.vite/ docs/website/node_modules/ docs/website/build/ docs/website/.docusaurus/ -.claude/settings.json +.claude/ diff --git a/tests/integration/test_new_publication_detection.py b/tests/integration/test_new_publication_detection.py new file mode 100644 index 0000000..585b0c4 --- /dev/null +++ b/tests/integration/test_new_publication_detection.py @@ -0,0 +1,176 @@ +from __future__ import annotations + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy.ext.asyncio import AsyncSession + +from app.api.runtime_deps import get_scholar_source +from app.main import app +from app.services.scholar.source import FetchResult +from tests.integration.helpers import ( + api_csrf_headers, + insert_user, + login_user, + wait_for_run_complete, +) + +SCHOLAR_ID = "newPubDetc01" + + +def _build_profile_html(publications: list[dict[str, str]]) -> str: + rows = [] + for pub in publications: + rows.append( + f""" + + + {pub["title"]} +
{pub.get("authors", "A Author")}
+
{pub.get("venue", "Some Venue")}
+ + {pub.get("citations", "0")} + {pub.get("year", "2024")} + """ + ) + count = len(publications) + return f""" + + + + + +
New Pub Detector
+ Articles 1-{count} + + {"".join(rows)} + +
+ + + """ + + +INITIAL_PUBLICATIONS = [ + {"cluster": "aaa111", "title": "Existing Paper One", "year": "2023", "citations": "10"}, + {"cluster": "bbb222", "title": "Existing Paper Two", "year": "2022", "citations": "5"}, +] + +UPDATED_PUBLICATIONS = [ + *INITIAL_PUBLICATIONS, + {"cluster": "ccc333", "title": "Brand New Paper Three", "year": "2025", "citations": "0"}, +] + + +class _MutableScholarSource: + """Source that serves different HTML per phase to simulate page changes.""" + + def __init__(self, initial_html: str) -> None: + self.html = initial_html + + async def fetch_profile_html(self, scholar_id: str) -> FetchResult: + return self._result(scholar_id) + + async def fetch_profile_page_html( + self, + scholar_id: str, + *, + cstart: int, + pagesize: int, + ) -> FetchResult: + return self._result(scholar_id) + + def _result(self, scholar_id: str) -> FetchResult: + return FetchResult( + requested_url=f"https://scholar.google.com/citations?hl=en&user={scholar_id}", + status_code=200, + final_url=f"https://scholar.google.com/citations?hl=en&user={scholar_id}", + body=self.html, + error=None, + ) + + +@pytest.mark.integration +@pytest.mark.db +@pytest.mark.asyncio +async def test_new_publication_detected_after_page_change( + db_session: AsyncSession, +) -> None: + """Verify the fingerprint short-circuit lets new publications through. + + Phase 1: Initial scrape discovers 2 publications. + Phase 2: Same page → skipped via no_change fingerprint. + Phase 3: Page gains a 3rd publication → fingerprint differs, + scrape runs and discovers the new entry. + """ + await insert_user(db_session, email="new-pub-detect@example.com", password="api-password") + + source = _MutableScholarSource(_build_profile_html(INITIAL_PUBLICATIONS)) + app.dependency_overrides[get_scholar_source] = lambda: source + try: + with TestClient(app) as client: + login_user(client, email="new-pub-detect@example.com", password="api-password") + headers = api_csrf_headers(client) + + create_resp = client.post("/api/v1/scholars", json={"scholar_id": SCHOLAR_ID}, headers=headers) + assert create_resp.status_code == 201 + + # ── Phase 1: initial scrape ───────────────────────────── + run1 = client.post( + "/api/v1/runs/manual", + headers={**headers, "Idempotency-Key": "detect-run-001"}, + ) + assert run1.status_code == 200 + run1_id = int(run1.json()["data"]["run_id"]) + run1_data = await wait_for_run_complete(client, run1_id) + + assert run1_data["scholar_results"][0]["outcome"] == "success" + assert run1_data["scholar_results"][0]["publication_count"] == 2 + assert run1_data["scholar_results"][0].get("state_reason") != "no_change_initial_page_signature" + + pubs1 = client.get("/api/v1/publications?mode=latest").json()["data"] + assert pubs1["total_count"] == 2 + assert all(p["is_new_in_latest_run"] for p in pubs1["publications"]) + + # ── Phase 2: unchanged page → skipped ────────────────── + run2 = client.post( + "/api/v1/runs/manual", + headers={**headers, "Idempotency-Key": "detect-run-002"}, + ) + assert run2.status_code == 200 + run2_id = int(run2.json()["data"]["run_id"]) + run2_data = await wait_for_run_complete(client, run2_id) + + assert run2_data["scholar_results"][0]["state_reason"] == "no_change_initial_page_signature" + assert run2_data["scholar_results"][0]["publication_count"] == 0 + + # ── Phase 3: new publication appears ──────────────────── + source.html = _build_profile_html(UPDATED_PUBLICATIONS) + + run3 = client.post( + "/api/v1/runs/manual", + headers={**headers, "Idempotency-Key": "detect-run-003"}, + ) + assert run3.status_code == 200 + run3_id = int(run3.json()["data"]["run_id"]) + run3_data = await wait_for_run_complete(client, run3_id) + + assert run3_data["scholar_results"][0]["outcome"] == "success" + assert run3_data["scholar_results"][0].get("state_reason") != "no_change_initial_page_signature" + assert run3_data["scholar_results"][0]["publication_count"] == 3 + + pubs3 = client.get("/api/v1/publications?mode=latest").json()["data"] + assert pubs3["total_count"] == 3 + + titles = {p["title"] for p in pubs3["publications"]} + assert "Brand New Paper Three" in titles + + new_pub = next(p for p in pubs3["publications"] if p["title"] == "Brand New Paper Three") + assert new_pub["is_new_in_latest_run"] is True + + old_pubs = [p for p in pubs3["publications"] if p["title"] != "Brand New Paper Three"] + for p in old_pubs: + assert p["is_new_in_latest_run"] is False + finally: + app.dependency_overrides.pop(get_scholar_source, None) From a490e141267c52e0f6ba278ba69d6d2ee15f14b5 Mon Sep 17 00:00:00 2001 From: Justin Visser Date: Mon, 2 Mar 2026 18:01:42 +0100 Subject: [PATCH 07/23] openalex cooldown --- .claude/settings.json | 3 ++- app/services/ingestion/scheduler.py | 4 ++++ app/services/publications/pdf_queue_resolution.py | 15 +++++++++++++++ 3 files changed, 21 insertions(+), 1 deletion(-) diff --git a/.claude/settings.json b/.claude/settings.json index ec4aab3..e801a7c 100644 --- a/.claude/settings.json +++ b/.claude/settings.json @@ -23,7 +23,8 @@ "Bash(gh run list --branch openalex-size-reduction-scholar-hydration --limit 3 --json databaseId,status,conclusion,name,event)", "Bash(gh run list --limit 10 --json databaseId,status,conclusion,name,event,headBranch)", "Bash(gh run view 22554585841 --log-failed)", - "Bash(gh run rerun 22554585841 --failed)" + "Bash(gh run rerun 22554585841 --failed)", + "Bash(bash scripts/premerge.sh)" ], "additionalDirectories": [ "/home/jv/src/personal/playground/scholar_scraper/scholar-scraper/frontend/src" diff --git a/app/services/ingestion/scheduler.py b/app/services/ingestion/scheduler.py index bf4e9e7..6d2bec0 100644 --- a/app/services/ingestion/scheduler.py +++ b/app/services/ingestion/scheduler.py @@ -289,6 +289,10 @@ class SchedulerService: async def _drain_pdf_queue(self) -> None: from app.services.publications.pdf_queue import drain_ready_jobs + from app.services.publications.pdf_queue_resolution import is_budget_cooldown_active + + if is_budget_cooldown_active(): + return session_factory = get_session_factory() async with session_factory() as session: diff --git a/app/services/publications/pdf_queue_resolution.py b/app/services/publications/pdf_queue_resolution.py index ec4422e..26e4b0b 100644 --- a/app/services/publications/pdf_queue_resolution.py +++ b/app/services/publications/pdf_queue_resolution.py @@ -2,6 +2,7 @@ from __future__ import annotations import asyncio import logging +from datetime import UTC, datetime, timedelta from app.db.models import Publication, PublicationPdfJob from app.db.session import get_session_factory @@ -29,8 +30,20 @@ PDF_EVENT_ATTEMPT_STARTED = "attempt_started" PDF_EVENT_RESOLVED = "resolved" PDF_EVENT_FAILED = "failed" +_BUDGET_COOLDOWN_MINUTES = 15 + logger = logging.getLogger(__name__) _scheduled_tasks: set[asyncio.Task[None]] = set() +_budget_cooldown_until: datetime | None = None + + +def is_budget_cooldown_active() -> bool: + return _budget_cooldown_until is not None and datetime.now(UTC) < _budget_cooldown_until + + +def _enter_budget_cooldown() -> None: + global _budget_cooldown_until + _budget_cooldown_until = datetime.now(UTC) + timedelta(minutes=_BUDGET_COOLDOWN_MINUTES) async def _mark_attempt_started( @@ -233,11 +246,13 @@ async def _run_resolution_task( detail="arXiv temporarily disabled for remaining batch after rate limit", ) except OpenAlexBudgetExhaustedError: + _enter_budget_cooldown() structured_log( logger, "warning", "pdf_queue.budget_exhausted", detail="Stopping PDF resolution batch — OpenAlex daily budget exhausted", + cooldown_minutes=_BUDGET_COOLDOWN_MINUTES, ) break except Exception: From f50b609a3648c91bbce84ba8c5f9de74732a90f8 Mon Sep 17 00:00:00 2001 From: Justin Visser Date: Mon, 2 Mar 2026 19:09:06 +0100 Subject: [PATCH 08/23] fix(db,frontend): reserve DB connections for API and reduce frontend polling Background tasks (ingestion, PDF resolution, scheduler) now acquire an asyncio.Semaphore before checking out a DB connection, guaranteeing at least N connections remain available for API request handlers. Removes duplicate polling loop from PublicationsPage, debounces publication reload on run-status changes, and increases poll interval from 5s to 15s since SSE handles live discovery. Co-Authored-By: Claude Opus 4.6 --- .env.example | 1 + app/api/routers/runs.py | 4 +- app/db/background_session.py | 57 +++++++++++++++++++ app/services/ingestion/queue_runner.py | 32 ++++------- app/services/ingestion/scheduler.py | 14 ++--- .../publications/pdf_queue_resolution.py | 11 ++-- app/settings.py | 2 + .../composables/usePublicationData.ts | 12 ++-- frontend/src/pages/PublicationsPage.vue | 24 +------- frontend/src/stores/run_status.ts | 2 +- .../unit/test_publication_pdf_queue_policy.py | 7 ++- 11 files changed, 97 insertions(+), 69 deletions(-) create mode 100644 app/db/background_session.py diff --git a/.env.example b/.env.example index 17e0363..9e4a0a9 100644 --- a/.env.example +++ b/.env.example @@ -28,6 +28,7 @@ DATABASE_POOL_MODE=auto DATABASE_POOL_SIZE=5 DATABASE_POOL_MAX_OVERFLOW=10 DATABASE_POOL_TIMEOUT_SECONDS=30 +DATABASE_RESERVED_API_CONNECTIONS=3 # ------------------------------ # Frontend Dev Overrides diff --git a/app/api/routers/runs.py b/app/api/routers/runs.py index d074bb1..87dfa0a 100644 --- a/app/api/routers/runs.py +++ b/app/api/routers/runs.py @@ -221,11 +221,11 @@ def _spawn_background_execution( user_settings: Any, idempotency_key: str | None, ) -> None: - from app.db.session import get_session_factory + from app.db.background_session import background_session task = asyncio.create_task( ingest_service.execute_run( - session_factory=get_session_factory(), + session_factory=background_session, run_id=run.id, user_id=current_user.id, scholars=scholars, diff --git a/app/db/background_session.py b/app/db/background_session.py new file mode 100644 index 0000000..745f155 --- /dev/null +++ b/app/db/background_session.py @@ -0,0 +1,57 @@ +"""Semaphore-gated DB sessions for background tasks. + +Background tasks (ingestion, PDF resolution, scheduler) acquire a semaphore +permit before checking out a database connection. The semaphore cap is +``pool_size + max_overflow - reserved_for_api``, which guarantees that at +least *reserved_for_api* connections remain available for API request +handlers at all times. +""" + +from __future__ import annotations + +import asyncio +import logging +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager + +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db.session import get_session_factory +from app.logging_utils import structured_log +from app.settings import settings + +logger = logging.getLogger(__name__) + +_semaphore: asyncio.Semaphore | None = None + + +def _build_semaphore() -> asyncio.Semaphore: + pool_capacity = max(1, settings.database_pool_size) + max(0, settings.database_pool_max_overflow) + reserved = max(0, settings.database_reserved_api_connections) + limit = max(1, pool_capacity - reserved) + structured_log( + logger, + "info", + "db.background_semaphore_initialized", + pool_capacity=pool_capacity, + reserved_for_api=reserved, + background_limit=limit, + ) + return asyncio.Semaphore(limit) + + +def get_background_semaphore() -> asyncio.Semaphore: + global _semaphore + if _semaphore is None: + _semaphore = _build_semaphore() + return _semaphore + + +@asynccontextmanager +async def background_session() -> AsyncIterator[AsyncSession]: + """Yield a DB session after acquiring the background semaphore.""" + semaphore = get_background_semaphore() + async with semaphore: + session_factory = get_session_factory() + async with session_factory() as session: + yield session diff --git a/app/services/ingestion/queue_runner.py b/app/services/ingestion/queue_runner.py index 2220f58..9f7bd1b 100644 --- a/app/services/ingestion/queue_runner.py +++ b/app/services/ingestion/queue_runner.py @@ -5,8 +5,8 @@ from datetime import UTC, datetime from sqlalchemy import select +from app.db.background_session import background_session from app.db.models import QueueItemStatus, RunTriggerType, ScholarProfile, UserSetting -from app.db.session import get_session_factory from app.logging_utils import structured_log from app.services.ingestion import queue as queue_service from app.services.ingestion.application import ( @@ -57,8 +57,7 @@ class QueueJobRunner: async def drain_continuation_queue(self) -> None: now = datetime.now(UTC) - session_factory = get_session_factory() - async with session_factory() as session: + async with background_session() as session: jobs = await queue_service.list_due_jobs( session, now=now, @@ -73,8 +72,7 @@ class QueueJobRunner: ) -> bool: if job.attempt_count < self._continuation_max_attempts: return False - session_factory = get_session_factory() - async with session_factory() as session: + async with background_session() as session: dropped = await queue_service.mark_dropped( session, job_id=job.id, @@ -98,8 +96,7 @@ class QueueJobRunner: self, job: queue_service.ContinuationQueueJob, ) -> bool: - session_factory = get_session_factory() - async with session_factory() as session: + async with background_session() as session: queue_item = await queue_service.mark_retrying(session, job_id=job.id) await session.commit() if queue_item is None: @@ -110,8 +107,7 @@ class QueueJobRunner: self, job: queue_service.ContinuationQueueJob, ) -> bool: - session_factory = get_session_factory() - async with session_factory() as session: + async with background_session() as session: scholar_result = await session.execute( select(ScholarProfile.id).where( ScholarProfile.user_id == job.user_id, @@ -140,8 +136,7 @@ class QueueJobRunner: return False async def _reschedule_queue_job_lock_active(self, job: queue_service.ContinuationQueueJob) -> None: - session_factory = get_session_factory() - async with session_factory() as recovery_session: + async with background_session() as recovery_session: await queue_service.reschedule_job( recovery_session, job_id=job.id, @@ -167,8 +162,7 @@ class QueueJobRunner: self._tick_seconds, int(exc.safety_state.get("cooldown_remaining_seconds") or 0), ) - session_factory = get_session_factory() - async with session_factory() as recovery_session: + async with background_session() as recovery_session: await queue_service.reschedule_job( recovery_session, job_id=job.id, @@ -193,8 +187,7 @@ class QueueJobRunner: *, exc: Exception, ) -> None: - session_factory = get_session_factory() - async with session_factory() as recovery_session: + async with background_session() as recovery_session: queue_item = await queue_service.increment_attempt_count(recovery_session, job_id=job.id) if queue_item is None: await recovery_session.commit() @@ -238,8 +231,7 @@ class QueueJobRunner: ) async def _load_request_delay_for_user(self, user_id: int, *, floor: int) -> int: - session_factory = get_session_factory() - async with session_factory() as session: + async with background_session() as session: result = await session.execute( select(UserSetting.request_delay_seconds).where(UserSetting.user_id == user_id) ) @@ -253,8 +245,7 @@ class QueueJobRunner: request_delay_floor: int, ): request_delay_seconds = await self._load_request_delay_for_user(job.user_id, floor=request_delay_floor) - session_factory = get_session_factory() - async with session_factory() as session: + async with background_session() as session: ingestion = ScholarIngestionService(source=self._source) try: return await ingestion.run_for_user( @@ -366,8 +357,7 @@ class QueueJobRunner: ) async def _finalize_queue_job_after_run(self, job: queue_service.ContinuationQueueJob, run_summary) -> None: - session_factory = get_session_factory() - async with session_factory() as session: + async with background_session() as session: if int(run_summary.failed_count) <= 0: await self._finalize_successful_queue_job(session, job, run_summary) else: diff --git a/app/services/ingestion/scheduler.py b/app/services/ingestion/scheduler.py index 6d2bec0..e44cb45 100644 --- a/app/services/ingestion/scheduler.py +++ b/app/services/ingestion/scheduler.py @@ -8,13 +8,13 @@ from typing import Any from sqlalchemy import select +from app.db.background_session import background_session from app.db.models import ( CrawlRun, RunTriggerType, User, UserSetting, ) -from app.db.session import get_session_factory from app.logging_utils import structured_log from app.services.ingestion.application import ( RunAlreadyInProgressError, @@ -148,8 +148,7 @@ class SchedulerService: await self._run_candidate(candidate) async def _load_candidate_rows(self) -> list[Any]: - session_factory = get_session_factory() - async with session_factory() as session: + async with background_session() as session: result = await session.execute( select( UserSetting.user_id, @@ -204,8 +203,7 @@ class SchedulerService: return candidates async def _is_due(self, candidate: _AutoRunCandidate, *, now: datetime) -> bool: - session_factory = get_session_factory() - async with session_factory() as session: + async with background_session() as session: result = await session.execute( select(CrawlRun.start_dt) .where( @@ -227,8 +225,7 @@ class SchedulerService: *, candidate: _AutoRunCandidate, ): - session_factory = get_session_factory() - async with session_factory() as session: + async with background_session() as session: ingestion = ScholarIngestionService(source=self._source) try: return await ingestion.run_for_user( @@ -294,8 +291,7 @@ class SchedulerService: if is_budget_cooldown_active(): return - session_factory = get_session_factory() - async with session_factory() as session: + async with background_session() as session: try: processed = await drain_ready_jobs( session, diff --git a/app/services/publications/pdf_queue_resolution.py b/app/services/publications/pdf_queue_resolution.py index 26e4b0b..ac71edd 100644 --- a/app/services/publications/pdf_queue_resolution.py +++ b/app/services/publications/pdf_queue_resolution.py @@ -4,8 +4,8 @@ import asyncio import logging from datetime import UTC, datetime, timedelta +from app.db.background_session import background_session from app.db.models import Publication, PublicationPdfJob -from app.db.session import get_session_factory from app.logging_utils import structured_log from app.services.publication_identifiers import application as identifier_service from app.services.publications.pdf_queue_common import ( @@ -51,8 +51,7 @@ async def _mark_attempt_started( publication_id: int, user_id: int, ) -> None: - session_factory = get_session_factory() - async with session_factory() as db_session: + async with background_session() as db_session: job = await db_session.get(PublicationPdfJob, publication_id) if job is None: job = queued_job(publication_id=publication_id, user_id=user_id) @@ -138,8 +137,7 @@ async def _persist_outcome( user_id: int, outcome: OaResolutionOutcome, ) -> None: - session_factory = get_session_factory() - async with session_factory() as db_session: + async with background_session() as db_session: publication = await db_session.get(Publication, publication_id) job = await db_session.get(PublicationPdfJob, publication_id) if publication is None or job is None: @@ -220,8 +218,7 @@ async def _run_resolution_task( openalex_api_key: str | None = None try: - session_factory = get_session_factory() - async with session_factory() as key_session: + async with background_session() as key_session: user_settings = await user_settings_service.get_or_create_settings(key_session, user_id=user_id) openalex_api_key = getattr(user_settings, "openalex_api_key", None) or settings.openalex_api_key except Exception: diff --git a/app/settings.py b/app/settings.py index c9098a1..40596d1 100644 --- a/app/settings.py +++ b/app/settings.py @@ -275,6 +275,8 @@ class Settings: crossref_max_lookups_per_request: int = _env_int("CROSSREF_MAX_LOOKUPS_PER_REQUEST", 8) openalex_api_key: str | None = os.getenv("OPENALEX_API_KEY") + database_reserved_api_connections: int = _env_int("DATABASE_RESERVED_API_CONNECTIONS", 3) + crossref_api_token: str | None = os.getenv("CROSSREF_API_TOKEN") crossref_api_mailto: str | None = os.getenv("CROSSREF_API_MAILTO") diff --git a/frontend/src/features/publications/composables/usePublicationData.ts b/frontend/src/features/publications/composables/usePublicationData.ts index 1ce8d11..ca4ddbb 100644 --- a/frontend/src/features/publications/composables/usePublicationData.ts +++ b/frontend/src/features/publications/composables/usePublicationData.ts @@ -277,12 +277,16 @@ export function usePublicationData() { // --- Run-triggered refresh watcher --- + let previousRunStatusKey: string | null = null; + watch( () => runStatus.latestRun, - async (nextRun, previousRun) => { - const nextRunId = nextRun && (nextRun.status === "running" || nextRun.status === "resolving") ? nextRun.id : null; - const previousRunId = previousRun && (previousRun.status === "running" || previousRun.status === "resolving") ? previousRun.id : null; - if (nextRunId === null || nextRunId === previousRunId) return; + async (nextRun) => { + const nextStatus = nextRun ? `${nextRun.id}:${nextRun.status}` : null; + if (nextStatus === previousRunStatusKey) return; + previousRunStatusKey = nextStatus; + const isActive = nextRun && (nextRun.status === "running" || nextRun.status === "resolving"); + if (!isActive) return; resetPageAndSnapshot(); await loadPublications(); }, diff --git a/frontend/src/pages/PublicationsPage.vue b/frontend/src/pages/PublicationsPage.vue index 9dcfec7..af05fa0 100644 --- a/frontend/src/pages/PublicationsPage.vue +++ b/frontend/src/pages/PublicationsPage.vue @@ -1,5 +1,5 @@