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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LVjrnpQjRgHdvq95rASoiR

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,
+        )

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

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}

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

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