feat: serialize official record imports

apps/api/app/importer.py

@@ -3,11 +3,12 @@ from __future__ import annotations
 import hashlib
 import re
 import time as time_module
+from contextlib import contextmanager
 from dataclasses import asdict, dataclass
 from datetime import date, datetime, time, timezone
 
 import httpx
-from sqlalchemy import select
+from sqlalchemy import select, text
 from sqlalchemy.orm import Session
 from rf4_research.official_parser import RecordsContractError, parse_official_records
 
@@ -24,6 +25,10 @@ class ImportSourceError(ValueError):
     pass
 
 
+class ImportAlreadyRunning(RuntimeError):
+    pass
+
+
 @dataclass(frozen=True, slots=True)
 class RawRecord:
     region: str
@@ -105,7 +110,42 @@ def fetch_records(
     raise AssertionError("unreachable")
 
 
+def _lock_key(url: str, region: str, category: str) -> int:
+    digest = hashlib.sha256(f"{url}|{region.upper()}|{category}".encode()).digest()
+    return int.from_bytes(digest[:8], byteorder="big", signed=True)
+
+
+@contextmanager
+def _official_import_lock(session: Session, *, url: str, region: str, category: str):
+    bind = session.get_bind()
+    if bind.dialect.name != "postgresql":
+        yield
+        return
+    connection = bind.connect()
+    key = _lock_key(url, region, category)
+    try:
+        acquired = bool(connection.scalar(text("SELECT pg_try_advisory_lock(:key)"), {"key": key}))
+    except Exception:
+        connection.close()
+        raise
+    if not acquired:
+        connection.close()
+        raise ImportAlreadyRunning("official import is already running for this source and category")
+    try:
+        yield
+    finally:
+        try:
+            connection.execute(text("SELECT pg_advisory_unlock(:key)"), {"key": key})
+        finally:
+            connection.close()
+
+
 def import_records(session: Session, *, url: str, region: str, category: str, html: str | None = None) -> OfficialRecordImport:
+    with _official_import_lock(session, url=url, region=region, category=category):
+        return _import_records_locked(session, url=url, region=region, category=category, html=html)
+
+
+def _import_records_locked(session: Session, *, url: str, region: str, category: str, html: str | None = None) -> OfficialRecordImport:
     run = OfficialRecordImport(started_at=datetime.now(timezone.utc), status=ImportStatus.running, source_url=url, rows_seen=0, rows_created=0, rows_updated=0)
     session.add(run)
     session.commit()

apps/api/app/main.py

@@ -22,7 +22,7 @@ from .activity import activity_rows
 from .database import get_session
 from .config import settings
 from .community_review import ExternalReviewError, map_observation, publish_observation, reject_observation
-from .importer import ImportSourceError, import_records, normalize
+from .importer import ImportAlreadyRunning, ImportSourceError, import_records, normalize
 from .logging_config import configure_logging
 from .models import Bait, BaitKind, CatchReport, ExternalObservation, Fish, ModerationEvent, ModerationStatus, OfficialRecordImport, SourceType, Spot, SubmissionAttempt, Waterbody
 from .readiness import readiness_report
@@ -200,6 +200,8 @@ def admin_start_official_import(db: Db, _: Annotated[str, Depends(_admin)]) -> O
             region=settings.official_records_region,
             category=settings.official_records_category,
         )
+    except ImportAlreadyRunning as exc:
+        raise HTTPException(status_code=409, detail=str(exc)) from exc
     except (ImportSourceError, httpx.HTTPError) as exc:
         raise HTTPException(status_code=502, detail=f"official records import failed: {exc}") from exc
 

apps/api/app/scheduler.py

@@ -9,7 +9,7 @@ from sqlalchemy.orm import Session
 
 from .config import settings
 from .database import SessionLocal
-from .importer import import_records
+from .importer import ImportAlreadyRunning, import_records
 from .logging_config import configure_logging
 from .models import OfficialRecordImport
 
@@ -36,12 +36,16 @@ def run_due_import() -> bool:
     with SessionLocal() as session:
         if not import_is_due(session):
             return False
-        run = import_records(
-            session,
-            url=settings.official_records_url,
-            region=settings.official_records_region,
-            category=settings.official_records_category,
-        )
+        try:
+            run = import_records(
+                session,
+                url=settings.official_records_url,
+                region=settings.official_records_region,
+                category=settings.official_records_category,
+            )
+        except ImportAlreadyRunning:
+            logger.info("official import skipped because it is already running", extra={"event": "official_import_locked"})
+            return False
         logger.info(
             "official import completed",
             extra={

apps/api/tests/test_api.py

@@ -10,6 +10,7 @@ from sqlalchemy.pool import StaticPool
 
 from app.database import Base, get_session
 from app.community_importer import stage_observations
+from app.importer import ImportAlreadyRunning
 from app.main import app
 from app.models import Bait, BaitKind, CatchReport, ExternalEntityAlias, ExternalObservation, Fish, ImportStatus, ModerationEvent, ModerationStatus, OfficialRecordImport, SourceType, Spot, Waterbody
 
@@ -186,6 +187,12 @@ def test_admin_can_start_and_list_official_import(monkeypatch) -> None:
     listed = client.get("/api/v1/admin/imports?limit=1&offset=0", headers=headers)
     assert listed.status_code == 200
     assert listed.json()[0]["id"] == started.json()["id"]
+    def busy_import(*args, **kwargs):
+        raise ImportAlreadyRunning("official import is already running")
+
+    monkeypatch.setattr("app.main.import_records", busy_import)
+    conflict = client.post("/api/v1/admin/imports/official-records", headers=headers)
+    assert conflict.status_code == 409
 
 
 def test_pending_report_accepts_one_validated_screenshot(monkeypatch) -> None:

apps/api/tests/test_import_lock_postgres.py

@@ -0,0 +1,27 @@
+import os
+
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+from app.importer import ImportAlreadyRunning, _official_import_lock
+
+
+@pytest.mark.skipif(not os.environ.get("DATABASE_URL", "").startswith("postgresql"), reason="requires PostgreSQL")
+def test_postgresql_import_lock_blocks_only_same_source_category() -> None:
+    engine = create_engine(os.environ["DATABASE_URL"])
+    first = Session(engine)
+    second = Session(engine)
+    try:
+        with _official_import_lock(first, url="https://example.test/records", region="RU", category="records"):
+            with pytest.raises(ImportAlreadyRunning):
+                with _official_import_lock(second, url="https://example.test/records", region="RU", category="records"):
+                    pass
+            with _official_import_lock(second, url="https://example.test/records", region="RU", category="weekly"):
+                pass
+        with _official_import_lock(second, url="https://example.test/records", region="RU", category="records"):
+            pass
+    finally:
+        first.close()
+        second.close()
+        engine.dispose()

apps/api/tests/test_importer.py

@@ -7,7 +7,7 @@ from sqlalchemy import create_engine, func, select
 from sqlalchemy.orm import Session
 
 from app.database import Base
-from app.importer import FetchResult, ImportSourceError, import_records, parse_html
+from app.importer import FetchResult, ImportAlreadyRunning, ImportSourceError, _lock_key, _official_import_lock, import_records, parse_html
 from app.models import CatchReport, ImportStatus, OfficialRecordImport, SourceType
 
 
@@ -15,6 +15,26 @@ FIXTURE = Path(__file__).parents[3] / "tests" / "fixtures" / "records_ru_sample.
 WEEKLY_FIXTURE = Path(__file__).parents[3] / "tests" / "fixtures" / "weekly_records_sample.html"
 
 
+def test_import_lock_is_stable_and_fails_closed_when_busy() -> None:
+    class Connection:
+        def scalar(self, statement, parameters):
+            assert "pg_try_advisory_lock" in str(statement)
+            assert parameters == {"key": _lock_key("https://example.test", "RU", "records")}
+            return False
+
+        def close(self):
+            self.closed = True
+
+    connection = Connection()
+    bind = type("Bind", (), {"dialect": type("Dialect", (), {"name": "postgresql"})(), "connect": lambda self: connection})()
+    session = type("Session", (), {"get_bind": lambda self: bind})()
+    assert _lock_key("https://example.test", "ru", "records") == _lock_key("https://example.test", "RU", "records")
+    with pytest.raises(ImportAlreadyRunning, match="already running"):
+        with _official_import_lock(session, url="https://example.test", region="RU", category="records"):
+            raise AssertionError("busy lock must not enter import")
+    assert connection.closed is True
+
+
 def test_parser_and_import_are_idempotent() -> None:
     html = FIXTURE.read_text(encoding="utf-8")
     parsed = parse_html(html, region="RU", category="records")