bvandeusenandClaude Opus 5.5
f2e4ee4d17
CI and images / lint (push) Successful in 3s
CI and images / extension-version (push) Successful in 3s
CI and images / frontend-build (push) Successful in 20s
CI and images / backend-lint-and-test (push) Successful in 33s
CI and images / integration (push) Successful in 2m18s
CI and images / sign-extension (push) Successful in 4s
CI and images / build-agent (push) Successful in 5s
CI and images / build-web (push) Successful in 1m41s
CI and images / smoke-web (push) Successful in 53s
CI and images / promote (push) Skipped
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
64 lines
2.0 KiB
Python
64 lines
2.0 KiB
Python
"""#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
|