openalex cooldown #45

Merged
JustinZeus merged 2 commits from bugfix/broken-import into main 2026-03-03 16:17:39 +01:00
3 changed files with 21 additions and 1 deletions
Showing only changes of commit a490e14126 - Show all commits

View file

@ -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 --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 list --limit 10 --json databaseId,status,conclusion,name,event,headBranch)",
"Bash(gh run view 22554585841 --log-failed)", "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": [ "additionalDirectories": [
"/home/jv/src/personal/playground/scholar_scraper/scholar-scraper/frontend/src" "/home/jv/src/personal/playground/scholar_scraper/scholar-scraper/frontend/src"

View file

@ -289,6 +289,10 @@ class SchedulerService:
async def _drain_pdf_queue(self) -> None: async def _drain_pdf_queue(self) -> None:
from app.services.publications.pdf_queue import drain_ready_jobs 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() session_factory = get_session_factory()
async with session_factory() as session: async with session_factory() as session:

View file

@ -2,6 +2,7 @@ from __future__ import annotations
import asyncio import asyncio
import logging import logging
from datetime import UTC, datetime, timedelta
from app.db.models import Publication, PublicationPdfJob from app.db.models import Publication, PublicationPdfJob
from app.db.session import get_session_factory from app.db.session import get_session_factory
@ -29,8 +30,20 @@ PDF_EVENT_ATTEMPT_STARTED = "attempt_started"
PDF_EVENT_RESOLVED = "resolved" PDF_EVENT_RESOLVED = "resolved"
PDF_EVENT_FAILED = "failed" PDF_EVENT_FAILED = "failed"
_BUDGET_COOLDOWN_MINUTES = 15
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
_scheduled_tasks: set[asyncio.Task[None]] = set() _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( async def _mark_attempt_started(
@ -233,11 +246,13 @@ async def _run_resolution_task(
detail="arXiv temporarily disabled for remaining batch after rate limit", detail="arXiv temporarily disabled for remaining batch after rate limit",
) )
except OpenAlexBudgetExhaustedError: except OpenAlexBudgetExhaustedError:
_enter_budget_cooldown()
structured_log( structured_log(
logger, logger,
"warning", "warning",
"pdf_queue.budget_exhausted", "pdf_queue.budget_exhausted",
detail="Stopping PDF resolution batch — OpenAlex daily budget exhausted", detail="Stopping PDF resolution batch — OpenAlex daily budget exhausted",
cooldown_minutes=_BUDGET_COOLDOWN_MINUTES,
) )
break break
except Exception: except Exception: