Paul Trowbridge
a66488d1f2
Scheduler now evaluates cron expressions against local time instead of UTC, so schedules fire at the user's local clock time. All timestamp displays in templates use a new `localtime` Jinja filter that converts UTC strings from SQLite to the server's local timezone. Updated CLAUDE.md to reflect the systemd service setup. Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
96 lines
3.0 KiB
Python
96 lines
3.0 KiB
Python
"""Background scheduler: fires group runs when cron expressions are due.
|
|
|
|
One daemon thread wakes every 60 s, checks all enabled schedules, and
|
|
spawns a per-run thread for any that are due. ``last_fired_at`` is written
|
|
to the DB before the run starts so a slow or crashing run cannot double-fire
|
|
the same occurrence. Survives server restarts: missed ticks while pipekit was
|
|
down are detected on the next startup check (5 s after start).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import threading
|
|
import time
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
def _now() -> datetime:
|
|
return datetime.now()
|
|
|
|
|
|
def _check_and_fire() -> None:
|
|
from . import engine, repo
|
|
try:
|
|
schedules = repo.list_all_enabled_schedules()
|
|
except Exception as e: # noqa: BLE001
|
|
log.error("scheduler: failed to load schedules: %s", e)
|
|
return
|
|
|
|
now = _now()
|
|
for sched in schedules:
|
|
try:
|
|
_maybe_fire(sched, now)
|
|
except Exception as e: # noqa: BLE001
|
|
log.error("scheduler: error on schedule id=%s: %s", sched["id"], e)
|
|
|
|
|
|
def _maybe_fire(sched: dict, now: datetime) -> None:
|
|
from croniter import croniter, CroniterBadCronError
|
|
from . import engine, repo
|
|
|
|
last_str = sched["last_fired_at"]
|
|
if last_str:
|
|
# SQLite stores UTC; convert to local for cron evaluation.
|
|
last_dt = datetime.fromisoformat(last_str).replace(
|
|
tzinfo=timezone.utc).astimezone().replace(tzinfo=None)
|
|
else:
|
|
last_dt = now - timedelta(seconds=120)
|
|
|
|
try:
|
|
cron = croniter(sched["cron_expr"], last_dt)
|
|
next_dt = cron.get_next(datetime)
|
|
except CroniterBadCronError:
|
|
log.warning("scheduler: invalid cron_expr %r for schedule id=%s — skipping",
|
|
sched["cron_expr"], sched["id"])
|
|
return
|
|
|
|
if next_dt > now:
|
|
return
|
|
|
|
log.info("scheduler: firing group_id=%s (schedule id=%s expr=%r)",
|
|
sched["group_id"], sched["id"], sched["cron_expr"])
|
|
|
|
# Mark fired before spawning so a slow run can't double-fire.
|
|
repo.mark_schedule_fired(sched["id"])
|
|
|
|
group_id = sched["group_id"]
|
|
sched_id = sched["id"]
|
|
|
|
def _run() -> None:
|
|
try:
|
|
group_run_id = repo.create_group_run(
|
|
group_id, triggered_by=f"schedule:{sched_id}"
|
|
)
|
|
engine.run_group(group_id, group_run_id=group_run_id)
|
|
except Exception as exc: # noqa: BLE001
|
|
log.error("scheduler: run failed for group_id=%s: %s", group_id, exc)
|
|
|
|
threading.Thread(target=_run, daemon=True).start()
|
|
|
|
|
|
def start_scheduler() -> None:
|
|
"""Start the background scheduler daemon thread. Safe to call once at startup."""
|
|
|
|
def _loop() -> None:
|
|
time.sleep(5) # let the app finish initialising
|
|
while True:
|
|
_check_and_fire()
|
|
time.sleep(60)
|
|
|
|
t = threading.Thread(target=_loop, daemon=True, name="pipekit-scheduler")
|
|
t.start()
|
|
log.info("scheduler: started")
|