diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 9f39e29..f43fb27 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -21,6 +21,10 @@ jobs: python-version: "3.12" - uses: astral-sh/setup-uv@v4 - run: uv sync --extra dev + - name: Lint + run: uv run ruff check + - name: Test + run: uv run pytest - name: Semantic Release id: release run: uv run semantic-release publish diff --git a/app/api/routers/run_manual.py b/app/api/routers/run_manual.py index 0632e39..cdb5e99 100644 --- a/app/api/routers/run_manual.py +++ b/app/api/routers/run_manual.py @@ -1,7 +1,7 @@ from __future__ import annotations import logging -from typing import Any +from typing import Any, NoReturn from fastapi import Request from sqlalchemy.exc import IntegrityError @@ -215,7 +215,7 @@ async def execute_manual_run( raise_manual_failed(exc=exc, user_id=user_id) -def raise_manual_blocked_safety(*, exc, user_id: int) -> None: +def raise_manual_blocked_safety(*, exc, user_id: int) -> NoReturn: structured_log( logger, "info", @@ -233,7 +233,7 @@ def raise_manual_blocked_safety(*, exc, user_id: int) -> None: ) from exc -def raise_manual_failed(*, exc: Exception, user_id: int) -> None: +def raise_manual_failed(*, exc: Exception, user_id: int) -> NoReturn: structured_log( logger, "exception", diff --git a/app/api/schemas/admin.py b/app/api/schemas/admin.py index f5f73b5..932f93c 100644 --- a/app/api/schemas/admin.py +++ b/app/api/schemas/admin.py @@ -35,7 +35,7 @@ class AdminUsersListEnvelope(BaseModel): class AdminUserCreateRequest(BaseModel): email: str = Field(max_length=254) - password: str = Field(max_length=128) + password: str = Field(min_length=8, max_length=128) is_admin: bool = False model_config = ConfigDict(extra="forbid") @@ -55,7 +55,7 @@ class AdminUserEnvelope(BaseModel): class AdminResetPasswordRequest(BaseModel): - new_password: str = Field(max_length=128) + new_password: str = Field(min_length=8, max_length=128) model_config = ConfigDict(extra="forbid") @@ -64,7 +64,7 @@ class AdminScholarHttpSettingsData(BaseModel): user_agent: str rotate_user_agent: bool accept_language: str - cookie: str + cookie: str | None = None model_config = ConfigDict(extra="forbid") @@ -80,7 +80,7 @@ class AdminScholarHttpSettingsUpdateRequest(BaseModel): user_agent: str rotate_user_agent: bool accept_language: str - cookie: str + cookie: str | None = None model_config = ConfigDict(extra="forbid") diff --git a/app/api/schemas/auth.py b/app/api/schemas/auth.py index f160b24..14efa57 100644 --- a/app/api/schemas/auth.py +++ b/app/api/schemas/auth.py @@ -45,7 +45,7 @@ class CsrfBootstrapEnvelope(BaseModel): class LoginRequest(BaseModel): email: str = Field(max_length=254) - password: str = Field(max_length=128) + password: str = Field(min_length=8, max_length=128) model_config = ConfigDict(extra="forbid") @@ -66,8 +66,8 @@ class LoginEnvelope(BaseModel): class ChangePasswordRequest(BaseModel): - current_password: str = Field(max_length=128) - new_password: str = Field(max_length=128) - confirm_password: str = Field(max_length=128) + current_password: str = Field(min_length=8, max_length=128) + new_password: str = Field(min_length=8, max_length=128) + confirm_password: str = Field(min_length=8, max_length=128) model_config = ConfigDict(extra="forbid") diff --git a/app/main.py b/app/main.py index feda0c8..f8610e1 100644 --- a/app/main.py +++ b/app/main.py @@ -81,6 +81,26 @@ async def lifespan(_: FastAPI): error=str(exc), ) + try: + session_factory = get_session_factory() + async with session_factory() as session: + await session.execute( + text( + "UPDATE publication_pdf_jobs SET status = 'queued'" + " WHERE status = 'running'" + " AND (last_attempt_at IS NULL OR last_attempt_at < NOW() - INTERVAL '10 minutes')" + ) + ) + await session.commit() + structured_log(logger, "info", "app.startup_stuck_pdf_jobs_recovered") + except Exception as exc: + structured_log( + logger, + "error", + "app.startup_stuck_pdf_jobs_recovery_failed", + error=str(exc), + ) + await scheduler_service.start() yield await scheduler_service.stop() diff --git a/app/services/arxiv/rate_limit.py b/app/services/arxiv/rate_limit.py index 1f3ab3f..8d1a5a1 100644 --- a/app/services/arxiv/rate_limit.py +++ b/app/services/arxiv/rate_limit.py @@ -53,7 +53,7 @@ async def get_arxiv_cooldown_status(*, now_utc: datetime | None = None) -> Arxiv result = await db_session.execute( select(ArxivRuntimeState.cooldown_until).where(ArxivRuntimeState.state_key == ARXIV_RUNTIME_STATE_KEY) ) - cooldown_until = _normalize_datetime(result.scalar_one_or_none()) + cooldown_until = _normalize_datetime(result.scalar_one_or_none()) remaining_seconds = _cooldown_remaining_seconds(cooldown_until, now_utc=timestamp) return ArxivCooldownStatus( is_active=remaining_seconds > 0.0, @@ -68,26 +68,31 @@ async def _run_serialized_fetch( source_path: str, ) -> tuple[httpx.Response, bool]: session_factory = get_session_factory() - async with session_factory() as db_session, db_session.begin(): - await _acquire_arxiv_lock(db_session) - runtime_state = await _load_runtime_state_for_update(db_session) - wait_seconds = await _wait_for_allowed_slot_or_raise( - runtime_state, - source_path=source_path, - ) - runtime_state.next_allowed_at = datetime.now(UTC) + timedelta(seconds=_min_interval_seconds()) + async with session_factory() as lock_session: + await _acquire_arxiv_lock(lock_session) + try: + async with session_factory() as db_session, db_session.begin(): + runtime_state = await _load_runtime_state_for_update(db_session) + wait_seconds = await _wait_for_allowed_slot_or_raise( + runtime_state, + source_path=source_path, + ) + runtime_state.next_allowed_at = datetime.now(UTC) + timedelta(seconds=_min_interval_seconds()) - if wait_seconds > 0: - await asyncio.sleep(wait_seconds) - response = await fetch() + if wait_seconds > 0: + await asyncio.sleep(wait_seconds) + response = await fetch() - async with session_factory() as db_session, db_session.begin(): - runtime_state = await _load_runtime_state_for_update(db_session) - hit_rate_limit = _record_post_response_state( - runtime_state, - response_status=int(response.status_code), - source_path=source_path, - ) + async with session_factory() as db_session, db_session.begin(): + runtime_state = await _load_runtime_state_for_update(db_session) + hit_rate_limit = _record_post_response_state( + runtime_state, + response_status=int(response.status_code), + source_path=source_path, + ) + cooldown_until_value = runtime_state.cooldown_until + finally: + await _release_arxiv_lock(lock_session) structured_log( logger, "info", @@ -95,7 +100,7 @@ async def _run_serialized_fetch( status_code=int(response.status_code), wait_seconds=wait_seconds, cooldown_remaining_seconds=_cooldown_remaining_seconds( - runtime_state.cooldown_until, now_utc=datetime.now(UTC) + cooldown_until_value, now_utc=datetime.now(UTC) ), source_path=source_path, ) @@ -104,7 +109,17 @@ async def _run_serialized_fetch( async def _acquire_arxiv_lock(db_session: AsyncSession) -> None: await db_session.execute( - text("SELECT pg_advisory_xact_lock(:namespace, :lock_key)"), + text("SELECT pg_advisory_lock(:namespace, :lock_key)"), + { + "namespace": ARXIV_RATE_LIMIT_LOCK_NAMESPACE, + "lock_key": ARXIV_RATE_LIMIT_LOCK_KEY, + }, + ) + + +async def _release_arxiv_lock(db_session: AsyncSession) -> None: + await db_session.execute( + text("SELECT pg_advisory_unlock(:namespace, :lock_key)"), { "namespace": ARXIV_RATE_LIMIT_LOCK_NAMESPACE, "lock_key": ARXIV_RATE_LIMIT_LOCK_KEY, diff --git a/app/services/crossref/application.py b/app/services/crossref/application.py index 7a2908b..a546a9f 100644 --- a/app/services/crossref/application.py +++ b/app/services/crossref/application.py @@ -20,7 +20,7 @@ if TYPE_CHECKING: from app.services.publications.types import PublicationListItem, UnreadPublicationItem TOKEN_RE = re.compile(r"[a-z0-9]+") -NON_ALNUM_RE = re.compile(r"[^a-z0-9\\s]+") +NON_ALNUM_RE = re.compile(r"[^a-z0-9\s]+") STOP_WORDS = {"the", "and", "for", "with", "from", "method", "study", "analysis"} _RATE_LOCK = threading.Lock() _LAST_REQUEST_AT = 0.0 @@ -35,8 +35,9 @@ def _rate_limit_wait(min_interval_seconds: float) -> None: with _RATE_LOCK: elapsed = time.monotonic() - _LAST_REQUEST_AT remaining = interval - elapsed - if remaining > 0: - time.sleep(remaining) + if remaining > 0: + time.sleep(remaining) + with _RATE_LOCK: _LAST_REQUEST_AT = time.monotonic() diff --git a/app/services/dbops/near_duplicate_repair.py b/app/services/dbops/near_duplicate_repair.py index 8d50570..70c902c 100644 --- a/app/services/dbops/near_duplicate_repair.py +++ b/app/services/dbops/near_duplicate_repair.py @@ -162,7 +162,10 @@ async def _complete_job( async def _fail_job(db_session: AsyncSession, *, job: DataRepairJob, error: Exception) -> None: + from sqlalchemy.orm import make_transient + await db_session.rollback() + make_transient(job) job.status = REPAIR_STATUS_FAILED job.error_text = str(error) job.finished_at = _utcnow() diff --git a/app/services/ingestion/application.py b/app/services/ingestion/application.py index cd5e6ea..1a6293e 100644 --- a/app/services/ingestion/application.py +++ b/app/services/ingestion/application.py @@ -401,6 +401,7 @@ class ScholarIngestionService: if run_to_fail: run_to_fail.status = RunStatus.FAILED run_to_fail.end_dt = datetime.now(UTC) + run_to_fail.error_log = run_to_fail.error_log or {} run_to_fail.error_log["terminal_exception"] = str(exc) await cleanup_session.commit() except Exception: diff --git a/app/services/ingestion/publication_upsert.py b/app/services/ingestion/publication_upsert.py index d2ba42b..ae7b1b8 100644 --- a/app/services/ingestion/publication_upsert.py +++ b/app/services/ingestion/publication_upsert.py @@ -176,45 +176,51 @@ async def upsert_profile_publications( seen_publication_ids: set[int] = set() discovered_count = 0 - for candidate in publications: - publication = await resolve_publication(db_session, candidate) - if publication.id in seen_publication_ids: - continue - seen_publication_ids.add(publication.id) + try: + for candidate in publications: + publication = await resolve_publication(db_session, candidate) + if publication.id in seen_publication_ids: + continue + seen_publication_ids.add(publication.id) - link_result = await db_session.execute( - select(ScholarPublication).where( - ScholarPublication.scholar_profile_id == scholar.id, - ScholarPublication.publication_id == publication.id, + link_result = await db_session.execute( + select(ScholarPublication).where( + ScholarPublication.scholar_profile_id == scholar.id, + ScholarPublication.publication_id == publication.id, + ) ) - ) - link = link_result.scalar_one_or_none() - if link is not None: - continue + link = link_result.scalar_one_or_none() + if link is not None: + continue - link = ScholarPublication( - scholar_profile_id=scholar.id, - publication_id=publication.id, - is_read=False, - first_seen_run_id=run.id, - ) - db_session.add(link) - discovered_count += 1 + link = ScholarPublication( + scholar_profile_id=scholar.id, + publication_id=publication.id, + is_read=False, + first_seen_run_id=run.id, + ) + db_session.add(link) + discovered_count += 1 - await commit_discovered_publication( - db_session, - run=run, - scholar=scholar, - publication=publication, - ) + await flush_discovered_publication( + db_session, + run=run, + scholar=scholar, + publication=publication, + ) - if not scholar.baseline_completed: - scholar.baseline_completed = True + if not scholar.baseline_completed: + scholar.baseline_completed = True + + await db_session.commit() + except Exception: + await db_session.rollback() + raise return discovered_count -async def commit_discovered_publication( +async def flush_discovered_publication( db_session: AsyncSession, *, run: CrawlRun, @@ -222,7 +228,7 @@ async def commit_discovered_publication( publication: Publication, ) -> None: run.new_pub_count = int(run.new_pub_count or 0) + 1 - await db_session.commit() + await db_session.flush() await run_events.publish( run_id=run.id, event_type="publication_discovered", diff --git a/app/services/ingestion/queue.py b/app/services/ingestion/queue.py index 2016f63..8bdddbc 100644 --- a/app/services/ingestion/queue.py +++ b/app/services/ingestion/queue.py @@ -196,6 +196,7 @@ async def list_due_jobs( IngestionQueueItem.id.asc(), ) .limit(limit) + .with_for_update(skip_locked=True) ) rows = list(result.scalars().all()) jobs: list[ContinuationQueueJob] = [] diff --git a/app/services/publications/pdf_queue_common.py b/app/services/publications/pdf_queue_common.py index 9eb95ed..0bff88b 100644 --- a/app/services/publications/pdf_queue_common.py +++ b/app/services/publications/pdf_queue_common.py @@ -47,4 +47,5 @@ def queued_job( status=PDF_STATUS_QUEUED, queued_at=now, last_requested_by_user_id=user_id, + attempt_count=0, ) diff --git a/app/services/publications/pdf_queue_resolution.py b/app/services/publications/pdf_queue_resolution.py index aec3f74..ec4422e 100644 --- a/app/services/publications/pdf_queue_resolution.py +++ b/app/services/publications/pdf_queue_resolution.py @@ -46,7 +46,7 @@ async def _mark_attempt_started( db_session.add(job) job.status = PDF_STATUS_RUNNING job.last_attempt_at = utcnow() - job.attempt_count = int(job.attempt_count) + 1 + job.attempt_count = int(job.attempt_count or 0) + 1 db_session.add( event_row( publication_id=publication_id, @@ -240,6 +240,26 @@ async def _run_resolution_task( detail="Stopping PDF resolution batch — OpenAlex daily budget exhausted", ) break + except Exception: + structured_log( + logger, + "exception", + "pdf_queue.row_failed", + publication_id=row.publication_id, + ) + try: + await _persist_outcome( + publication_id=row.publication_id, + user_id=user_id, + outcome=_failed_outcome(row=row), + ) + except Exception: + structured_log( + logger, + "exception", + "pdf_queue.row_fail_persist_error", + publication_id=row.publication_id, + ) def _register_task(task: asyncio.Task[None]) -> None: diff --git a/app/services/runs/events.py b/app/services/runs/events.py index 17baa62..dbcac2d 100644 --- a/app/services/runs/events.py +++ b/app/services/runs/events.py @@ -8,6 +8,8 @@ from app.logging_utils import structured_log logger = logging.getLogger(__name__) +_SUBSCRIBER_QUEUE_MAXSIZE = 256 + class RunEventPublisher: def __init__(self) -> None: @@ -17,7 +19,7 @@ class RunEventPublisher: def subscribe(self, run_id: int) -> asyncio.Queue: if run_id not in self._subscribers: self._subscribers[run_id] = set() - queue: asyncio.Queue[Any] = asyncio.Queue() + queue: asyncio.Queue[Any] = asyncio.Queue(maxsize=_SUBSCRIBER_QUEUE_MAXSIZE) self._subscribers[run_id].add(queue) structured_log( logger, @@ -51,6 +53,10 @@ class RunEventPublisher: "runs.event_subscriber_queue_full", run_id=run_id, ) + self._subscribers[run_id].discard(queue) + + async def publish_run_complete(self, run_id: int) -> None: + await self.publish(run_id, "run_complete", {}) run_events = RunEventPublisher() @@ -62,8 +68,11 @@ async def event_generator(run_id: int) -> AsyncGenerator[str, None]: while True: # Wait for a new event message = await queue.get() - # Server-Sent Events format: "event: \ndata: \n\n" event_type = message["type"] + if event_type == "run_complete": + yield f"event: {event_type}\ndata: {{}}\n\n" + break + # Server-Sent Events format: "event: \ndata: \n\n" data_str = json.dumps(message["data"]) yield f"event: {event_type}\ndata: {data_str}\n\n" except asyncio.CancelledError: diff --git a/frontend/src/features/publications/components/PublicationTableRow.test.ts b/frontend/src/features/publications/components/PublicationTableRow.test.ts index 3a1110d..e3d8915 100644 --- a/frontend/src/features/publications/components/PublicationTableRow.test.ts +++ b/frontend/src/features/publications/components/PublicationTableRow.test.ts @@ -13,7 +13,7 @@ function buildItem(overrides: Partial = {}): PublicationItem { citation_count: 42, pub_url: "https://example.com/pub", pdf_url: null, - pdf_status: null, + pdf_status: "untracked", is_read: false, is_favorite: false, is_new_in_latest_run: true, diff --git a/frontend/src/features/publications/composables/usePublicationData.ts b/frontend/src/features/publications/composables/usePublicationData.ts index e48209e..1ce8d11 100644 --- a/frontend/src/features/publications/composables/usePublicationData.ts +++ b/frontend/src/features/publications/composables/usePublicationData.ts @@ -1,4 +1,4 @@ -import { computed, ref, watch } from "vue"; +import { computed, onScopeDispose, ref, watch } from "vue"; import { useRoute, useRouter } from "vue-router"; import { listPublications, @@ -271,6 +271,10 @@ export function usePublicationData() { }, 300); }); + onScopeDispose(() => { + if (searchDebounceTimer !== null) clearTimeout(searchDebounceTimer); + }); + // --- Run-triggered refresh watcher --- watch( diff --git a/frontend/src/features/settings/SettingsAdminPanel.vue b/frontend/src/features/settings/SettingsAdminPanel.vue index 585dd6a..b306d1e 100644 --- a/frontend/src/features/settings/SettingsAdminPanel.vue +++ b/frontend/src/features/settings/SettingsAdminPanel.vue @@ -36,10 +36,8 @@ async function refreshForSection(): Promise { return; } if (props.section === SECTION_REPAIRS) { - await Promise.all([ - usersRef.value?.load(), - repairsRef.value?.load(), - ]); + await usersRef.value?.load(); + await repairsRef.value?.load(); return; } if (props.section === SECTION_PDF && pdfQueueRef.value) { diff --git a/frontend/src/features/settings/components/AdminPdfQueueSection.vue b/frontend/src/features/settings/components/AdminPdfQueueSection.vue index f1500ca..fdeb98a 100644 --- a/frontend/src/features/settings/components/AdminPdfQueueSection.vue +++ b/frontend/src/features/settings/components/AdminPdfQueueSection.vue @@ -7,6 +7,7 @@ import AppHelpHint from "@/components/ui/AppHelpHint.vue"; import AppRefreshButton from "@/components/ui/AppRefreshButton.vue"; import AppSelect from "@/components/ui/AppSelect.vue"; import AppTable from "@/components/ui/AppTable.vue"; +import RequestStateAlerts from "@/components/patterns/RequestStateAlerts.vue"; import { listAdminPdfQueue, requeueAdminPdfLookup, @@ -15,7 +16,7 @@ import { } from "@/features/admin_dbops"; import { useRequestState } from "@/composables/useRequestState"; -const { clearAlerts, assignError, setSuccess } = useRequestState(); +const { errorMessage, errorRequestId, successMessage, clearAlerts, assignError, setSuccess } = useRequestState(); const refreshingPdfQueue = ref(false); const requeueingPublicationId = ref(null); @@ -131,7 +132,17 @@ defineExpose({ load }); diff --git a/frontend/src/features/settings/components/AdminRepairsSection.vue b/frontend/src/features/settings/components/AdminRepairsSection.vue index 3f3a88a..05eab7a 100644 --- a/frontend/src/features/settings/components/AdminRepairsSection.vue +++ b/frontend/src/features/settings/components/AdminRepairsSection.vue @@ -9,6 +9,7 @@ import AppInput from "@/components/ui/AppInput.vue"; import AppRefreshButton from "@/components/ui/AppRefreshButton.vue"; import AppSelect from "@/components/ui/AppSelect.vue"; import AppTable from "@/components/ui/AppTable.vue"; +import RequestStateAlerts from "@/components/patterns/RequestStateAlerts.vue"; import { dropAllPublications, listAdminDbRepairJobs, @@ -34,7 +35,7 @@ const DROP_PUBLICATIONS_CONFIRM_TEXT = "DROP ALL PUBLICATIONS"; type RepairScopeMode = typeof SCOPE_SINGLE_USER | typeof SCOPE_ALL_USERS; -const { clearAlerts, assignError, setSuccess } = useRequestState(); +const { errorMessage, errorRequestId, successMessage, clearAlerts, assignError, setSuccess } = useRequestState(); const refreshingJobs = ref(false); const runningRepair = ref(false); @@ -246,6 +247,15 @@ defineExpose({ load });