diff --git a/backend/politrack/core/loader.py b/backend/politrack/core/loader.py index ed1cee4..dfdfba8 100644 --- a/backend/politrack/core/loader.py +++ b/backend/politrack/core/loader.py @@ -254,6 +254,7 @@ class Loader: VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ON CONFLICT (event_key) DO UPDATE SET result = EXCLUDED.result, raw_text = EXCLUDED.raw_text, roll_call = EXCLUDED.roll_call, item_title = EXCLUDED.item_title, part_vote = EXCLUDED.part_vote, + item_ref = EXCLUDED.item_ref, stage = EXCLUDED.stage, protocol_url = COALESCE(EXCLUDED.protocol_url, vote_event.protocol_url) RETURNING id""", (rec.key, chamber, rec.item_ref, rec.title, rec.stage, rec.voted_on, rec.result, rec.roll_call, rec.url, diff --git a/backend/politrack/sources/eu.py b/backend/politrack/sources/eu.py index f9760d4..cee3ace 100644 --- a/backend/politrack/sources/eu.py +++ b/backend/politrack/sources/eu.py @@ -200,7 +200,7 @@ class EuropeanParliament(Adapter): LD_JSON = "application/ld+json" EP_DECISIONS_PAGE = 500 -EP_REFERENCE = re.compile(r"^[AB]\d+-\d+/\d{4}$") +EP_REFERENCE = re.compile(r"^(?:RC-)?[A-C]\d+-\d+/\d{4}$") # A10-0272/2026, RC-B9-0253/2024, C9-0010/2024 EP_OUTCOME = {"ADOPTED": ("accepted", "angenommen", "adopted"), "REJECTED": ("rejected", "abgelehnt", "rejected")} EP_VOTERS = (("had_voter_favor", "for", "Stimmte für", "Voted for"), ("had_voter_against", "against", "Stimmte gegen", "Voted against"), @@ -233,7 +233,7 @@ class ParliamentVotes(Adapter): description = "Roll-call votes of Austrian MEPs in the European Parliament plenary, including amendments" schedule = "45 5 * * *" queue = "scrape" - parser_version = 2 # amendment and part votes + parser_version = 3 # 2: amendment and part votes; 3: references of joint motions and Council documents def start(self, ctx: RunContext) -> Iterator[Request]: ep_ids = [] diff --git a/backend/politrack/worker.py b/backend/politrack/worker.py index ca45189..d334a9e 100644 --- a/backend/politrack/worker.py +++ b/backend/politrack/worker.py @@ -9,21 +9,38 @@ One periodic task is generated per registered adapter/enricher from its `schedul Periodic tasks are deferred by every running worker and de-duplicated by Procrastinate; queueing locks prevent two runs of the same adapter at once. Workers can run on other machines that reach the DB. + +Procrastinate only remembers in memory when it last deferred a task, and skips a slot that is more than +ten minutes overdue. In November 2026 every daily and weekly source silently missed two days that way, +and jobs of killed workers stayed "doing" forever. The watchdog below makes both self-healing. """ from __future__ import annotations +import asyncio import logging +from datetime import datetime, timedelta, timezone +from croniter import croniter from procrastinate import App, PsycopgConnector +from procrastinate.exceptions import AlreadyEnqueued +from procrastinate.jobs import Status from politrack.config import get_settings from politrack.core import registry from politrack.core.runner import run +from politrack.db import connection logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s") +log = logging.getLogger("politrack.watchdog") app = App(connector=PsycopgConnector(conninfo=get_settings().database_url)) +TASKS: dict = {} +SCHEDULES: dict[str, str] = {} + +STALLED_AFTER = 300 # seconds without a worker heartbeat +OVERDUE_GRACE = timedelta(minutes=15) # normal deferral happens within a minute of the slot +ABANDONED_RUN = timedelta(hours=6) # CLI runs have no job; don't close them while they may still run def _make_task(name: str, queue: str, schedule: str | None) -> None: @@ -32,10 +49,65 @@ def _make_task(name: str, queue: str, schedule: str | None) -> None: job.__name__ = f"run_{name.replace('.', '_')}" task = app.task(name=f"politrack.run.{name}", queue=queue, queueing_lock=name, pass_context=False)(job) + TASKS[name] = task if schedule: + SCHEDULES[name] = schedule app.periodic(cron=schedule, periodic_id=name)(task) registry.load_builtin() for _name, _cls in {**registry.ADAPTERS, **registry.ENRICHERS}.items(): _make_task(_name, _cls.queue, _cls.schedule) + + +def overdue_tasks(now: datetime) -> list[str]: + """Scheduled tasks whose latest slot passed without a run, a waiting job or a running job.""" + due: list[str] = [] + with connection() as conn: + for name, schedule in SCHEDULES.items(): + slot = croniter(schedule, now).get_prev(datetime) + if now - slot < OVERDUE_GRACE: + continue + seen = conn.execute( + """SELECT EXISTS (SELECT 1 FROM run WHERE adapter = %(name)s AND started_at >= %(slot)s) + OR EXISTS (SELECT 1 FROM procrastinate_jobs WHERE task_name = %(task)s + AND status IN ('todo', 'doing')) AS seen""", + {"name": name, "slot": slot, "task": f"politrack.run.{name}"}, + ).fetchone()["seen"] + if not seen: + due.append(name) + return due + + +def close_abandoned_runs() -> int: + """Runs still marked running although no job for them is in progress (the worker was stopped).""" + with connection() as conn: + closed = conn.execute( + """UPDATE run r SET status = 'error', finished_at = now(), + error = 'abandoned: the worker stopped before the run finished (closed by the watchdog)' + WHERE r.status = 'running' AND r.started_at < now() - %s + AND NOT EXISTS (SELECT 1 FROM procrastinate_jobs j + WHERE j.status = 'doing' AND j.task_name = 'politrack.run.' || r.adapter)""", + (ABANDONED_RUN,), + ).rowcount + conn.commit() + return closed + + +@app.periodic(cron="*/10 * * * *", periodic_id="watchdog") +@app.task(name="politrack.watchdog", queue="social", queueing_lock="watchdog", pass_context=False) +async def watchdog(timestamp: int | None = None) -> dict: + stats = {"stalled_failed": 0, "overdue_deferred": 0} + for job in await app.job_manager.get_stalled_jobs(seconds_since_heartbeat=STALLED_AFTER): + await app.job_manager.finish_job(job, status=Status.FAILED, delete_job=False) + stats["stalled_failed"] += 1 + log.warning("failed stalled job %s (%s): its worker is gone", job.id, job.task_name) + stats["abandoned_runs_closed"] = await asyncio.to_thread(close_abandoned_runs) + for name in await asyncio.to_thread(overdue_tasks, datetime.now(timezone.utc)): + try: + await TASKS[name].defer_async() + stats["overdue_deferred"] += 1 + log.warning("deferred %s: its last scheduled slot passed without a run", name) + except AlreadyEnqueued: + pass + return stats diff --git a/backend/pyproject.toml b/backend/pyproject.toml index 52e70d7..bd7d1e5 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -17,6 +17,7 @@ dependencies = [ "python-slugify>=8.0", "rapidfuzz>=3.9", "procrastinate>=2.14,<4", + "croniter>=2.0", "typer>=0.12", "pyyaml>=6.0", "mcp>=1.9,<2", diff --git a/backend/tests/Dockerfile.debian b/backend/tests/Dockerfile.debian index 2db4f8b..ce0fcde 100644 --- a/backend/tests/Dockerfile.debian +++ b/backend/tests/Dockerfile.debian @@ -13,7 +13,7 @@ RUN apt-get update -qq && apt-get install -y -qq --no-install-recommends eatmyda && eatmydata apt-get install -y -qq --no-install-recommends \ python3 python3-psycopg python3-psycopg-pool python3-pydantic-settings python3-fastapi python3-httpx \ python3-protego python3-feedparser python3-bs4 python3-slugify python3-rapidfuzz python3-typer \ - python3-yaml python3-uvicorn python3-pytest python3-openpyxl ca-certificates >/dev/null \ + python3-yaml python3-uvicorn python3-pytest python3-openpyxl python3-croniter ca-certificates >/dev/null \ && rm -rf /var/lib/apt/lists/* # Keeps the agent image's user, entrypoint and /workspace as they are.