From b7b7f2f64854349b7943282266d98fdd63f9916a Mon Sep 17 00:00:00 2001 From: Simon Diesenreiter Date: Thu, 12 Nov 2026 23:42:19 +0100 Subject: [PATCH] Add a watchdog that repairs missed schedules and stalled jobs Every daily and weekly source missed two days in November 2026: Procrastinate skips a periodic slot that is more than ten minutes overdue, and jobs of killed workers stayed "doing" forever. Every ten minutes the watchdog fails jobs of dead workers, closes abandoned run records and defers any scheduled task whose last slot passed without a run. Also: EP vote references of joint motions and Council documents, and the loader now refreshes a vote's reference and stage. Co-Authored-By: Claude Opus 5.5 --- backend/politrack/core/loader.py | 1 + backend/politrack/sources/eu.py | 4 +- backend/politrack/worker.py | 72 ++++++++++++++++++++++++++++++++ backend/pyproject.toml | 1 + backend/tests/Dockerfile.debian | 2 +- 5 files changed, 77 insertions(+), 3 deletions(-) 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.