From f2e4ee4d17c02d1e987453ed10f80fac4f7166eb Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Thu, 24 Sep 2026 16:36:58 -0400 Subject: [PATCH] fix: beat takes each job's last run from task_run, so a redeploy no longer resets the schedule (4408) Beat's default scheduler kept its memory in a shelve file nothing persists, and a scheduler that remembers nothing waits a full interval before any job. Since the one-container image, every redeploy restarts beat, and no daily or weekly job had run since 2026-09-21 (cleanup, backup and download-event pruning, membership sync, thumbnail backfill, integrity check, vacuum). TaskRunScheduler seeds each entry's last_run_at at startup from task_run's newest start for that task: an overdue job runs at once, one not yet due waits the remainder, and one never recorded is due now. prune_task_runs now keeps each task's newest row however old, or a weekly job would look never-run a day after it ran and fire on every restart. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01LVjrnpQjRgHdvq95rASoiR --- backend/app/beat_scheduler.py | 77 ++++++++++++++++++++++++++++++++ backend/app/celery_app.py | 3 ++ backend/app/tasks/maintenance.py | 7 +++ tests/test_beat_scheduler.py | 63 ++++++++++++++++++++++++++ tests/test_maintenance.py | 37 +++++++++++++++ 5 files changed, 187 insertions(+) create mode 100644 backend/app/beat_scheduler.py create mode 100644 tests/test_beat_scheduler.py diff --git a/backend/app/beat_scheduler.py b/backend/app/beat_scheduler.py new file mode 100644 index 0000000..5193163 --- /dev/null +++ b/backend/app/beat_scheduler.py @@ -0,0 +1,77 @@ +"""Celery beat that remembers when each job last ran — from task_run, not a file. + +Celery's default PersistentScheduler keeps its memory in a shelve file in the +working directory. Nothing mounts that directory, so every container recreate +forgets it, and a scheduler that remembers nothing seeds every entry with +`last_run_at = now`: each job waits a FULL interval after startup. A daily job +therefore needs 24 hours without a redeploy to fire. Since the one-container +image (172e33d) every redeploy restarts beat, and on 2026-09-24 no daily or +weekly job had run since the 21st (#4408). + +task_run already records every task that starts (celery_signals), indexed on +(task_name, started_at DESC), and prune_task_runs keeps the newest row of each +task however old it is. So on startup each entry takes its last_run_at from +there: + - a job that is overdue runs at once; + - a job that is not due waits only the remainder of its interval; + - a job that has never run is due now. + +Beat keeps last_run_at in memory from then on, as the default scheduler does; +only the startup seed changes. If the database cannot be read, the entries keep +Celery's own default rather than beat failing to start. +""" + +from __future__ import annotations + +import logging +from datetime import UTC, datetime + +from celery.beat import Scheduler +from sqlalchemy import func, select + +log = logging.getLogger(__name__) + +# Seed for a job with no recorded run: far enough back that any interval or +# crontab reads as due. +NEVER = datetime(2000, 1, 1, tzinfo=UTC) + + +def last_runs(session, task_names: list[str]) -> dict[str, datetime]: + """The latest recorded start of each task, by task name.""" + from .models import TaskRun + + if not task_names: + return {} + rows = session.execute( + select(TaskRun.task_name, func.max(TaskRun.started_at)) + .where(TaskRun.task_name.in_(sorted(set(task_names)))) + .group_by(TaskRun.task_name) + ).all() + return dict(rows) + + +def seed(entries, last: dict[str, datetime]) -> None: + """Set each entry's last_run_at from `last`; a task with none is due now.""" + for entry in entries: + entry.last_run_at = last.get(entry.task, NEVER) + + +class TaskRunScheduler(Scheduler): + """An in-memory beat seeded from task_run history at startup.""" + + def setup_schedule(self): + super().setup_schedule() + try: + from .tasks._sync_engine import sync_session_factory + + with sync_session_factory()() as session: + last = last_runs(session, [e.task for e in self.schedule.values()]) + except Exception: + log.exception("beat: could not read task_run; every job waits a full interval") + return + seed(self.schedule.values(), last) + due = sum(1 for e in self.schedule.values() if e.is_due()[0]) + log.info( + "beat: seeded %d job(s) from task_run, %d due now", + len(self.schedule), due, + ) diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index d341530..5f45567 100644 --- a/backend/app/celery_app.py +++ b/backend/app/celery_app.py @@ -353,6 +353,9 @@ def make_celery() -> Celery: }, }, timezone="UTC", + # Beat's memory of when each job last ran comes from task_run, not a + # shelve file nothing persists — see beat_scheduler (#4408). + beat_scheduler="backend.app.beat_scheduler:TaskRunScheduler", ) # FC-3i: register task_run signal handlers (side-effect import). from . import celery_signals # noqa: F401 diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index 8e41efd..8b653de 100644 --- a/backend/app/tasks/maintenance.py +++ b/backend/app/tasks/maintenance.py @@ -473,6 +473,10 @@ def prune_task_runs() -> dict: (recover_stalled_task_runs) is the mechanism that flips them to terminal state; prune doesn't touch in-flight state. - 'retry' rows: treated as failures (>7d). + - The NEWEST row of each task is never deleted, whatever its age: it is + what the beat scheduler reads to know when a job last ran (#4408). + Without it a weekly job's last run would be pruned after a day, and beat + would think it had never run and fire it on every restart. Returns dict of how many rows were deleted in each bucket. """ @@ -480,16 +484,19 @@ def prune_task_runs() -> dict: now = datetime.now(UTC) ok_cutoff = now - timedelta(seconds=TASK_RUN_KEEP_OK_SECONDS) fail_cutoff = now - timedelta(seconds=TASK_RUN_KEEP_FAILURE_SECONDS) + newest = select(func.max(TaskRun.id)).group_by(TaskRun.task_name) with SessionLocal() as session: ok_deleted = session.execute( delete(TaskRun) .where(TaskRun.status == "ok") .where(TaskRun.finished_at < ok_cutoff) + .where(TaskRun.id.not_in(newest)) ).rowcount or 0 fail_deleted = session.execute( delete(TaskRun) .where(TaskRun.status.in_(["error", "timeout", "retry"])) .where(TaskRun.finished_at < fail_cutoff) + .where(TaskRun.id.not_in(newest)) ).rowcount or 0 session.commit() return {"ok_deleted": ok_deleted, "failures_deleted": fail_deleted} diff --git a/tests/test_beat_scheduler.py b/tests/test_beat_scheduler.py new file mode 100644 index 0000000..5224585 --- /dev/null +++ b/tests/test_beat_scheduler.py @@ -0,0 +1,63 @@ +"""#4408: beat takes each job's last run from task_run, not a shelve file. + +The default scheduler forgot everything on each container recreate and waited +a full interval before any job, so with several redeploys a day no daily or +weekly job had run since 2026-09-21. +""" +from datetime import UTC, datetime, timedelta + +import pytest +from celery.beat import ScheduleEntry +from celery.schedules import schedule + +from backend.app.beat_scheduler import NEVER, last_runs, seed +from backend.app.celery_app import celery +from backend.app.models import TaskRun + +DAY = 86400.0 + + +def _entry(task, every=DAY): + return ScheduleEntry(name=task, task=task, schedule=schedule(every, app=celery), app=celery) + + +def test_a_job_that_ran_recently_waits_only_the_remainder(): + entry = _entry("t.daily") + seed([entry], {"t.daily": datetime.now(UTC) - timedelta(hours=20)}) + due, next_in = entry.is_due() + assert not due + assert 3 * 3600 < next_in <= 4 * 3600 + 5 + + +def test_an_overdue_job_is_due_at_once(): + """The live case: daily jobs last ran three days before the restart.""" + entry = _entry("t.daily") + seed([entry], {"t.daily": datetime.now(UTC) - timedelta(days=3)}) + assert entry.is_due()[0] + + +def test_a_job_with_no_recorded_run_is_due_now(): + entry = _entry("t.never") + seed([entry], {}) + assert entry.last_run_at == NEVER + assert entry.is_due()[0] + + +def test_the_scheduler_is_the_one_celery_uses(): + assert celery.conf.beat_scheduler == "backend.app.beat_scheduler:TaskRunScheduler" + + +@pytest.mark.integration +def test_last_runs_reads_the_newest_start_per_task(db_sync): + now = datetime.now(UTC) + for name, ago in [("t.a", 30), ("t.a", 2), ("t.b", 5)]: + db_sync.add(TaskRun( + celery_task_id="x", queue="maintenance", task_name=name, + started_at=now - timedelta(hours=ago), status="ok", + )) + db_sync.flush() + + got = last_runs(db_sync, ["t.a", "t.b", "t.c"]) + + assert set(got) == {"t.a", "t.b"} + assert abs((got["t.a"] - (now - timedelta(hours=2))).total_seconds()) < 1 diff --git a/tests/test_maintenance.py b/tests/test_maintenance.py index 5f7e3ed..74030af 100644 --- a/tests/test_maintenance.py +++ b/tests/test_maintenance.py @@ -569,6 +569,12 @@ def test_prune_task_runs_deletes_failures_older_than_7d(db_sync): started_at=now - timedelta(days=10), finished_at=now - timedelta(days=9), ) + # A later run of the same task, so the old failure is not its newest row + # (the newest is kept for beat — see the test below). + _make_task_run( + db_sync, status="ok", + started_at=now - timedelta(hours=2), finished_at=now - timedelta(hours=1), + ) db_sync.commit() result = prune_task_runs.apply().get() @@ -581,6 +587,37 @@ def test_prune_task_runs_deletes_failures_older_than_7d(db_sync): assert surviving is None +def test_prune_task_runs_keeps_each_tasks_newest_row_however_old(db_sync): + """#4408: beat reads a job's last run from task_run. A weekly job's only + row is older than the 24h ok-retention, and pruning it would make beat + think the job never ran and fire it on every restart.""" + from sqlalchemy import select + + from backend.app.models import TaskRun + from backend.app.tasks.maintenance import prune_task_runs + + now = datetime.now(UTC) + weekly = "backend.app.tasks.fake.weekly" + older = _make_task_run( + db_sync, status="ok", task_name=weekly, + started_at=now - timedelta(days=14), finished_at=now - timedelta(days=14), + ) + newest = _make_task_run( + db_sync, status="ok", task_name=weekly, + started_at=now - timedelta(days=7), finished_at=now - timedelta(days=7), + ) + db_sync.commit() + + prune_task_runs.apply().get() + + db_sync.expire_all() + surviving = set(db_sync.execute( + select(TaskRun.id).where(TaskRun.task_name == weekly) + ).scalars().all()) + assert surviving == {newest} + assert older not in surviving + + def test_prune_task_runs_keeps_recent_failures(db_sync): from sqlalchemy import select