diff --git a/app/services/ingestion/application.py b/app/services/ingestion/application.py index f1a906e..01e4dd1 100644 --- a/app/services/ingestion/application.py +++ b/app/services/ingestion/application.py @@ -22,7 +22,7 @@ from app.services.ingestion.constants import RUN_LOCK_NAMESPACE from app.services.ingestion.enrichment import EnrichmentRunner from app.services.ingestion.pagination import PaginationEngine from app.services.ingestion.run_completion import ( - _int_or_default, + int_or_default, complete_run_for_user, run_execution_summary, ) @@ -135,7 +135,7 @@ class ScholarIngestionService: policy_minimum = user_settings_service.resolve_request_delay_minimum( settings.ingestion_min_request_delay_seconds ) - return max(policy_minimum, _int_or_default(value, policy_minimum)) + return max(policy_minimum, int_or_default(value, policy_minimum)) async def _load_user_settings_for_run( self, @@ -333,13 +333,13 @@ class ScholarIngestionService: alert_retry_scheduled_threshold: int = 3, ) -> tuple[CrawlRun, list[ScholarProfile], dict[int, int]]: effective_delay = self._effective_request_delay_seconds(request_delay_seconds) - if effective_delay != _int_or_default(request_delay_seconds, effective_delay): + if effective_delay != int_or_default(request_delay_seconds, effective_delay): structured_log( logger, "warning", "ingestion.delay_coerced", user_id=user_id, - requested_request_delay_seconds=_int_or_default(request_delay_seconds, 0), + requested_request_delay_seconds=int_or_default(request_delay_seconds, 0), effective_request_delay_seconds=effective_delay, policy_minimum_request_delay_seconds=user_settings_service.resolve_request_delay_minimum( settings.ingestion_min_request_delay_seconds diff --git a/app/services/ingestion/queue_runner.py b/app/services/ingestion/queue_runner.py index 82baab3..4eaaa98 100644 --- a/app/services/ingestion/queue_runner.py +++ b/app/services/ingestion/queue_runner.py @@ -20,7 +20,7 @@ from app.settings import settings logger = logging.getLogger(__name__) -def _effective_request_delay_seconds(value: int | None, *, floor: int) -> int: +def effective_request_delay_seconds(value: int | None, *, floor: int) -> int: try: parsed = int(value) if value is not None else floor except (TypeError, ValueError): @@ -238,7 +238,7 @@ class QueueJobRunner: select(UserSetting.request_delay_seconds).where(UserSetting.user_id == user_id) ) delay = result.scalar_one_or_none() - return _effective_request_delay_seconds(delay, floor=floor) + return effective_request_delay_seconds(delay, floor=floor) async def _run_ingestion_for_queue_job( self, diff --git a/app/services/ingestion/run_completion.py b/app/services/ingestion/run_completion.py index 450550d..b50f770 100644 --- a/app/services/ingestion/run_completion.py +++ b/app/services/ingestion/run_completion.py @@ -35,7 +35,7 @@ from app.settings import settings logger = logging.getLogger(__name__) -def _int_or_default(value: Any, default: int = 0) -> int: +def int_or_default(value: Any, default: int = 0) -> int: try: return int(value) except (TypeError, ValueError): @@ -68,7 +68,7 @@ def summarize_failures( scholars_with_retries_count = 0 retry_exhausted_count = 0 for entry in scholar_results: - retries_for_entry = max(0, _int_or_default(entry.get("attempt_count"), 0) - 1) + retries_for_entry = max(0, int_or_default(entry.get("attempt_count"), 0) - 1) if retries_for_entry > 0: retries_scheduled_count += retries_for_entry scholars_with_retries_count += 1 @@ -347,7 +347,7 @@ def find_scholar_result_index( scholar_profile_id: int, ) -> int | None: for index, result_entry in enumerate(scholar_results): - current_scholar_id = _int_or_default(result_entry.get("scholar_profile_id"), 0) + current_scholar_id = int_or_default(result_entry.get("scholar_profile_id"), 0) if current_scholar_id == scholar_profile_id: return index return None @@ -372,7 +372,7 @@ def apply_outcome_to_progress( progress: RunProgress, outcome: ScholarProcessingOutcome, ) -> None: - scholar_profile_id = _int_or_default(outcome.result_entry.get("scholar_profile_id"), 0) + scholar_profile_id = int_or_default(outcome.result_entry.get("scholar_profile_id"), 0) if scholar_profile_id <= 0: raise RuntimeError("Scholar outcome missing valid scholar_profile_id.") prior_index = find_scholar_result_index( diff --git a/app/services/ingestion/scheduler.py b/app/services/ingestion/scheduler.py index 23b70d0..1e5f08c 100644 --- a/app/services/ingestion/scheduler.py +++ b/app/services/ingestion/scheduler.py @@ -21,7 +21,7 @@ from app.services.ingestion.application import ( RunBlockedBySafetyPolicyError, ScholarIngestionService, ) -from app.services.ingestion.queue_runner import QueueJobRunner +from app.services.ingestion.queue_runner import QueueJobRunner, effective_request_delay_seconds from app.services.scholar.source import LiveScholarSource from app.services.settings import application as user_settings_service from app.settings import settings @@ -29,19 +29,6 @@ from app.settings import settings logger = logging.getLogger(__name__) -def _request_delay_floor_seconds() -> int: - return user_settings_service.resolve_request_delay_minimum(settings.ingestion_min_request_delay_seconds) - - -def _effective_request_delay_seconds(value: int | None) -> int: - floor = _request_delay_floor_seconds() - try: - parsed = int(value) if value is not None else floor - except (TypeError, ValueError): - parsed = floor - return max(floor, parsed) - - @dataclass(frozen=True) class _AutoRunCandidate: user_id: int @@ -195,7 +182,12 @@ class SchedulerService: return _AutoRunCandidate( user_id=int(user_id), run_interval_minutes=int(run_interval_minutes), - request_delay_seconds=_effective_request_delay_seconds(request_delay_seconds), + request_delay_seconds=effective_request_delay_seconds( + request_delay_seconds, + floor=user_settings_service.resolve_request_delay_minimum( + settings.ingestion_min_request_delay_seconds + ), + ), cooldown_until=cooldown_until, cooldown_reason=(str(cooldown_reason).strip() if cooldown_reason else None), ) diff --git a/app/services/publications/pdf_queue.py b/app/services/publications/pdf_queue.py index eef0d61..25f2437 100644 --- a/app/services/publications/pdf_queue.py +++ b/app/services/publications/pdf_queue.py @@ -2,16 +2,24 @@ from __future__ import annotations import logging from dataclasses import dataclass -from datetime import UTC, datetime +from datetime import datetime from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.db.models import ( PublicationPdfJob, - PublicationPdfJobEvent, User, ) +from app.services.publications.pdf_queue_common import ( + PDF_STATUS_FAILED, + PDF_STATUS_QUEUED, + PDF_STATUS_RESOLVED, + PDF_STATUS_RUNNING, + event_row, + queued_job, + utcnow, +) from app.services.publications.pdf_queue_queries import ( missing_pdf_candidates, retry_item_for_publication_id, @@ -21,10 +29,6 @@ from app.services.publications.types import PublicationListItem from app.settings import settings PDF_STATUS_UNTRACKED = "untracked" -PDF_STATUS_QUEUED = "queued" -PDF_STATUS_RUNNING = "running" -PDF_STATUS_RESOLVED = "resolved" -PDF_STATUS_FAILED = "failed" PDF_EVENT_QUEUED = "queued" @@ -43,10 +47,6 @@ class PdfBulkQueueResult: queued_count: int -def _utcnow() -> datetime: - return datetime.now(UTC) - - def _publication_ids(rows: list[PublicationListItem]) -> list[int]: return sorted({row.publication_id for row in rows}) @@ -122,7 +122,7 @@ def _cooldown_active( ) -> bool: if last_attempt_at is None: return False - elapsed = (_utcnow() - last_attempt_at).total_seconds() + elapsed = (utcnow() - last_attempt_at).total_seconds() return elapsed < _retry_interval_seconds_for_attempt_count(int(attempt_count)) @@ -147,37 +147,8 @@ def _can_enqueue_job( ) -def _event_row( - *, - publication_id: int, - user_id: int | None, - event_type: str, - status: str | None, -) -> PublicationPdfJobEvent: - return PublicationPdfJobEvent( - publication_id=publication_id, - user_id=user_id, - event_type=event_type, - status=status, - ) - - -def _queued_job( - *, - publication_id: int, - user_id: int, -) -> PublicationPdfJob: - now = _utcnow() - return PublicationPdfJob( - publication_id=publication_id, - status=PDF_STATUS_QUEUED, - queued_at=now, - last_requested_by_user_id=user_id, - ) - - def _mark_job_queued(job: PublicationPdfJob, *, user_id: int) -> None: - now = _utcnow() + now = utcnow() job.status = PDF_STATUS_QUEUED job.queued_at = now job.last_requested_by_user_id = user_id @@ -236,13 +207,13 @@ async def _enqueue_rows( if not _can_enqueue_job(job, force_retry=force_retry): continue if job is None: - job = _queued_job(publication_id=row.publication_id, user_id=user_id) + job = queued_job(publication_id=row.publication_id, user_id=user_id) jobs[row.publication_id] = job db_session.add(job) else: _mark_job_queued(job, user_id=user_id) db_session.add( - _event_row( + event_row( publication_id=row.publication_id, user_id=user_id, event_type=PDF_EVENT_QUEUED, diff --git a/app/services/publications/pdf_queue_common.py b/app/services/publications/pdf_queue_common.py new file mode 100644 index 0000000..46c2143 --- /dev/null +++ b/app/services/publications/pdf_queue_common.py @@ -0,0 +1,49 @@ +from __future__ import annotations + +from datetime import UTC, datetime + +from app.db.models import PublicationPdfJob, PublicationPdfJobEvent + +PDF_STATUS_QUEUED = "queued" +PDF_STATUS_RUNNING = "running" +PDF_STATUS_RESOLVED = "resolved" +PDF_STATUS_FAILED = "failed" + + +def utcnow() -> datetime: + return datetime.now(UTC) + + +def event_row( + *, + publication_id: int, + user_id: int | None, + event_type: str, + status: str | None, + source: str | None = None, + failure_reason: str | None = None, + message: str | None = None, +) -> PublicationPdfJobEvent: + return PublicationPdfJobEvent( + publication_id=publication_id, + user_id=user_id, + event_type=event_type, + status=status, + source=source, + failure_reason=failure_reason, + message=message, + ) + + +def queued_job( + *, + publication_id: int, + user_id: int, +) -> PublicationPdfJob: + now = utcnow() + return PublicationPdfJob( + publication_id=publication_id, + status=PDF_STATUS_QUEUED, + queued_at=now, + last_requested_by_user_id=user_id, + ) diff --git a/app/services/publications/pdf_queue_resolution.py b/app/services/publications/pdf_queue_resolution.py index d2d0fea..2e02a65 100644 --- a/app/services/publications/pdf_queue_resolution.py +++ b/app/services/publications/pdf_queue_resolution.py @@ -3,10 +3,19 @@ from __future__ import annotations import asyncio import logging -from app.db.models import Publication, PublicationPdfJob, PublicationPdfJobEvent +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 ( + PDF_STATUS_FAILED, + PDF_STATUS_QUEUED, + PDF_STATUS_RESOLVED, + PDF_STATUS_RUNNING, + event_row, + queued_job, + utcnow, +) from app.services.publications.pdf_resolution_pipeline import ( resolve_publication_pdf_outcome_for_row, ) @@ -17,11 +26,6 @@ from app.services.unpaywall.application import ( ) from app.settings import settings -PDF_STATUS_QUEUED = "queued" -PDF_STATUS_RUNNING = "running" -PDF_STATUS_RESOLVED = "resolved" -PDF_STATUS_FAILED = "failed" - PDF_EVENT_ATTEMPT_STARTED = "attempt_started" PDF_EVENT_RESOLVED = "resolved" PDF_EVENT_FAILED = "failed" @@ -30,47 +34,6 @@ logger = logging.getLogger(__name__) _scheduled_tasks: set[asyncio.Task[None]] = set() -def _utcnow(): - from datetime import UTC, datetime - - return datetime.now(UTC) - - -def _event_row( - *, - publication_id: int, - user_id: int | None, - event_type: str, - status: str | None, - source: str | None = None, - failure_reason: str | None = None, - message: str | None = None, -) -> PublicationPdfJobEvent: - return PublicationPdfJobEvent( - publication_id=publication_id, - user_id=user_id, - event_type=event_type, - status=status, - source=source, - failure_reason=failure_reason, - message=message, - ) - - -def _queued_job( - *, - publication_id: int, - user_id: int, -) -> PublicationPdfJob: - now = _utcnow() - return PublicationPdfJob( - publication_id=publication_id, - status=PDF_STATUS_QUEUED, - queued_at=now, - last_requested_by_user_id=user_id, - ) - - async def _mark_attempt_started( *, publication_id: int, @@ -80,13 +43,13 @@ async def _mark_attempt_started( async with session_factory() 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) + job = queued_job(publication_id=publication_id, user_id=user_id) db_session.add(job) job.status = PDF_STATUS_RUNNING - job.last_attempt_at = _utcnow() + job.last_attempt_at = utcnow() job.attempt_count = int(job.attempt_count) + 1 db_session.add( - _event_row( + event_row( publication_id=publication_id, user_id=user_id, event_type=PDF_EVENT_ATTEMPT_STARTED, @@ -142,7 +105,7 @@ def _apply_job_outcome(job: PublicationPdfJob, *, outcome: OaResolutionOutcome) job.last_source = outcome.source if outcome.pdf_url: job.status = PDF_STATUS_RESOLVED - job.resolved_at = _utcnow() + job.resolved_at = utcnow() job.last_failure_reason = None job.last_failure_detail = None return @@ -178,7 +141,7 @@ async def _persist_outcome( _apply_job_outcome(job, outcome=outcome) event_type, status = _result_event(outcome) db_session.add( - _event_row( + event_row( publication_id=publication_id, user_id=user_id, event_type=event_type, diff --git a/tests/unit/test_publication_pdf_queue_policy.py b/tests/unit/test_publication_pdf_queue_policy.py index ffc78d7..ca909a6 100644 --- a/tests/unit/test_publication_pdf_queue_policy.py +++ b/tests/unit/test_publication_pdf_queue_policy.py @@ -48,7 +48,7 @@ def _row( def test_pdf_queue_auto_enqueue_blocks_recent_attempt(monkeypatch: pytest.MonkeyPatch) -> None: now = datetime(2026, 2, 21, 12, 0, tzinfo=UTC) - monkeypatch.setattr(pdf_queue, "_utcnow", lambda: now) + monkeypatch.setattr(pdf_queue, "utcnow", lambda: now) monkeypatch.setattr(pdf_queue, "_auto_retry_first_interval_seconds", lambda: 3_600) monkeypatch.setattr(pdf_queue, "_auto_retry_interval_seconds", lambda: 86_400) monkeypatch.setattr(pdf_queue, "_auto_retry_max_attempts", lambda: 3) @@ -62,7 +62,7 @@ def test_pdf_queue_auto_enqueue_blocks_recent_attempt(monkeypatch: pytest.Monkey def test_pdf_queue_auto_enqueue_blocks_recent_first_retry(monkeypatch: pytest.MonkeyPatch) -> None: now = datetime(2026, 2, 21, 12, 0, tzinfo=UTC) - monkeypatch.setattr(pdf_queue, "_utcnow", lambda: now) + monkeypatch.setattr(pdf_queue, "utcnow", lambda: now) monkeypatch.setattr(pdf_queue, "_auto_retry_first_interval_seconds", lambda: 3_600) monkeypatch.setattr(pdf_queue, "_auto_retry_interval_seconds", lambda: 86_400) monkeypatch.setattr(pdf_queue, "_auto_retry_max_attempts", lambda: 3) @@ -76,7 +76,7 @@ def test_pdf_queue_auto_enqueue_blocks_recent_first_retry(monkeypatch: pytest.Mo def test_pdf_queue_auto_enqueue_blocks_after_max_attempts(monkeypatch: pytest.MonkeyPatch) -> None: now = datetime(2026, 2, 21, 12, 0, tzinfo=UTC) - monkeypatch.setattr(pdf_queue, "_utcnow", lambda: now) + monkeypatch.setattr(pdf_queue, "utcnow", lambda: now) monkeypatch.setattr(pdf_queue, "_auto_retry_first_interval_seconds", lambda: 3_600) monkeypatch.setattr(pdf_queue, "_auto_retry_interval_seconds", lambda: 86_400) monkeypatch.setattr(pdf_queue, "_auto_retry_max_attempts", lambda: 3) @@ -90,7 +90,7 @@ def test_pdf_queue_auto_enqueue_blocks_after_max_attempts(monkeypatch: pytest.Mo def test_pdf_queue_auto_enqueue_blocks_second_retry_within_day(monkeypatch: pytest.MonkeyPatch) -> None: now = datetime(2026, 2, 21, 12, 0, tzinfo=UTC) - monkeypatch.setattr(pdf_queue, "_utcnow", lambda: now) + monkeypatch.setattr(pdf_queue, "utcnow", lambda: now) monkeypatch.setattr(pdf_queue, "_auto_retry_first_interval_seconds", lambda: 3_600) monkeypatch.setattr(pdf_queue, "_auto_retry_interval_seconds", lambda: 86_400) monkeypatch.setattr(pdf_queue, "_auto_retry_max_attempts", lambda: 3) @@ -106,7 +106,7 @@ def test_pdf_queue_manual_requeue_bypasses_cooldown_and_max_attempts( monkeypatch: pytest.MonkeyPatch, ) -> None: now = datetime(2026, 2, 21, 12, 0, tzinfo=UTC) - monkeypatch.setattr(pdf_queue, "_utcnow", lambda: now) + monkeypatch.setattr(pdf_queue, "utcnow", lambda: now) monkeypatch.setattr(pdf_queue, "_auto_retry_first_interval_seconds", lambda: 3_600) monkeypatch.setattr(pdf_queue, "_auto_retry_interval_seconds", lambda: 86_400) monkeypatch.setattr(pdf_queue, "_auto_retry_max_attempts", lambda: 3) diff --git a/tests/unit/test_request_delay_policy.py b/tests/unit/test_request_delay_policy.py index 680e53d..411ae3c 100644 --- a/tests/unit/test_request_delay_policy.py +++ b/tests/unit/test_request_delay_policy.py @@ -4,6 +4,7 @@ from datetime import UTC, datetime from app.services.ingestion import scheduler as scheduler_module from app.services.ingestion.application import ScholarIngestionService +from app.services.ingestion.queue_runner import effective_request_delay_seconds from app.settings import settings @@ -25,9 +26,9 @@ def test_ingestion_effective_request_delay_respects_policy_minimum() -> None: def test_scheduler_effective_request_delay_respects_policy_minimum() -> None: previous = _set_policy_minimum(9) try: - assert scheduler_module._effective_request_delay_seconds(None) == 9 - assert scheduler_module._effective_request_delay_seconds(6) == 9 - assert scheduler_module._effective_request_delay_seconds(11) == 11 + assert effective_request_delay_seconds(None, floor=9) == 9 + assert effective_request_delay_seconds(6, floor=9) == 9 + assert effective_request_delay_seconds(11, floor=9) == 11 finally: object.__setattr__(settings, "ingestion_min_request_delay_seconds", previous)