alexei.dolgolyov/notify-bridge
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 31 Wiki Activity

fix(scheduler): honor app timezone for cron triggers and log scheduled events

Browse Source 
CronTrigger.from_crontab was constructed without a timezone, so a cron like
'0 9 * * *' fired at 09:00 host-local instead of 09:00 in the admin-configured
timezone. Now all tracker/action cron triggers are built with the app tz, and
the setting endpoint rebuilds existing cron jobs when the tz changes (since
CronTrigger freezes its tz at construction time).

The scheduler provider also renders current_date/time/datetime/weekday in the
configured tz and exposes a new 'timezone' template variable.

EventLog entries for scheduled_message now include schedule_type,
cron_expression/interval_seconds, timezone, and fire_count, and the dashboard
shows the event type with a label/icon/color.
This commit is contained in:
Admin
2026-04-23 13:35:49 +03:00
parent 5604c733d1
commit 1024085cdd
10 changed files with 209 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/notify_bridge_server/services/scheduler.py
+144 -10
View File
@@ -3,11 +3,39 @@
from __future__ import annotations
import logging
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
from apscheduler.schedulers.asyncio import AsyncIOScheduler
_LOGGER = logging.getLogger(__name__)
def _resolve_zoneinfo(tz_name: str | None) -> ZoneInfo:
"""Resolve an IANA tz string to a ZoneInfo, falling back to UTC on any error.
Kept local to avoid importing from api/dispatch layers inside the scheduler
module (which is loaded at startup, before the API routers).
"""
if not tz_name:
return ZoneInfo("UTC")
try:
return ZoneInfo(tz_name)
except (ZoneInfoNotFoundError, ValueError):
_LOGGER.warning("Unknown timezone %r; falling back to UTC", tz_name)
return ZoneInfo("UTC")
async def _load_app_timezone() -> ZoneInfo:
"""Load the admin-configured app timezone from AppSetting (falls back to UTC)."""
from sqlmodel.ext.asyncio.session import AsyncSession
from ..api.app_settings import get_setting
from ..database.engine import get_engine
async with AsyncSession(get_engine()) as session:
tz_name = await get_setting(session, "timezone")
return _resolve_zoneinfo(tz_name)
_scheduler: AsyncIOScheduler | None = None
# ---------------------------------------------------------------------------
@@ -293,6 +321,8 @@ async def _load_tracker_jobs() -> None:
)
provider_types = {p.id: p.type for p in provider_result.all()}
tz = await _load_app_timezone()
for tracker in trackers:
job_id = f"tracker_{tracker.id}"
if scheduler.get_job(job_id):
@@ -306,7 +336,7 @@ async def _load_tracker_jobs() -> None:
cron_expr = filters.get("cron_expression", "")
if cron_expr:
try:
_add_cron_job(scheduler, job_id, tracker.id, cron_expr, tracker.name)
_add_cron_job(scheduler, job_id, tracker.id, cron_expr, tracker.name, tz)
continue
except Exception as e:
_LOGGER.error(
@@ -337,10 +367,18 @@ def _add_cron_job(
tracker_id: int,
cron_expression: str,
tracker_name: str,
tz: ZoneInfo,
) -> None:
"""Add a cron-triggered job for a scheduler-type tracker."""
"""Add a cron-triggered job for a scheduler-type tracker.
``tz`` is the user-configured app timezone; without it APScheduler
interprets the crontab in the host's local timezone, which surfaces as
events firing at the "wrong" wall-clock time for operators in a non-UTC
zone (see the companion fix in ``update_settings`` which reschedules on
timezone changes).
"""
from apscheduler.triggers.cron import CronTrigger
trigger = CronTrigger.from_crontab(cron_expression)
trigger = CronTrigger.from_crontab(cron_expression, timezone=tz)
scheduler.add_job(
_poll_tracker,
trigger,
@@ -349,7 +387,10 @@ def _add_cron_job(
replace_existing=True,
max_instances=1,
)
_LOGGER.info("Scheduled tracker %d (%s) with cron: %s", tracker_id, tracker_name, cron_expression)
_LOGGER.info(
"Scheduled tracker %d (%s) with cron: %s [tz=%s]",
tracker_id, tracker_name, cron_expression, tz.key,
)
async def schedule_tracker(
@@ -371,7 +412,8 @@ async def schedule_tracker(
if cron_expression:
try:
_add_cron_job(scheduler, job_id, tracker_id, cron_expression, f"tracker-{tracker_id}")
tz = await _load_app_timezone()
_add_cron_job(scheduler, job_id, tracker_id, cron_expression, f"tracker-{tracker_id}", tz)
return
except Exception as e:
_LOGGER.error("Invalid cron for tracker %d: %s — using interval", tracker_id, e)
@@ -506,6 +548,8 @@ async def _load_action_jobs() -> None:
)
actions = result.all()
tz = await _load_app_timezone()
for action in actions:
job_id = f"action_{action.id}"
if scheduler.get_job(job_id):
@@ -514,7 +558,7 @@ async def _load_action_jobs() -> None:
if action.schedule_type == "cron" and action.schedule_cron:
try:
from apscheduler.triggers.cron import CronTrigger
trigger = CronTrigger.from_crontab(action.schedule_cron)
trigger = CronTrigger.from_crontab(action.schedule_cron, timezone=tz)
scheduler.add_job(
_run_action,
trigger,
@@ -523,8 +567,8 @@ async def _load_action_jobs() -> None:
replace_existing=True,
)
_LOGGER.info(
"Scheduled action %d (%s) with cron: %s",
action.id, action.name, action.schedule_cron,
"Scheduled action %d (%s) with cron: %s [tz=%s]",
action.id, action.name, action.schedule_cron, tz.key,
)
continue
except Exception as e:
@@ -563,7 +607,8 @@ async def schedule_action(
if schedule_type == "cron" and cron_expression:
try:
from apscheduler.triggers.cron import CronTrigger
trigger = CronTrigger.from_crontab(cron_expression)
tz = await _load_app_timezone()
trigger = CronTrigger.from_crontab(cron_expression, timezone=tz)
scheduler.add_job(
_run_action,
trigger,
@@ -571,7 +616,10 @@ async def schedule_action(
args=[action_id],
replace_existing=True,
)
_LOGGER.info("Scheduled action %d with cron: %s", action_id, cron_expression)
_LOGGER.info(
"Scheduled action %d with cron: %s [tz=%s]",
action_id, cron_expression, tz.key,
)
return
except Exception as e:
_LOGGER.error("Invalid cron for action %d: %s — using interval", action_id, e)
@@ -596,6 +644,92 @@ async def unschedule_action(action_id: int) -> None:
_LOGGER.info("Unscheduled action %d", action_id)
async def reschedule_cron_jobs_for_timezone_change() -> None:
"""Re-add every cron-triggered tracker/action job under the new app timezone.
Called by the admin settings endpoint after the ``timezone`` AppSetting is
updated. APScheduler's ``CronTrigger`` freezes its timezone at construction
time, so a timezone change has no effect on jobs already in the scheduler
— we have to rebuild those jobs. Interval-triggered jobs are tz-agnostic
and are left alone.
"""
from sqlmodel import select
from sqlmodel.ext.asyncio.session import AsyncSession
from ..database.engine import get_engine
from ..database.models import Action, NotificationTracker, ServiceProvider as ServiceProviderModel
engine = get_engine()
scheduler = get_scheduler()
tz = await _load_app_timezone()
rescheduled = 0
async with AsyncSession(engine) as session:
# Trackers with cron scheduling (scheduler provider + schedule_type=cron).
trackers = (await session.exec(
select(NotificationTracker).where(NotificationTracker.enabled == True) # noqa: E712
)).all()
provider_ids = list({t.provider_id for t in trackers})
provider_types: dict[int, str] = {}
if provider_ids:
rows = await session.exec(
select(ServiceProviderModel).where(ServiceProviderModel.id.in_(provider_ids))
)
provider_types = {p.id: p.type for p in rows.all()}
for tracker in trackers:
if provider_types.get(tracker.provider_id) != "scheduler":
continue
filters = tracker.filters or {}
if filters.get("schedule_type") != "cron":
continue
cron_expr = filters.get("cron_expression", "")
if not cron_expr:
continue
job_id = f"tracker_{tracker.id}"
if scheduler.get_job(job_id):
scheduler.remove_job(job_id)
try:
_add_cron_job(scheduler, job_id, tracker.id, cron_expr, tracker.name, tz)
rescheduled += 1
except Exception as e: # noqa: BLE001
_LOGGER.error(
"Failed to re-apply cron for tracker %d on tz change: %s",
tracker.id, e,
)
# Actions with cron schedules.
actions = (await session.exec(
select(Action).where(Action.enabled == True) # noqa: E712
)).all()
from apscheduler.triggers.cron import CronTrigger
for action in actions:
if action.schedule_type != "cron" or not action.schedule_cron:
continue
job_id = f"action_{action.id}"
if scheduler.get_job(job_id):
scheduler.remove_job(job_id)
try:
scheduler.add_job(
_run_action,
CronTrigger.from_crontab(action.schedule_cron, timezone=tz),
id=job_id,
args=[action.id],
replace_existing=True,
)
rescheduled += 1
except Exception as e: # noqa: BLE001
_LOGGER.error(
"Failed to re-apply cron for action %d on tz change: %s",
action.id, e,
)
_LOGGER.info(
"Rescheduled %d cron job(s) for new app timezone %s", rescheduled, tz.key,
)
async def _run_action(action_id: int) -> None:
"""Run an action (called by APScheduler)."""
from .action_runner import run_action